Skip to main content

​
Introduction

Plug-ins inspect and modify the input or output of an agent run. A plug-in is a tool that is automatically started either before the agent begins generating a response (pre-invoke) or after the agent completes its response (post-invoke). Use plug-ins to adjust the message history before the agent processes it or refine the final output before it is returned to the user. Common use cases include applying guardrails, masking sensitive data, or transforming output such as translation.

​
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: default
name: email_agent
llm: watsonx/meta-llama/llama-3-2-90b-vision-instruct
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. 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