Plugins

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?")

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

export PRAISONAI_ALLOW_PROJECT_PLUGINS=true

plugins:
  allow_project_plugins: true

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

from praisonaiagents.plugins.manager import PluginManager

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

from praisonaiagents.plugins import get_plugin_registry

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

"""
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}")

"""
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")

$ 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

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

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).

"""
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}"

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", []),
    )

"""
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()

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

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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

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!")

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

from praisonaiagents import Agent, plugins

plugins.enable()

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

"""
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()

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.")

"""
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()

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

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

export PRAISONAI_PLUGINS=true

from praisonaiagents import Agent

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

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

praisonai run --pure "Summarise this file"

from praisonaiagents.plugins import PluginManager

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

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

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

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)

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

"""
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()

from praisonaiagents import plugins

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

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)