Plug-ins

Plug-ins for agents

Plug-ins play a role in enhancing the capabilities and robustness of agents. They help enable custom behavior to be easily added to an agent’s processing flow, allowing modifications to incoming input or outgoing output. This customization is essential for applications where agents must comply with safety, security, and regulatory requirements. Plug-ins protect the agent from problematic inputs by filtering or sanitizing content and enforce compliance by applying guardrails to sensitive or restricted information. They also improve the reliability and trustworthiness of outputs by masking sensitive data or transforming results to meet specific standards.

Input and output plug-ins overview

The agent workflow includes two types of plug-ins:

Input plug-ins - operate before the agent processes a request, inspecting and potentially modifying incoming messages.
Output plug-ins - run after the agent generates a response, refining or changing the final output before it returns to the user.

Both types of plug-ins receive the message text and the agent run context, which provides additional information that can influence their behavior. Plug-ins offer multiple options for interacting with messages. They can check or validate content, rewrite, or mask parts of messages, and even stop further processing if certain criteria are met. This flexibility makes plug-ins powerful tools for enforcing business rules, safety protocols, or compliance policies dynamically during agent execution. Plug-ins are implemented as Python tools with a type that determines their invocation phase. Specify the type with the tool decorator: kind=PythonToolKind.AGENTPREINVOKE for pre-invoke plug-ins or kind=PythonToolKind.AGENTPOSTINVOKE for post-invoke plug-ins. For example, a pre-invoke plug-in is declared inline like this:

PYTHON

@tool(description="input plugin tool", kind=PythonToolKind.AGENTPREINVOKE)
def my_preinvoke_plugin(...):
    ...

Developers can register or update these Python tools in the watsonx Orchestrate environment, consistent with the overall tool lifecycle in the platform. This setup helps developers extend and modify agent behavior by deploying new or updated plug-ins as needed. Plug-ins access the incoming message context through AgentPreInvokePayload and decide whether to accept or reject processing by setting continue_processing in AgentPreInvokeResult. This approach provides precise control over agent execution. Pre-invoke plug-ins address two primary situations. First, they control agent access by integrating with identity providers for authentication and authorization. Second, they process each message to accept it as-is, reject it entirely (for example, blocking unsafe content), or reformulate the input for better handling. This dual capability helps ensure robust security and input quality before the agent generates responses.

Implementation details

A plug-in receives two key pieces of information and must return a result when it finishes processing them:

Context – Contains details about the current state and user environment.
Payload – Includes the data that is required for the plug-in to perform its task.

Schemas such as PluginContext, and AgentPostInvokePayload, can be imported from the ibm_watsonx_orchestrate library. The content, displayed as JSON for illustrative purposes, typically contains details about the current state and user environment, for example:

"state": {
  "context": {
    "wxo_email_id": "xxxx@yyyyy.com",
    "wxo_tenant_id": "AAAAA-BBBBB-CCCCC-...",
    "wxo_user_name": "AAAA BBBB"
  }
},
"global_context": {
  "request_id": "1764763622520999908_404676ed-bdbe-66b5-7915-503a793bbd16",
  "user": "xxxx@yyyyy.com",
  "tenant_id": "AAAAA-BBBBB-CCCCC-...",
  "server_id": null,
  "state": {},
  "metadata": {}
},
"metadata": {
  "action": "ALL"
}

Action value	Description
`AgentPreInvokeType.ALL`	Used when the plugin is invoked for the root agent on every utterance. In this case, the pre-invoke plugin is executed with the full set of implemented logic.
`AgentPreInvokeType.RBAC_ONLY`	Used when validating access to collaborator agents during a parent agent pre-invoke plugin invocation. This ensures the parent agent includes only those collaborator agents that the user is authorized to access.
`AgentPreInvokeType.SKIP_RBAC`	Used when access has already been validated for the same thread and the result is available in the cache. In this case, access checks are skipped, and processing always continues (`continue_processing = True`) with the same payload.

Collaborator access control

