Plugins

Extend agent functionality with single-file plugins

Plugins add tools, hooks, and guardrails that can rewrite — or now block — what an agent sees and does; drop a Python file in ~/.praisonai/plugins/ and load it in one line.

from praisonaiagents import Agent, discover_and_load_plugins

discover_and_load_plugins()

agent = Agent(
    name="Assistant",
    instructions="Help users with weather queries.",
    tools=["get_weather"],
)
agent.start("What's the weather in Paris?")

The user asks a question; plugins add tools and hooks before the agent answers.

Quick Start

1. Simple Usage

   Create ~/.praisonai/plugins/my_tools.py:

   """
   Plugin Name: My Tools
   Description: Custom tools for my agent
   Version: 1.0.0
   """
   
   from praisonaiagents import tool
   
   @tool
   def get_weather(city: str) -> str:
       """Get weather for a city."""
       return f"Sunny in {city}, 22°C"

   Load and run:

   from praisonaiagents import Agent, discover_and_load_plugins
   
   discover_and_load_plugins()
   
   agent = Agent(
       name="Assistant",
       instructions="Help users",
       tools=["get_weather"],
   )
   agent.start("What's the weather in Paris?")

2. With Configuration

   Point the environment (or config file) at plugins and Agent construction wires them for you — no explicit plugins.enable() call needed:

   export PRAISONAI_PLUGINS=true

   from praisonaiagents import Agent
   
   agent = Agent(name="Assistant", instructions="Help users")
   agent.start("Hello!")   # entry-point plugins are wired in before this runs

   Prefer to turn plugins on in code? Call plugins.enable(...) before creating the agent:

   from praisonaiagents import Agent, plugins
   
   plugins.enable(["logging", "metrics"])
   
   agent = Agent(name="Assistant", instructions="Help users")
   agent.start("Hello!")

Selective Enable and Late Registration

After plugins.enable([names]), any plugin registered later stays disabled unless its name is in that list.

from praisonaiagents import Agent, plugins
from praisonaiagents.plugins.manager import get_plugin_manager

# Only "logging" is allowed.
plugins.enable(["logging"])

# Anything registered after this stays disabled unless it is in the allow-list.
get_plugin_manager().register(MyMetricsPlugin())     # registered but disabled
get_plugin_manager().register(MyLoggingPlugin())     # allow-listed → enabled

agent = Agent(name="Assistant", instructions="Help users")
agent.start("Hello!")     # only the logging plugin's tools/hooks are active

The allow-list keeps working even for plugins you add later, so nothing slips in after you declare the selected set.

Entry-point discovery, auto-discovery, and manual PluginManager.register(...) all inherit the same rule, because each path flows through register().

Widen or Reset the Allow-List

How It Works

plugins.enable() auto-calls wire_into_hook_registry(), which registers each enabled plugin’s lifecycle methods on the default hook registry the agent consults at runtime — and Agent init calls this for you when the env var or config file requests it (see Auto-Enable from Env or Config).

The hooks fire in this order around each agent run:

On Agent construction the plugin manager feeds more than the hook registry: a plugin declared PluginType.GUARDRAIL / POLICY / SKILL is routed into the Agent’s typed subsystems — GuardrailChain, PolicyEngine, and SkillManager — via _merge_plugin_subsystems():

Choosing How to Load Plugins

Pick the loading method that fits your setup.

Plugin Locations

Core Compatibility (Load-Time Gate)

A plugin can declare which praisonaiagents versions it supports in its existing dependencies field — PraisonAI checks it at load and skips mismatched plugins before they wire into the hook runtime.

"""
Plugin Name: PII Guardrail
Dependencies: praisonaiagents>=0.9,<2
"""

from praisonaiagents import Plugin, PluginInfo

class PIIGuardrail(Plugin):
    @property
    def info(self) -> PluginInfo:
        return PluginInfo(
            name="pii_guardrail",
            dependencies=["praisonaiagents>=0.9,<2"],
        )

from praisonaiagents import get_plugin_manager

manager = get_plugin_manager()
manager.register(PIIGuardrail())  # returns False if core is out of range

Security: Project-Plugin Trust Gate