To check access to a collaborator agent, bind a pre-invoke plug-in to the collaborator agent.
The plug-in is invoked during the parent agent pre-invoke execution. When checking collaborator access using a pre-invoke plug-in (AgentPreInvokeType.RBAC_ONLY), if access to a specific collaborator agent is denied, that agent is removed from the parent agent’s list of collaborators. Code example:

PYTHON

result = AgentPreInvokeResult(modified_payload=agent_pre_invoke_payload)
metadata = plugin_context.metadata
action = metadata["action"]
result.continue_processing = False
result.modified_payload = 'Access Denied'
if action == AgentPreInvokeType.RBAC_ONLY.value:
    is_allowed = check_access()
    result.continue_processing = is_allowed
    result.modified_payload = 'Access Verified' if is_allowed else 'Access Denied'
return result

Within the plug-in, use the appropriate conditions to ensure access checks are performed only when necessary, avoiding repeated calls to the access-check API for the same thread.

The payload usually contains the agent ID and the list of messages to process. For agent_pre_invoke_payload, the messages list contains only one message, such as:

"agent_id": "f07c7e31-f6ad-40a9-ad2e-1257982e282e",
"messages": [
  {
    "role": "user",
    "content": {
      "type": "text",
      "text": "This is the text to process….."
    }
  }
]

A common plug-in pattern is to copy the input messages to the output and optionally modify them. The plug-in controls whether the agent continues processing by returning a flag such as continue_processing = True or False. This mechanism helps enable the plug-in to accept, reject, or modify the agent’s workflow dynamically.

Email agent

Use this configuration to include an optional tool that the agent can call, such as a tool that simulates sending an email. Other plug-ins in this example run automatically, regardless of the agent’s actions. Code example:

YAML

spec_version: v1
kind: native
style: react_core
name: email_agent
llm: groq/openai/gpt-oss-120b
description: Send Email to given email_id and body.

tools: [send_email_tool]

plugins:
  agent_pre_invoke:
      - plugin_name: guardrail_plugin
  agent_post_invoke:
      - plugin_name: email_masking_plugin

Email masking

The post-invoke plug-in masks email addresses after the agent completes its response. The plug-in runs automatically and modifies the final output before it is sent back to the user. The next utterance will use the updated context or conversation produced by the post-invoke plug-in processing of the previous utterance. Code example:

PYTHON

import re
from enum import Enum
from typing import Generic, Optional, Any, List, Dict, TypeAlias, TypeVar, Union, Literal
from pydantic import BaseModel, Field, PrivateAttr, RootModel
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.tools.types import PythonToolKind, PluginContext, AgentPostInvokePayload, \
    AgentPostInvokeResult, TextContent, Message


@tool(description="plugin tool", kind=PythonToolKind.AGENTPOSTINVOKE)
def email_masking_plugin(plugin_context: PluginContext, agent_post_invoke_payload: AgentPostInvokePayload) -> AgentPostInvokeResult:
    result = AgentPostInvokeResult()

    def mask_emails_in_text(text: str, mask_char: str = '*') -> str:
        """
        Finds and masks all email addresses in a given text.
        """
        def mask_email(match):
            email = match.group(0)
            local, domain = email.split('@', 1)
            visible = min(2, len(local))
            masked_local = local[:visible] + mask_char * (len(local) - visible)
            return f"{masked_local}@{domain}"

        email_pattern = r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}'
        return re.sub(email_pattern, mask_email, text)

    if agent_post_invoke_payload is None or agent_post_invoke_payload.messages is None or len(agent_post_invoke_payload.messages) == 0:
        result.continue_processing = False
        return result

    first_msg = agent_post_invoke_payload.messages[0]
    content = getattr(first_msg, "content", None)

    if content is None or not hasattr(content, "text") or content.text is None:
        result.continue_processing = False
        return result

    masked_text = mask_emails_in_text(content.text)
    new_content = TextContent(type="text", text=masked_text)
    new_message = Message(role=first_msg.role, content=new_content)

    modified_payload = agent_post_invoke_payload.copy(deep=True)
    modified_payload.messages[0] = new_message

    result.continue_processing = True
    result.modified_payload = modified_payload

    return result

Guardrail plug-in

The pre-invoke plug-in applies guardrails before the agent processes the request. It can strip or adjust content to enforce compliance or safety requirements. Code example:

PYTHON

import os
import re
from enum import Enum
from typing import Generic, Optional, Any, List, Dict, TypeAlias, TypeVar, Union, Literal
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.tools.types import PythonToolKind, PluginContext, AgentPreInvokePayload, \
    AgentPreInvokeResult
from pydantic import BaseModel, Field, PrivateAttr, RootModel


@tool( description="plugin tool", kind=PythonToolKind.AGENTPREINVOKE)
def guardrail_plugin(plugin_context: PluginContext, agent_pre_invoke_payload: AgentPreInvokePayload) -> AgentPreInvokeResult:

    user_input = ''
    modified_payload = agent_pre_invoke_payload
    res = AgentPreInvokeResult()
    if agent_pre_invoke_payload and agent_pre_invoke_payload.messages:
        user_input = agent_pre_invoke_payload.messages[-1].content.text


    def mask_words_in_text(text: str, words_to_mask: list, mask_char: str = '*') -> str:
        """
        Masks all occurrences of selected words in a given text.

        Args:
            text (str): The input text.
            words_to_mask (list): List of words (case-insensitive) to mask.
            mask_char (str): Character to use for masking.

        Returns:
            str: The masked text.
        """
        if not words_to_mask:
            return text

        # Escape special regex characters and join words into a regex pattern
        pattern = r'\b(' + '|'.join(map(re.escape, words_to_mask)) + r')\b'

        def mask_match(match):
            word = match.group(0)
            return mask_char * len(word)

        # Replace words, ignoring case
        return re.sub(pattern, mask_match, text, flags=re.IGNORECASE)

    words = [ 'silly', 'Silly']
    if 'stupid' in user_input:
        modified_text = 'blocked'
        res.continue_processing = False
    else:
        modified_text = mask_words_in_text(text=user_input, words_to_mask=words)
        res.continue_processing = True
    modified_payload.messages[-1].content.text = modified_text
    res.modified_payload = modified_payload
    return res

Import Python tools

Use this command to import Python-based tools for use as plug-ins. The tools can then be configured as pre-invoke or post-invoke plug-ins in the agent workflow. Command example:

BASH

orchestrate tools import -k python -f path/to/plugin.py

Import Python toolkits

You can configure Python toolkit tools as plugins. You define each tool individually in the plugins section of the configuration. To reference a specific tool, use the following format:

<toolkit_name>:<plugin_name>

Example

YAML

plugins:
  agent_pre_invoke:
      - plugin_name: <toolkit_name>:<plugin_name>
  agent_post_invoke:
      - plugin_name: <toolkit_name>:<plugin_name> 

Before you reference a Python toolkit as a plugin, import the toolkit into watsonx Orchestrate. You can complete this task by using the orchestrate toolkits add or orchestrate toolkits import commands. For more information, see Add Python toolkits using the ADK CLI. Command examples

orchestrate toolkits add \
  --kind python \
  --name <toolkit-name> \
  --description <toolkit-description> \
  --tier <deployment-tier> \
  --package_root <toolkit-path>

orchestrate toolkits import -f <path to spec> --app-id <connection>

Tutorial: Implement agent guardrails with watsonx Orchestrate plug-ins

Release Notes

Get Started

Build

Deploy

Analyze

Developer experience

Legal notices

Plug-ins for agents

Input and output plug-ins overview

Implementation details

Collaborator access control

Email agent

Email masking

Guardrail plug-in

Import Python tools

Import Python toolkits

​Plug-ins for agents

​Input and output plug-ins overview

​Implementation details

​Collaborator access control

​Email agent

​Email masking

​Guardrail plug-in

​Import Python tools

​Import Python toolkits

​Related links

Plug-ins for agents

Input and output plug-ins overview

Implementation details

Collaborator access control

Email agent

Email masking

Guardrail plug-in

Import Python tools

Import Python toolkits

Related links