A single-file plugin is the most privileged extension surface there is — it can hook every lifecycle event and intercept every tool call. So a plugin dropped into a project’s ./.praisonai/plugins/*.py (for example, one that arrived with a cloned repository) does not run until you opt in. This mirrors PRAISONAI_ALLOW_LOCAL_TOOLS for tools: a cloned repo carrying a malicious .praisonai/plugins/exfil.py stays inert by default.

Open the gate with an environment variable (accepts true/1/yes/on):

export PRAISONAI_ALLOW_PROJECT_PLUGINS=true

Or in .praisonai/config.yaml:

plugins:
  allow_project_plugins: true

Security: Conversation-Content Capability Gate

A plugin that isn’t bundled must be explicitly granted allow_conversation before it can read or rewrite prompts, responses, and outbound messages — its tools still run either way.

A third-party plugin’s tools still register and fire; only the eight conversation-content hooks are withheld until you grant access.

Which hooks are gated

These eight hooks carry — and can rewrite — the user’s prompt, the model’s response, or an outbound reply, so they are default-deny for third-party plugins:

Grant a third-party plugin access

Add allow_conversation: true to the plugin’s own option block in .praisonai/config.yaml:

plugins:
  my_pii_redactor:
    allow_conversation: true      # opt in to before_llm / after_llm / message_*
    # ...any other plugin options are still delivered to on_config as before

Or grant it in code:

from praisonaiagents.plugins.manager import PluginManager

manager = PluginManager()
manager.set_plugin_options({"my_pii_redactor": {"allow_conversation": True}})

What “first-party” means

First-party plugins — those bundled inside the installed praisonaiagents/ package on disk — are granted conversation access automatically. This is a different notion from the location trust levels (project vs user-global vs entry-point) of the Project-Plugin Trust Gate.

A plugin is first-party only when both hold:

1. its class’s module name is praisonaiagents or starts with praisonaiagents., and
2. that module’s real on-disk file resolves to a path inside the installed praisonaiagents/ package directory.

Interaction with the project-plugin trust gate

The two gates are complementary. A project-local single-file plugin authorised to load via PRAISONAI_ALLOW_PROJECT_PLUGINS=true is still not first-party — it must also carry allow_conversation: true in its config block before it sees any conversation-content hook.

Tools and channels are never gated

An ungranted plugin still registers get_tools(), before_tool, after_tool, before_tool_definitions, on_permission_ask, channel adapters, on_init / on_shutdown, session_start / session_end, on_error, on_config, and on_auth. Only the eight conversation-content hooks above require the grant.

Which plugins see prompt content

Inspect the Plugin Registry

get_plugin_registry() returns a truthful, unified view of every discoverable plugin from three real sources — no hardcoded entries. It reads without a running agent, so a CLI or health check can render it directly.

from praisonaiagents.plugins import get_plugin_registry

for p in get_plugin_registry():
    print(p["name"], p["source"], p["enabled"], p["hooks"])

Each entry is a dict: {name, version, description, source, enabled, hooks, requests_conversation, conversation_granted} (single-file entries also include path). The source field records provenance:

Two fields surface the conversation-content capability gate:

Reload Plugins at Runtime

Edited a plugin, or dropped a new one in, and want it live in the current process? Run praisonai plugins reload — it forces rediscovery and rewires hooks so newly-added or edited plugins take effect without a restart. See the Plugins CLI.

Hook Plugin Example

Create ~/.praisonai/plugins/my_logger.py:

"""
Plugin Name: My Logger
Description: Logs tool calls
Version: 1.0.0
Hooks: before_tool, after_tool
"""

from praisonaiagents.hooks import add_hook

@add_hook("before_tool")
def log_before(data):
    print(f"Calling: {data.tool_name}")

@add_hook("after_tool")
def log_after(data):
    print(f"Done: {data.tool_name}")

Create ~/.praisonai/plugins/tool_sandbox.py to filter advertised tools:

"""
Plugin Name: Tool Sandbox
Description: Filters tool definitions sent to the LLM
Version: 1.0.0
Hooks: before_tool_definitions
"""

from praisonaiagents.hooks import add_hook

BLOCKED = {"delete_file", "shell_exec"}

@add_hook("before_tool_definitions")
def sandbox_tools(data):
    data.tool_definitions[:] = [
        t for t in data.tool_definitions
        if t["function"]["name"] not in BLOCKED
    ]

from praisonaiagents import Agent, discover_and_load_plugins

discover_and_load_plugins()
agent = Agent(name="Assistant", instructions="Help users")
agent.start("Search for Python tutorials")

Typed Plugins (Guardrail / Policy / Skill)

Declaring PluginInfo(plugin_type=PluginType.GUARDRAIL) (or POLICY / SKILL) moves a plugin off the generic lifecycle-hook path and into the Agent’s typed subsystem. A guardrail plugin then sees the raw tool call and the raw tool result — the blind spot the after_llm hook could never cover, because after_llm only ever saw the model’s paraphrase.

On Agent.__init__, right after _setup_guardrail() runs and _skills / _policy are set, _merge_plugin_subsystems() collects the typed plugins and folds them in:

Each collector dispatches on the declared plugin.info.plugin_type and iterates enabled plugins only, snapshotting (name, plugin) pairs under the manager’s lock:

Quick Start

1. Guardrail plugin — redact a raw tool result

   A PluginType.GUARDRAIL plugin returns a guardrail object from as_guardrail(). The object joins the Agent’s GuardrailChain and sees validate_tool_result — the raw tool output, before the model paraphrases it.

   from praisonaiagents import Agent, Plugin, PluginInfo, PluginType
   from praisonaiagents.plugins import get_plugin_manager
   
   class PIIGuardrail:
       """Sees the raw tool result, not the model's paraphrase."""
       def validate_tool_result(self, tool_name, result, **kw):
           if isinstance(result, str) and "SECRET" in result:
               return True, result.replace("SECRET", "[REDACTED]")
           return True, result
   
   class PIIGuardrailPlugin(Plugin):
       @property
       def info(self):
           return PluginInfo(
               name="pii_redactor",
               plugin_type=PluginType.GUARDRAIL,   # the dispatch key
           )
       def as_guardrail(self):
           return PIIGuardrail()
   
   get_plugin_manager().register(PIIGuardrailPlugin())
   get_plugin_manager().enable("pii_redactor")
   
   agent = Agent(name="Assistant", instructions="Help users.", llm="gpt-4o-mini")
   # The plugin's PIIGuardrail is now on agent._tool_result_guardrails.
   # Any tool that returns "a SECRET value" becomes "a [REDACTED] value"
   # before the model or the user ever sees it.

   The same effect through after_llm (the pre-PR pattern) could not intercept the raw tool result — only the model’s paraphrase — so a secret leaked whenever the model quoted a tool verbatim or a downstream tool consumed the raw output.

2. Policy plugin — add execution rules

   A PluginType.POLICY plugin returns Policy objects from get_policies(). They are added to the Agent’s PolicyEngine (one is created if the Agent has no policy=).

   from praisonaiagents import Agent, Plugin, PluginInfo, PluginType
   from praisonaiagents.plugins import get_plugin_manager
   from praisonaiagents.policy import Policy, PolicyRule, PolicyAction
   
   class NoDeletePlugin(Plugin):
       @property
       def info(self):
           return PluginInfo(name="no_delete", plugin_type=PluginType.POLICY)
       def get_policies(self):
           return [Policy(
               name="no_delete",
               rules=[PolicyRule(
                   action=PolicyAction.DENY,
                   resource="tool:delete_*",
                   reason="Delete operations blocked by plugin",
               )],
           )]
   
   get_plugin_manager().register(NoDeletePlugin())
   get_plugin_manager().enable("no_delete")
   
   agent = Agent(name="Assistant", instructions="Manage files safely.")
   # agent.policy now denies tool:delete_* before dispatch.
   agent.start("Please delete /etc/passwd")

3. Skill plugin — contribute loadable skills

   A PluginType.SKILL plugin returns filesystem path strings from get_skills(). They are appended to the Agent’s _skills so SkillManager loads them as real SKILL.md skills.

   from praisonaiagents import Agent, Plugin, PluginInfo, PluginType
   from praisonaiagents.plugins import get_plugin_manager
   
   class PdfSkillPlugin(Plugin):
       @property
       def info(self):
           return PluginInfo(name="pdf_skill", plugin_type=PluginType.SKILL)
       def get_skills(self):
           return ["./skills/pdf-processing"]   # a dir containing SKILL.md
   
   get_plugin_manager().register(PdfSkillPlugin())
   get_plugin_manager().enable("pdf_skill")
   
   agent = Agent(name="PDF Assistant", instructions="You are a helpful assistant.")
   # ./skills/pdf-processing is now on agent._skills — discovered like a user-configured dir.
   agent.start("Extract the main findings from report.pdf")

Composition and precedence rules

These rules are locked in by the PR’s TestPluginTypedSubsystems suite:

User Interaction Flow

A PII guardrail plugin redacts a secret from the raw tool result — before the model or the user sees it:

$ python app.py
User: fetch the demo token
Agent → tool call: fetch_token()
Tool  → "your token is SECRET-abc-123"    (raw)
Guardrail chain (plugin) → "your token is [REDACTED]-abc-123"
Agent → your token is [REDACTED]-abc-123

Before this wiring the redaction only ran on the model’s final paraphrase; if the model quoted the tool verbatim, or a downstream tool used the raw result, the secret leaked.

Backward compatibility

• Every new base method (as_guardrail(), get_skills(), get_policies()) is a no-op by default — existing Plugin subclasses that don’t override them behave identically.
• The default PluginInfo.plugin_type is PluginType.HOOK, so existing plugins that don’t set the field are not collected by the typed collectors and keep routing through the generic lifecycle-hook path.
• No new required parameters on Agent(...) or Plugin(...).

Tool Plugins (subclass Plugin with get_tools())

Subclass Plugin, return callables from get_tools(), register once — every agent in the process can call them.

A class-based PluginType.TOOL plugin’s get_tools() output is merged into self.tools during Agent.__init__ — you do not call plugins.enable(), touch PluginManager.get_all_tools(), or pass anything special to Agent(...). Registering the plugin marks it enabled in the manager, and that is enough for its tools to reach every subsequent agent.

Quick Start

1. Register a tool plugin in code

   Return callables from get_tools(), register once, and the tool reaches the agent — no plugins.enable(), no tools=[...].

   from praisonaiagents import Agent, Plugin, PluginInfo
   from praisonaiagents.plugins import get_plugin_manager
   
   def random_number() -> int:
       """Returns a random number."""
       return 42
   
   class BasicToolPlugin(Plugin):
       @property
       def info(self):
           return PluginInfo(name="basic_tools")
   
       def get_tools(self):
           return [random_number]
   
   # Register once for the process.
   get_plugin_manager().register(BasicToolPlugin())
   
   # No plugins.enable(), no tools=[...] — the plugin's tool reaches the agent.
   agent = Agent(instructions="Give me a random number using your tools.")
   agent.start()

2. Load the bundled plugin by name

   The reference basic_tools plugin ships in the separate praisonai-plugins package. Install it, enable it once, and every subsequent agent gains its tool.

   pip install praisonai-plugins
   praisonai plugins enable basic_tools

   from praisonaiagents import Agent
   
   agent = Agent(instructions="Give me a random number using your tools.")
   agent.start()   # random_number is callable — no tools=[...] needed

User Interaction Flow

Before this wiring, an agent asked for a random number would fabricate one. Now it actually calls the plugin’s tool:

$ python app.py
User: give me a random number using your tools
Agent → tool call: random_number()
Tool  → 42
Agent → 42

Precedence and Safety Rules

The collision rule in code — the agent’s own tool always wins:

from praisonaiagents import Agent
from praisonaiagents.plugins import get_plugin_manager
# BasicToolPlugin from the Quick Start already registered — it exposes random_number

def random_number() -> str:
    return "agent's own version"

agent = Agent(instructions="test", tools=[random_number])
# agent.tools contains ONE random_number: the user's.
# The plugin's random_number is skipped with:
#   WARNING  Plugin tool 'random_number' skipped: an agent tool with the same name
#            already exists (existing tool wins).

@tool vs Plugin.get_tools() vs FunctionPlugin

Two ways to add a tool from a plugin — pick the one that matches how the tool is shipped.

Static Capability Manifest

Declare a plugin’s capabilities in its header so onboarding, doctor and config validation can read them without importing the plugin’s runtime.

Add four optional fields to the header — they parse as plain text, so the plugin’s runtime is never imported to read them.

Create ~/.praisonai/plugins/telegram_channel.py:

"""
Plugin Name: Telegram Channel
Description: Telegram bot adapter
Version: 1.0.0
Author: Your Name
Hooks: before_tool, after_tool
Dependencies: python-telegram-bot
Channels: telegram
Provides: send_message, get_updates
Config: api_key, timeout
Auto Enable When Configured: TELEGRAM_TOKEN
"""

from praisonaiagents import tool

@tool
def send_message(chat_id: str, text: str) -> str:
    """Send a Telegram message."""
    return f"Sent to {chat_id}: {text}"

Field reference

Read the manifest from code

discover_plugins() scans plugin directories and returns metadata dicts — the plugin body is never imported.

from praisonaiagents.plugins.discovery import discover_plugins

for meta in discover_plugins():
    print(
        meta["name"],
        meta.get("channels", []),
        meta.get("provides", []),
        meta.get("config", []),
        meta.get("auto_enable_when_configured", []),
    )

Rules

• Fields are optional — omitting them yields empty lists (safe default).
• Field keys are case-insensitive — Channels, channels, and CHANNELS all work.
• Aliases: Provides = Provides Tools = Tools; Config = Config Schema; Auto Enable When Configured = auto_enable_when_configured.
• The plugin file’s runtime is never imported to read these fields — safe to scan untrusted plugin directories for a capability inventory.

Do I need a manifest?

Lifecycle-Method Plugins

Subclass Plugin and override a lifecycle method to transform prompts, messages, or responses. Call plugins.enable() — or set PRAISONAI_PLUGINS=true / [plugins] enabled = true and let Agent construction wire it — to activate the method at runtime.

Create ~/.praisonai/plugins/pii_redactor.py:

"""
Plugin Name: PII Redactor
Description: Scrubs SSNs from LLM responses
Version: 1.0.0
"""
import re
from praisonaiagents import Plugin, PluginInfo, PluginHook

class PIIRedactor(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="pii_redactor",
            version="1.0.0",
            description="Scrubs SSNs from LLM responses",
            hooks=[PluginHook.AFTER_LLM],
        )

    def after_llm(self, response: str, usage: dict) -> str:
        return re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN]", response)

def create_plugin():
    return PIIRedactor()

Load and enable:

from praisonaiagents import Agent, plugins

plugins.enable()   # bridges lifecycle methods into the runtime hook registry

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("My SSN is 123-45-6789")   # response is redacted before the user sees it

Every lifecycle method may also be async def — the dispatcher awaits it. Reach for it when a hook does I/O, such as warming a cache at registration:

from praisonaiagents import Plugin, PluginInfo

class AsyncWarmup(Plugin):
    @property
    def info(self):
        return PluginInfo(name="warmup")

    async def on_init(self, context):
        # Awaited before the plugin receives any hook dispatch
        self.cache = await warmup_cache()

    async def on_shutdown(self):
        await self.cache.aclose()

See Sync or Async Hook Methods for how async hooks run.

Plugin Lifecycle Observability

plugin.on_init and plugin.on_shutdown are the plugin’s own callbacks — they run on the plugin being registered or unregistered.

To have a different plugin (a metrics collector, an audit log, a gateway health monitor) observe the plugin set changing at runtime, subscribe to the ON_INIT / ON_SHUTDOWN hook events. They are emitted from PluginManager.register() / PluginManager.unregister() immediately after the plugin’s own callback runs.

from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult, PluginLifecycleInput

registry = HookRegistry()

@registry.on(HookEvent.ON_INIT)
def audit_plugin_registered(event_data: PluginLifecycleInput):
    print(f"plugin_registered name={event_data.plugin_name} version={event_data.plugin_version}")
    return HookResult.allow()

@registry.on(HookEvent.ON_SHUTDOWN)
def audit_plugin_unregistered(event_data: PluginLifecycleInput):
    print(f"plugin_unregistered name={event_data.plugin_name}")
    return HookResult.allow()

Observe-only. The emit is fully guarded — a failing hook never breaks plugin loading. Zero cost when nothing is listening (the runtime early-outs on registry.has_hooks(...)). The early-out is a lock-guarded any() with no snapshot copy or priority sort (PR #5853).

See Hook Events → Plugin Lifecycle Events for the payload reference.

Session & Error Lifecycle Plugins

Override session_start, session_end, on_error, on_config, or on_auth to react to session boundaries, errors, config, and credential resolution.

Choosing a lifecycle event

Observe sessions

session_start and session_end observe when a session opens and closes.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Hello!")

Create ~/.praisonai/plugins/session_logger.py:

"""
Plugin Name: Session Logger
Description: Records when sessions start and end
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class SessionLogger(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="session_logger",
            hooks=[PluginHook.SESSION_START, PluginHook.SESSION_END],
        )

    def session_start(self, context):
        print(f"session started: {context.get('session_name')}")

    def session_end(self, context):
        print(f"session ended: {context.get('reason')} after {context.get('total_turns')} turns")

def create_plugin():
    return SessionLogger()

Observe errors

on_error observes errors during a run — use it to log without changing behavior.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Do something that might fail")

Create ~/.praisonai/plugins/error_reporter.py:

"""
Plugin Name: Error Reporter
Description: Logs errors during agent runs
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class ErrorReporter(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="error_reporter",
            hooks=[PluginHook.ON_ERROR],
        )

    def on_error(self, error_type, error_message, context):
        print(f"{error_type}: {error_message}")
        print(context.get("stack_trace"))

def create_plugin():
    return ErrorReporter()

Rewrite config

on_config returns a dict to rewrite runtime configuration in place.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Hello!")

Create ~/.praisonai/plugins/config_defaults.py:

"""
Plugin Name: Config Defaults
Description: Injects a default temperature into runtime config
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class ConfigDefaults(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="config_defaults",
            hooks=[PluginHook.ON_CONFIG],
        )

    def on_config(self, config):
        config.setdefault("temperature", 0.2)
        return config

def create_plugin():
    return ConfigDefaults()

Inject credentials

on_auth returns a dict of credentials — the bridge writes them back even when credentials starts as None, so first-time injection works.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Call the authenticated API")

Create ~/.praisonai/plugins/token_injector.py:

"""
Plugin Name: Token Injector
Description: Supplies credentials on first-time auth
Version: 1.0.0
"""
import os
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class TokenInjector(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="token_injector",
            hooks=[PluginHook.ON_AUTH],
        )

    def on_auth(self, auth_type, credentials):
        return {"token": os.environ["MY_API_TOKEN"]}

def create_plugin():
    return TokenInjector()

Gateway, Schedule & Runtime Lifecycle Plugins

Override gateway_start, schedule_trigger, before_compaction, subagent_stop, model_fallback (and their siblings) to observe the framework’s runtime lifecycle from a plugin — no low-level HookRegistry needed.

Choosing a lifecycle event

Observe gateway lifecycle

gateway_start and gateway_stop fire when the gateway (BotOS) starts and stops.

from praisonaiagents import Agent, plugins

plugins.enable()   # bridges plugin methods into the runtime hook registry

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Hello!")

Create ~/.praisonai/plugins/gateway_watch.py:

"""
Plugin Name: Gateway Watch
Description: Observes gateway start / stop
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class GatewayWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="gateway_watch",
            hooks=[PluginHook.GATEWAY_START, PluginHook.GATEWAY_STOP],
        )

    def gateway_start(self, context):
        print(f"gateway_start platforms={context.get('platforms')} bots={context.get('bot_count')}")

    def gateway_stop(self, context):
        print(f"gateway_stop reason={context.get('reason')}")

def create_plugin():
    return GatewayWatch()

Observe schedule lifecycle

schedule_add, schedule_remove, and schedule_trigger fire when a job is persisted, deleted, or fires.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Schedule a daily report.")

Create ~/.praisonai/plugins/schedule_watch.py:

"""
Plugin Name: Schedule Watch
Description: Observes schedule add / remove / trigger
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class ScheduleWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="schedule_watch",
            hooks=[
                PluginHook.SCHEDULE_ADD,
                PluginHook.SCHEDULE_REMOVE,
                PluginHook.SCHEDULE_TRIGGER,
            ],
        )

    def schedule_add(self, context):
        print(f"schedule_add job={context.get('job_name')} id={context.get('job_id')}")

    def schedule_remove(self, context):
        print(f"schedule_remove job={context.get('job_name')} id={context.get('job_id')}")

    def schedule_trigger(self, context):
        print(f"schedule_trigger job={context.get('job_name')} message={context.get('message')}")

def create_plugin():
    return ScheduleWatch()

Observe compaction

before_compaction sees the pre-compaction snapshot; after_compaction sees the post-compaction summary.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Have a long conversation that triggers compaction.")

Create ~/.praisonai/plugins/compaction_watch.py:

"""
Plugin Name: Compaction Watch
Description: Observes conversation memory compaction
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class CompactionWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="compaction_watch",
            hooks=[PluginHook.BEFORE_COMPACTION, PluginHook.AFTER_COMPACTION],
        )

    def before_compaction(self, context):
        print(f"before_compaction session={context.get('session_id')} messages={context.get('message_count')}")

    def after_compaction(self, context):
        print(f"after_compaction {context.get('messages_before')} → {context.get('messages_after')}")

def create_plugin():
    return CompactionWatch()

Observe subagent completion

subagent_stop fires when a spawned subagent reaches a terminal state — useful for audit and metrics.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Delegate work to subagents.")
agent.start("Research three topics in parallel.")

Create ~/.praisonai/plugins/subagent_watch.py:

"""
Plugin Name: Subagent Watch
Description: Observes spawned subagent completion
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class SubagentWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="subagent_watch",
            hooks=[PluginHook.SUBAGENT_STOP],
        )

    def subagent_stop(self, context):
        print(f"subagent_stop task={context.get('task')} success={context.get('success')}")

def create_plugin():
    return SubagentWatch()

Observe model fallback

model_fallback fires when the runtime falls back to a secondary model — log the primary_model → fallback_model transition and its reason.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("Answer even if the primary model is unavailable.")

Create ~/.praisonai/plugins/fallback_watch.py:

"""
Plugin Name: Fallback Watch
Description: Observes runtime model fallback
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class FallbackWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="fallback_watch",
            hooks=[PluginHook.MODEL_FALLBACK],
        )

    def model_fallback(self, context):
        print(f"model_fallback {context.get('primary_model')} → {context.get('fallback_model')} ({context.get('reason')})")

def create_plugin():
    return FallbackWatch()

Observe kanban board changes

Declare one or more of KANBAN_TASK_CREATED, KANBAN_TASK_MOVED, KANBAN_TASK_DONE and define the matching kanban_task_*(context) methods.

from praisonaiagents import Agent, plugins

plugins.enable()

agent = Agent(name="Assistant", instructions="Manage a kanban board.")
agent.start("Create and move tasks on the board.")

Create ~/.praisonai/plugins/kanban_watch.py:

"""
Plugin Name: Kanban Watch
Description: Observes kanban board transitions
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class KanbanWatch(Plugin):
    @property
    def info(self):
        return PluginInfo(
            name="kanban_watch",
            hooks=[
                PluginHook.KANBAN_TASK_CREATED,
                PluginHook.KANBAN_TASK_MOVED,
                PluginHook.KANBAN_TASK_DONE,
            ],
        )

    def kanban_task_created(self, context):
        print(f"kanban_task_created {context.get('task_id')}")

    def kanban_task_moved(self, context):
        print(f"kanban_task_moved {context.get('task_id')} → {context.get('column')}")

    def kanban_task_done(self, context):
        print(f"kanban_task_done {context.get('task_id')}")

def create_plugin():
    return KanbanWatch()

How the declaration-driven bridge works

A Plugin method is bridged when both of these are true:

1. The matching HookEvent is listed in PluginInfo.hooks.
2. The method is either a genuine override of the base no-op, or the event has no base counterpart at all (e.g. kanban_*).

Declaring the hook in info.hooks without overriding the method registers nothing (no dead hooks). Overriding the method without declaring the hook also registers nothing — the runtime is strictly declaration-driven so authors keep control of what fires.

Message-Lifecycle Plugins

Override before_message, after_message, message_sent, or message_undelivered to react to every hop of a message’s lifecycle — per-user policy, per-channel redaction, delivery telemetry, or dead-letter escalation.

Choosing a message hook

Enriched payload keys

Every message-lifecycle method receives (at most) these keys:

React per user, per channel, and per delivery

Lead with the outcome: rate-limit a noisy sender, count deliveries per channel, or escalate a dead letter.

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision

class PerUserRateLimit(Plugin):
    def __init__(self):
        self._counts = {}

    @property
    def info(self):
        return PluginInfo(name="per_user_rate_limit",
                          hooks=[PluginHook.MESSAGE_RECEIVED])

    def before_message(self, message):
        sender = message.get("sender_id")
        if sender is None:
            return message              # outbound/unknown — no scope to enforce
        self._counts[sender] = self._counts.get(sender, 0) + 1
        if self._counts[sender] > 20:
            return PluginDecision.deny(f"Rate limit exceeded for {sender}")
        return message

def create_plugin():
    return PerUserRateLimit()

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class DeliveryCounter(Plugin):
    def __init__(self):
        self._sent = {}

    @property
    def info(self):
        return PluginInfo(name="delivery_counter",
                          hooks=[PluginHook.MESSAGE_SENT])

    def message_sent(self, message):
        key = (message.get("platform"), message.get("channel_id"))
        self._sent[key] = self._sent.get(key, 0) + 1

def create_plugin():
    return DeliveryCounter()

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class DeadLetter(Plugin):
    @property
    def info(self):
        return PluginInfo(name="dead_letter",
                          hooks=[PluginHook.MESSAGE_UNDELIVERED])

    def message_undelivered(self, message):
        # Payload adds error + notice_delivered on this hook only
        print(f"[dead-letter] {message.get('platform')} "
              f"channel={message.get('channel_id')} "
              f"error={message.get('error')} "
              f"notice_delivered={message.get('notice_delivered')}")

def create_plugin():
    return DeadLetter()

Function-Based Plugins (FunctionPlugin)

FunctionPlugin wraps plain functions into a plugin without subclassing — useful for one-off hooks or when the whole plugin is a single callable.

When to reach for it

Quick Start

Register a FunctionPlugin on the plugin manager, then start the agent — the callable runs before the agent sees the message.

from praisonaiagents import Agent, FunctionPlugin, PluginHook
from praisonaiagents.plugins import get_plugin_manager

def redact_secrets(message):
    text = message.get("content", "")
    return {"content": text.replace("sk-", "sk-****")}

get_plugin_manager().register(FunctionPlugin(
    name="redact_secrets",
    hooks={PluginHook.MESSAGE_RECEIVED: redact_secrets},
))

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("here is my key sk-live-abc")   # agent sees redacted content

Constructor

FunctionPlugin(
    name: str,
    hooks: Optional[Dict[PluginHook, Callable]] = None,
    version: str = "1.0.0",
    description: str = "",
)

Hook dispatch table

Each key in hooks={} is a PluginHook enum value; the callable’s return value drives the effect below.

Message-lifecycle examples

Lead with the outcome: rate-limit a channel, log every delivery, or capture permanent failures.

from praisonaiagents import FunctionPlugin, PluginHook, PluginDecision
from praisonaiagents.plugins import get_plugin_manager

counts = {}
def gate(message):
    ch = message.get("channel_id")
    if ch is None:
        return message
    counts[ch] = counts.get(ch, 0) + 1
    if counts[ch] > 20:
        return PluginDecision.deny(f"rate limit exceeded for {ch}")
    return message

get_plugin_manager().register(FunctionPlugin(
    name="channel_rate_limit",
    hooks={PluginHook.MESSAGE_RECEIVED: gate},
))

from praisonaiagents import FunctionPlugin, PluginHook
from praisonaiagents.plugins import get_plugin_manager

def on_sent(message):
    print(f"[sent] {message.get('platform')} channel={message.get('channel_id')}")

get_plugin_manager().register(FunctionPlugin(
    name="delivery_log",
    hooks={PluginHook.MESSAGE_SENT: on_sent},
))

from praisonaiagents import FunctionPlugin, PluginHook
from praisonaiagents.plugins import get_plugin_manager

def on_dead(message):
    # error + notice_delivered are only added on the MESSAGE_UNDELIVERED bridge
    print(f"[dead] {message.get('channel_id')} err={message.get('error')} "
          f"notice_delivered={message.get('notice_delivered')}")

get_plugin_manager().register(FunctionPlugin(
    name="dead_letter_log",
    hooks={PluginHook.MESSAGE_UNDELIVERED: on_dead},
))

Auto-Enable from Env or Config

Agent construction auto-enables plugins when the environment or config file requests it — no explicit plugins.enable() call needed.

Set the env var, then any Agent(...) wires plugins before it runs:

export PRAISONAI_PLUGINS=true

from praisonaiagents import Agent

agent = Agent(name="Assistant", instructions="Help users")
agent.start("Hello!")   # plugins are already active

Or turn them on in .praisonai/config.toml:

[plugins]
enabled = true          # or false, or ["logging", "metrics"]

Under the hood, Agent init calls plugins.maybe_enable_from_config(), which reads the env var and config file, then runs enable(get_enabled_plugins()) exactly once per process.

Precedence: explicit plugins.enable(...) > PRAISONAI_PLUGINS env var > [plugins] in config.yaml / config.toml > disabled.

Suppress Plugins for One Run (Pure Mode)

The inverse of auto-enable — skip external-plugin discovery for a single run without touching persisted state.

Pass --pure (or --no-plugins) to run, chat, or code:

praisonai run --pure "Summarise this file"

Or forced-off in embedded Python, regardless of the env var:

from praisonaiagents.plugins import PluginManager

manager = PluginManager(disabled=True)
manager.discover_entry_points()   # returns 0; no plugins loaded

Precedence: constructor disabled=True > PRAISONAI_NO_PLUGINS env var. The CLI flag is scoped and restored on return, and never mutates .praisonai/config.yaml.

praisonai gateway start honours the same --pure / --no-plugins suppression, with gateway-specific semantics (persisted across restart, also disables inbound YAML hooks) — see Gateway Safe Mode.

See Pure Mode for the full guide.

Configure Plugins from .praisonai/config.yaml

Turn plugins on and hand each one its own options from a single YAML file — no code.

~/.praisonai/config.yaml:

plugins:
  enabled: true
  pii_guardrail:
    redact: [email, phone]
  memory_watchdog:
    interval: 30

app.py:

from praisonaiagents import Agent

agent = Agent(name="Assistant", instructions="Help users.")
agent.start("My email is user@example.com")   # pii_guardrail redacts before the LLM sees it

Where the config file is found

1. .praisonai/config.toml   → .praisonai/config.yaml   → .praisonai/config.yml   (project-local)
2. praisonai.toml           → praisonai.yaml           → praisonai.yml           (project root)
3. ~/.praisonai/config.toml → ~/.praisonai/config.yaml → ~/.praisonai/config.yml (user global)

The four shapes of enabled

Per-plugin option blocks

Reserved keys (enabled, auto_discover, directories, allow_project_plugins) configure the plugin system; any other key whose value is a mapping is treated as a per-plugin option map delivered to that plugin’s on_config hook. Inside a per-plugin block, allow_conversation is also framework-reserved — the manager reads it to gate conversation-content hooks and does not forward it to on_config.

plugins:
  enabled: true
  allow_project_plugins: true   # authorise ./.praisonai/plugins/*.py
  pii_guardrail:
    redact: [email, phone]
    enabled: true
  memory_watchdog:
    interval: 30
    enabled: false     # disable just this one; pii_guardrail still runs

Reading options inside a plugin — on_config

apply_plugin_options() invokes plugin.on_config(options) for every enabled plugin that has a configured option map — errors in a single plugin are logged and do not abort delivery to the others.

Create ~/.praisonai/plugins/pii_guardrail.py:

"""
Plugin Name: PII Guardrail
Description: Redacts configured fields before the LLM sees them
Version: 1.0.0
"""
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook

class PIIGuardrail(Plugin):
    def __init__(self):
        self._fields = []

    @property
    def info(self):
        return PluginInfo(
            name="pii_guardrail",
            hooks=[PluginHook.ON_CONFIG, PluginHook.BEFORE_LLM],
        )

    def on_config(self, config):
        # config is the exact dict from `plugins.pii_guardrail:` in YAML.
        self._fields = list(config.get("redact", []))
        return config

    def before_llm(self, messages, params):
        # Use self._fields to scrub before the LLM call...
        return messages, params

def create_plugin():
    return PIIGuardrail()

Deliver options from code — options_by_name

Skip the file and hand each plugin its options directly. plugins.enable() takes an options_by_name map that it delivers to each plugin’s on_config hook:

from praisonaiagents import plugins

plugins.enable(
    ["pii_guardrail"],
    options_by_name={"pii_guardrail": {"redact": ["email"]}},
)

Choose a plugin config surface

Precedence

Explicit plugins.enable(...) in code > PRAISONAI_PLUGINS env var > [plugins] in config.yaml / config.toml > disabled by default.

Edge cases

• PyYAML missing — YAML configs silently return empty (debug log only); install pyyaml to enable.
• Unknown reserved key — the wrapper resolver’s typo-suggestion validator only checks against {enabled, auto_discover, directories, allow_project_plugins}; per-plugin blocks (dicts) are always accepted.
• Removed plugin block on reload — the manager stores options with replace=True by default, so a block dropped from a later config no longer delivers stale options.
• Plugin on_config raises — logged as a warning; delivery to other plugins is not aborted.

Config-write single source of truth

praisonai plugins enable/disable write to the same file the runtime reads — no separate JSON. The write target is resolved in this order:

1. .praisonai/config.toml   → .praisonai/config.yaml     (project-local)
2. praisonai.toml           → praisonai.yaml             (project root)
3. ~/.praisonai/config.toml → ~/.praisonai/config.yaml   (user global)
4. (none found)             → .praisonai/config.yaml     (default)

How the Bridge Works

plugins.enable() auto-calls wire_into_hook_registry() — no manual step. Only lifecycle methods a plugin actually overrides (or declares in PluginInfo.hooks) are bridged, so a plugin with one guardrail never fires on every event. Return a new value and the bridge writes it back onto the payload in place. Errors in a lifecycle method are non-fatal, and plugins.disable([...]) calls unwire_from_hook_registry(name) so the plugin truly stops firing.

Every method in this table may be declared either def or async def — see Sync or Async Hook Methods.

on_config and on_auth write their returned dict back onto the payload even when the target attribute starts as None — so a plugin can inject credentials the first time they’re requested, not only edit an existing dict.

Sync or Async Hook Methods

Any method on Plugin — before_tool, before_message, session_start, gateway_start, on_init, on_shutdown, on_config, and the rest — may be def or async def. Sync hooks behave exactly as before; an async hook’s coroutine is awaited before the next plugin’s hook runs.

import aiohttp
from praisonaiagents import Agent, plugins
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, GuardrailBlocked

class AsyncPolicyPlugin(Plugin):
    @property
    def info(self):
        return PluginInfo(name="async_policy", hooks=[PluginHook.BEFORE_TOOL])

    async def before_tool(self, tool_name, args):
        # Create the client inside the hook (see Warning below)
        timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.post(
                "https://policy.example.com/check",
                json={"tool": tool_name, "args": args},
            ) as resp:
                allowed = (await resp.json())["allowed"]
        if not allowed:
            raise GuardrailBlocked(f"{tool_name} denied by policy")
        return args

plugins.get_plugin_manager().register(AsyncPolicyPlugin())
plugins.enable()

agent = Agent(name="Policy-Guarded Agent", instructions="Help the user.", tools=["bash"])
agent.start("Delete the temp folder")

GuardrailBlocked and PluginDecision from an async hook block exactly like their sync forms — the model sees the reason and the tool is skipped.

Registration guarantees for an async on_init:

• The plugin receives no hook dispatch until on_init finishes.
• A failing on_init rolls the registration back — retrying with a fixed plugin works.
• on_init may call back into the manager (disable(name), enable(name)) without deadlocking, and a plugin that disables itself during on_init stays disabled.

Reference Plugins in praisonai-plugins

Batteries-included plugins ship in the separate praisonai-plugins package — install once, then enable by name.

cli_backend_tracer overrides Plugin.cli_backend_execute(context) and logs every CLI backend delegation to the standard praisonai logger — with the prompt already redacted, so the log stream is safe to ship to an aggregator.

pip install praisonai-plugins
export PRAISONAI_CLI_BACKEND_DEBUG=1
praisonai plugins enable cli_backend_tracer

See Hook Events → CLI Backend Events for the full CliBackendExecuteInput payload reference.

Blocking Plugins (Guardrails & Policies)

Guardrail and policy plugins can now stop an action before it happens, not just rewrite it — refuse a dangerous tool call, drop a spam message, or decline an LLM request.

Quick Start

1. Block a tool call

   Return PluginDecision.block(reason) from before_tool to skip a forbidden tool.

   from praisonaiagents import Agent, plugins
   from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision
   
   class BlockDeleteFiles(Plugin):
       @property
       def info(self):
           return PluginInfo(
               name="block_delete_files",
               hooks=[PluginHook.BEFORE_TOOL],
           )
   
       def before_tool(self, tool_name, args):
           if tool_name == "delete_file":
               return PluginDecision.block("File deletion is not permitted")
           return args
   
   plugins.get_plugin_manager().register(BlockDeleteFiles())
   plugins.enable()
   
   agent = Agent(name="Assistant", instructions="Help users.", tools=["delete_file", "read_file"])
   agent.start("Delete /etc/hosts")   # tool call skipped; block reason surfaced

2. Refuse an LLM request

   Return PluginDecision.deny(reason) from before_llm and the agent returns "[LLM request blocked by hook: <reason>]" without calling the model.

   from praisonaiagents import Agent, plugins
   from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision
   
   class BlockLLM(Plugin):
       @property
       def info(self):
           return PluginInfo(
               name="block_llm",
               hooks=[PluginHook.BEFORE_LLM],
           )
   
       def before_llm(self, messages, params):
           return PluginDecision.deny("LLM request refused by policy")
   
   plugins.get_plugin_manager().register(BlockLLM())
   plugins.enable()
   
   agent = Agent(name="Assistant", instructions="Help users.")
   print(agent.start("Hello"))   # [LLM request blocked by hook: LLM request refused by policy]

3. Drop an inbound message

   Return PluginDecision.deny(reason) from before_message to drop a message before the agent sees it.

   from praisonaiagents import Agent, plugins
   from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision
   
   class BlockSpam(Plugin):
       @property
       def info(self):
           return PluginInfo(
               name="block_spam",
               hooks=[PluginHook.MESSAGE_RECEIVED],
           )
   
       def before_message(self, message):
           if "spam" in message.get("content", ""):
               return PluginDecision.deny("Spam blocked")
           return message
   
   plugins.get_plugin_manager().register(BlockSpam())
   plugins.enable()
   
   agent = Agent(name="Assistant", instructions="Help users.")
   agent.start("buy spam now")   # message dropped before the agent runs

Three Ways to Block

The same block, written three ways — pick the one that reads cleanest in your code.

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision

class BlockDelete(Plugin):
    @property
    def info(self):
        return PluginInfo(name="block_delete", hooks=[PluginHook.BEFORE_TOOL])

    def before_tool(self, tool_name, args):
        if tool_name == "delete_file":
            return PluginDecision.block("File deletion not permitted")
        return args

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook
from praisonaiagents.hooks.types import HookResult

class BlockDelete(Plugin):
    @property
    def info(self):
        return PluginInfo(name="block_delete", hooks=[PluginHook.BEFORE_TOOL])

    def before_tool(self, tool_name, args):
        if tool_name == "delete_file":
            return HookResult.deny("Denied via HookResult")
        return args

from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, GuardrailBlocked

class BlockDelete(Plugin):
    @property
    def info(self):
        return PluginInfo(name="block_delete", hooks=[PluginHook.BEFORE_TOOL])

    def before_tool(self, tool_name, args):
        if tool_name == "delete_file":
            raise GuardrailBlocked("File deletion not permitted")
        return args

PluginDecision.deny(reason) and PluginDecision.block(reason) both stop the action (is_denied() is True for each); allow(reason=None) is an explicit no-op. GuardrailBlocked(reason: str = "Blocked by guardrail plugin") is caught by the bridge and converted to a block. It may be raised from any before_* method, from after_tool — the tool has already run there, but the output is suppressed before it reaches the model (see Redact / Block Tool Output) — and from after_agent, which withholds the final response (see Output Guardrails).

Where Blocks Fire

The five before_ methods, after_tool and after_agent can block; the remaining after_ / observational methods cannot. after_llm stays non-blocking because it also fires on intermediate tool-calling turns — use after_agent to gate the final response.

Output Guardrails

after_agent is the final-response gate — return a rewritten string to redact, or deny to withhold the response.

from praisonaiagents import Agent, plugins
from praisonaiagents.plugins import Plugin, PluginInfo, PluginHook, PluginDecision, GuardrailBlocked

class SecretRedactor(Plugin):
    @property
    def info(self):
        return PluginInfo(name="secret_redactor", hooks=[PluginHook.AFTER_AGENT])

    def after_agent(self, response, context):
        return response.replace("sk-abc123xyz", "[REDACTED]")   # rewrite is adopted

class PiiBlocker(Plugin):
    @property
    def info(self):
        return PluginInfo(name="pii_blocker", hooks=[PluginHook.AFTER_AGENT])

    def after_agent(self, response, context):
        if "123-45-6789" in response:
            raise GuardrailBlocked("PII in response")   # or: return PluginDecision.deny("PII in response")
        return response

manager = plugins.get_plugin_manager()
manager.register(SecretRedactor())
manager.register(PiiBlocker())
# after_agent is a conversation-content hook — your own plugins need the grant
plugins.enable(options_by_name={
    "secret_redactor": {"allow_conversation": True},
    "pii_blocker": {"allow_conversation": True},
})

agent = Agent(name="Assistant", instructions="Help users.")
print(agent.start("What is my SSN?"))   # [Response blocked by hook: PII in response]

• Stacked plugins compose. AFTER_AGENT plugins run sequentially in registration order (same as MESSAGE_SENDING), so each sees the previous rewrite; the first denial wins.
• Blocked text is not kept. The matching in-memory assistant turn is replaced with the final value, so later turns, auto-memory and self-learning never see the withheld or unredacted text.
• Non-string outputs pass through. Structured / Pydantic responses are left unchanged.
• Needs the conversation grant. Non-bundled plugins must set allow_conversation: true or after_agent never fires — see Conversation-Content Capability Gate.

How a Block Flows

Which Style Should I Use?

Ship a Plugin as a pip Package

Register your plugin class in the praisonai.plugins entry-point group in pyproject.toml:

[project.entry-points."praisonai.plugins"]
my_plugin = "my_package.plugins:MyPlugin"

Agent construction (or plugins.enable()) auto-discovers and bridges it — no user code changes needed.

Once installed, verify the plugin registered with praisonai plugins add <package> --dry-run — see plugins add.

CLI Commands

praisonai plugins list
praisonai plugins enable pii_guardrail
praisonai plugins disable pii_guardrail
praisonai plugins reload
praisonai plugins doctor
praisonai plugins add praisonai-my-plugin
praisonai plugins add praisonai-my-plugin --dry-run

praisonai plugins list shows every plugin with its source (entry_point:<dist>, registered, or single_file). enable/disable persist to the config the runtime reads, reload picks up edits without a restart, and doctor diagnoses issues. See the full reference on the Plugins CLI page.

Configuration Options

Only Plugin Name is required; every other field is optional.

Best Practices

Thread Safety

The process-wide plugin manager returned by get_plugin_manager() is safe to share across concurrently running agents.

All lifecycle methods — register, unregister, enable, disable, execute_hook — synchronise on a single RLock, so concurrent registration and hook firing cannot collide. See Thread-Safe Agent State → PluginManager (PR #5204) for the mechanism.

Related