Plugin Architecture for Tools in a Custom Runtime

Q: How do I version a breaking tool change?

Introduce refund_order_v2, migrate prompts/evals, then remove v1 after a deprecation window. Do not silently change required args.

Q: Where do human approvals plug in?

In the dispatcher: if risk is write/dangerous and policy requires approval, return error_type=needs_approval or pause the run before handle.

Q: Can plugins live in another package/repo?

Yes. Publish an internal package that exports register(registry) and pin versions. Review supply chain like any dependency.

Q: How do plugins interact with circuit breakers?

Use breaker_key metadata so the dispatcher wraps remote calls uniformly. See the circuit breakers article in this section.

Q: How should errors be shown to the model?

Stable JSON with error_type and a short message. Avoid huge stack traces in the transcript.

Q: How do I test a plugin?

Call handle with a fake AuthContext and fixed args. Separately test registry policy denials without invoking real side effects.

Hardcoding if tool_name == ... does not scale past a handful of tools. A plugin architecture lets each capability register a schema, handler, and metadata while the ReAct loop stays a thin dispatcher. This recipe stays framework-free and portable to later hosts (native SDKs or LangGraph adapters).

Summary

Define a ToolPlugin contract (name, description, JSON Schema, handler, risk flags), register plugins into a registry that builds model tool specs and dispatches calls, and filter the active set per route, tenant, or role before each run.

Recipe

Specify a plugin protocol: name, description, parameters schema, handler(args, ctx), metadata.
Keep handlers pure relative to the model SDK; inject AuthContext from the host.
Implement ToolRegistry.register / specs() / dispatch(name, args, ctx).
Support allowlists and denylists per agent profile.
Validate args against the schema before calling the handler (jsonschema or Pydantic).
Load plugins explicitly at startup (import side effects or an explicit list) - avoid magic autoload in production until you need it.
Version tool names carefully; treat renames as breaking API changes for prompts and evals.
Unit-test registry isolation: unknown tools, validation failures, and policy denials.

Working Example

from __future__ import annotations
 
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
 
 
@dataclass(frozen=True)
class AuthContext:
    user_id: str
    roles: frozenset[str]
    tenant_id: str
 
 
@dataclass
class ToolResult:
    ok: bool
    data: Any = None
    error_type: str | None = None
    message: str | None = None
 
 
class ToolPlugin(Protocol):
    name: str
    description: str
    parameters: dict[str, Any]  # JSON Schema object
    risk: str  # "read" | "write" | "dangerous"
 
    def handle(self, args: dict[str, Any], ctx: AuthContext) -> ToolResult: ...
 
 
@dataclass
class FunctionPlugin:
    name: str
    description: str
    parameters: dict[str, Any]
    risk: str
    fn: Callable[[dict[str, Any], AuthContext], Any]
 
    def handle(self, args: dict[str, Any], ctx: AuthContext) -> ToolResult:
        try:
            return ToolResult(ok=True, data=self.fn(args, ctx))
        except Exception as exc:
            return ToolResult(ok=False, error_type="tool_error", message=str(exc))
 
 
def _get_order_status(args: dict[str, Any], ctx: AuthContext) -> dict[str, Any]:
    # demo: ignore tenant; production: scope by ctx.tenant_id
    order_id = args["order_id"]
    return {"order_id": order_id, "status": "shipped", "tenant": ctx.tenant_id}
 
 
def _refund_order(args: dict[str, Any], ctx: AuthContext) -> dict[str, Any]:
    if "refunds" not in ctx.roles:
        raise PermissionError("role refunds required")
    return {"order_id": args["order_id"], "refunded": True}
 
 
class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolPlugin] = {}
 
    def register(self, plugin: ToolPlugin) -> None:
        if plugin.name in self._tools:
            raise ValueError(f"duplicate tool: {plugin.name}")
        self._tools[plugin.name] = plugin
 
    def specs(self, allow: set[str] | None = None) -> list[dict[str, Any]]:
        names = allow if allow is not None else set(self._tools)
        out: list[dict[str, Any]] = []
        for name in sorted(names):
            p = self._tools[name]
            out.append(
                {
                    "type": "function",
                    "function": {
                        "name": p.name,
                        "description": p.description,
                        "parameters": p.parameters,
                    },
                }
            )
        return out
 
    def dispatch(
        self,
        name: str,
        args: dict[str, Any],
        ctx: AuthContext,
        allow: set[str] | None = None,
    ) -> ToolResult:
        if allow is not None and name not in allow:
            return ToolResult(
                ok=False, error_type="tool_policy", message=f"tool not allowed: {name}"
            )
        plugin = self._tools.get(name)
        if plugin is None:
            return ToolResult(
                ok=False, error_type="unknown_tool", message=f"unknown tool: {name}"
            )
        # optional: jsonschema.validate(args, plugin.parameters)
        return plugin.handle(args, ctx)
 
 
def build_default_registry() -> ToolRegistry:
    reg = ToolRegistry()
    reg.register(
        FunctionPlugin(
            name="get_order_status",
            description="Look up an order status by id.",
            parameters={
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
            risk="read",
            fn=_get_order_status,
        )
    )
    reg.register(
        FunctionPlugin(
            name="refund_order",
            description="Refund an order. Requires refunds role.",
            parameters={
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
            risk="write",
            fn=_refund_order,
        )
    )
    return reg
 
 
# --- use inside the loop ---
registry = build_default_registry()
SUPPORT_ALLOW = {"get_order_status"}  # support agent cannot refund
 
 
def run_tool_for_support(name: str, raw_args: str, ctx: AuthContext) -> str:
    args = json.loads(raw_args or "{}")
    result = registry.dispatch(name, args, ctx, allow=SUPPORT_ALLOW)
    return json.dumps(result.__dict__)

Profiles (SUPPORT_ALLOW) prevent one mega-agent with every dangerous tool.
AuthZ stays in handlers or a policy layer, never "the model promised it was OK."
Duplicate registration fails at startup, not mid-run.

Deep Dive

Why plugins beat a switch statement

Concern	Switch / if-chain	Plugin registry
Adding a tool	Edit the loop module	Register in one place
Per-agent allowlists	Scattered conditionals	Filter `specs()` / `dispatch`
Testing	Hard to isolate	Test plugin + registry separately
Future host swap	Logic stuck in loop	Adapters wrap same plugins

Plugin metadata worth storing

Field	Use
`risk`	Default approval gates for write/dangerous
`timeout_s`	Per-tool deadlines
`breaker_key`	Map to dependency circuit breakers
`version`	Evals and changelog
`tags`	`pii`, `payments`, `read-only`

Loading strategies

Strategy	Pros	Cons
Explicit list in `build_default_registry`	Clear, reviewable	Manual wiring
Entry points / setuptools plugins	Third-party extensions	Supply-chain risk
Directory import of `plugins/*.py`	Convenient	Import-order and side-effect bugs
Remote MCP tool list	Share tools across runtimes	Network + auth complexity

For most product agents, explicit registration is the right default. Treat MCP as another plugin source that translates remote tools into the same ToolPlugin shape.

MCP as a plugin backend

MCP client list_tools -> ToolPlugin adapters -> registry
MCP call_tool        <- dispatch

The loop should not special-case MCP. It only sees registry specs and dispatch results.

Validation and coercion

Validate before side effects:

JSON parse arguments from the model.
Validate against JSON Schema / Pydantic model.
On failure, return error_type=validation_error observation; do not raise out of the loop unless it is a host bug.

Multi-agent profiles

PROFILES = {
    "support": {"get_order_status", "search_kb"},
    "billing": {"get_order_status", "refund_order"},
    "admin_read": {"get_order_status", "list_audit_events"},
}

Bind profiles to routes or handoff targets. Smaller catalogs improve tool selection quality.

Keeping domain code free of registry types

domain/orders.py          # refund_order_domain(order_id, tenant_id)
runtime/plugins/orders.py # FunctionPlugin wrapping domain
runtime/registry.py
runtime/loop.py

Domain functions remain callable from non-agent jobs and tests.

Gotchas

Autoloading untrusted plugins in production without a review process.
One global allowlist for every user role.
Letting the model pass user_id / tenant_id instead of injecting AuthContext.
Renaming tools casually and breaking prompts, caches, and evals.
No schema validation - handlers crash on missing keys mid-flight.
Registering write tools as risk=read and skipping approvals.
Giant descriptions that waste context; keep schemas tight.
Catching all exceptions and returning success-shaped data - models then invent happy paths.

Alternatives

Approach	When	Trade-off
Plugin registry (this page)	Custom runtime with growing tools	You maintain the contract
Decorator `@tool` on functions	Small codebases	Can hide allowlist/policy if overused
Framework tool base classes	Already on LangGraph/Crew/etc.	Framework coupling
MCP-only tool mesh	Many external tool servers	Operational overhead for simple apps
OpenAI Agents `function_tool`	Native SDK path	Provider-aligned packaging

FAQs

How many tools is too many on one agent?

When selection quality drops or descriptions no longer fit context budgets. Split profiles or use a router agent with small specialist catalogs.

Should plugins be async?

If your host is async and tools do I/O, yes. Keep the registry API consistent (async def handle or a thread offload policy) so the loop does not block.

How do I version a breaking tool change?

Introduce refund_order_v2, migrate prompts/evals, then remove v1 after a deprecation window. Do not silently change required args.

Where do human approvals plug in?

In the dispatcher: if risk is write/dangerous and policy requires approval, return error_type=needs_approval or pause the run before handle.

Can plugins live in another package/repo?

Yes. Publish an internal package that exports register(registry) and pin versions. Review supply chain like any dependency.

How do plugins interact with circuit breakers?

Use breaker_key metadata so the dispatcher wraps remote calls uniformly. See the circuit breakers article in this section.

Do I need a plugin system for three tools?

A tiny dict of callables is fine. Introduce a formal registry when allowlists, risk metadata, or multiple profiles appear.

How should errors be shown to the model?

Stable JSON with error_type and a short message. Avoid huge stack traces in the transcript.

Is this the same as dependency injection frameworks?

It is a narrow registry for tools, not a general DI container. Keep it small on purpose.

How do I test a plugin?

Call handle with a fake AuthContext and fixed args. Separately test registry policy denials without invoking real side effects.

Custom Agent Runtimes Basics - simple dispatcher before plugins
Designing a ReAct Loop from Scratch in FastAPI or Express - host wiring
Circuit Breakers and Observability Without a Third-Party Framework - dependency protection
Framework Lock-In Risks and How to Avoid Them - portable tool boundaries
Tool Design Rules for Safe Composable Agents - policy rules
Custom Agent Runtimes Best Practices - section checklist

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Summary

Recipe

Specify a plugin protocol: name, description, parameters schema, handler(args, ctx), metadata.
Keep handlers pure relative to the model SDK; inject AuthContext from the host.
Implement ToolRegistry.register / specs() / dispatch(name, args, ctx).
Support allowlists and denylists per agent profile.
Validate args against the schema before calling the handler (jsonschema or Pydantic).
Load plugins explicitly at startup (import side effects or an explicit list) - avoid magic autoload in production until you need it.
Version tool names carefully; treat renames as breaking API changes for prompts and evals.
Unit-test registry isolation: unknown tools, validation failures, and policy denials.

Working Example

from __future__ import annotations
 
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
 
 
@dataclass(frozen=True)
class AuthContext:
    user_id: str
    roles: frozenset[str]
    tenant_id: str
 
 
@dataclass
class ToolResult:
    ok: bool
    data: Any = None
    error_type: str | None = None
    message: str | None = None
 
 
class ToolPlugin(Protocol):
    name: str
    description: str
    parameters: dict[str, Any]  # JSON Schema object
    risk: str  # "read" | "write" | "dangerous"
 
    def handle(self, args: dict[str, Any], ctx: AuthContext) -> ToolResult: ...
 
 
@dataclass
class FunctionPlugin:
    name: str
    description: str
    parameters: dict[str, Any]
    risk: str
    fn: Callable[[dict[str, Any], AuthContext], Any]
 
    def handle(self, args: dict[str, Any], ctx: AuthContext) -> ToolResult:
        try:
            return ToolResult(ok=True, data=self.fn(args, ctx))
        except Exception as exc:
            return ToolResult(ok=False, error_type="tool_error", message=str(exc))
 
 
def _get_order_status(args: dict[str, Any], ctx: AuthContext) -> dict[str, Any]:
    # demo: ignore tenant; production: scope by ctx.tenant_id
    order_id = args["order_id"]
    return {"order_id": order_id, "status": "shipped", "tenant": ctx.tenant_id}
 
 
def _refund_order(args: dict[str, Any], ctx: AuthContext) -> dict[str, Any]:
    if "refunds" not in ctx.roles:
        raise PermissionError("role refunds required")
    return {"order_id": args["order_id"], "refunded": True}
 
 
class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolPlugin] = {}
 
    def register(self, plugin: ToolPlugin) -> None:
        if plugin.name in self._tools:
            raise ValueError(f"duplicate tool: {plugin.name}")
        self._tools[plugin.name] = plugin
 
    def specs(self, allow: set[str] | None = None) -> list[dict[str, Any]]:
        names = allow if allow is not None else set(self._tools)
        out: list[dict[str, Any]] = []
        for name in sorted(names):
            p = self._tools[name]
            out.append(
                {
                    "type": "function",
                    "function": {
                        "name": p.name,
                        "description": p.description,
                        "parameters": p.parameters,
                    },
                }
            )
        return out
 
    def dispatch(
        self,
        name: str,
        args: dict[str, Any],
        ctx: AuthContext,
        allow: set[str] | None = None,
    ) -> ToolResult:
        if allow is not None and name not in allow:
            return ToolResult(
                ok=False, error_type="tool_policy", message=f"tool not allowed: {name}"
            )
        plugin = self._tools.get(name)
        if plugin is None:
            return ToolResult(
                ok=False, error_type="unknown_tool", message=f"unknown tool: {name}"
            )
        # optional: jsonschema.validate(args, plugin.parameters)
        return plugin.handle(args, ctx)
 
 
def build_default_registry() -> ToolRegistry:
    reg = ToolRegistry()
    reg.register(
        FunctionPlugin(
            name="get_order_status",
            description="Look up an order status by id.",
            parameters={
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
            risk="read",
            fn=_get_order_status,
        )
    )
    reg.register(
        FunctionPlugin(
            name="refund_order",
            description="Refund an order. Requires refunds role.",
            parameters={
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
            risk="write",
            fn=_refund_order,
        )
    )
    return reg
 
 
# --- use inside the loop ---
registry = build_default_registry()
SUPPORT_ALLOW = {"get_order_status"}  # support agent cannot refund
 
 
def run_tool_for_support(name: str, raw_args: str, ctx: AuthContext) -> str:
    args = json.loads(raw_args or "{}")
    result = registry.dispatch(name, args, ctx, allow=SUPPORT_ALLOW)
    return json.dumps(result.__dict__)

Profiles (SUPPORT_ALLOW) prevent one mega-agent with every dangerous tool.
AuthZ stays in handlers or a policy layer, never "the model promised it was OK."
Duplicate registration fails at startup, not mid-run.

Deep Dive

Why plugins beat a switch statement

Concern	Switch / if-chain	Plugin registry
Adding a tool	Edit the loop module	Register in one place
Per-agent allowlists	Scattered conditionals	Filter `specs()` / `dispatch`
Testing	Hard to isolate	Test plugin + registry separately
Future host swap	Logic stuck in loop	Adapters wrap same plugins

Plugin metadata worth storing

Field	Use
`risk`	Default approval gates for write/dangerous
`timeout_s`	Per-tool deadlines
`breaker_key`	Map to dependency circuit breakers
`version`	Evals and changelog
`tags`	`pii`, `payments`, `read-only`

Loading strategies

Strategy	Pros	Cons
Explicit list in `build_default_registry`	Clear, reviewable	Manual wiring
Entry points / setuptools plugins	Third-party extensions	Supply-chain risk
Directory import of `plugins/*.py`	Convenient	Import-order and side-effect bugs
Remote MCP tool list	Share tools across runtimes	Network + auth complexity

For most product agents, explicit registration is the right default. Treat MCP as another plugin source that translates remote tools into the same ToolPlugin shape.

MCP as a plugin backend

MCP client list_tools -> ToolPlugin adapters -> registry
MCP call_tool        <- dispatch

The loop should not special-case MCP. It only sees registry specs and dispatch results.

Validation and coercion

Validate before side effects:

JSON parse arguments from the model.
Validate against JSON Schema / Pydantic model.
On failure, return error_type=validation_error observation; do not raise out of the loop unless it is a host bug.

Multi-agent profiles

PROFILES = {
    "support": {"get_order_status", "search_kb"},
    "billing": {"get_order_status", "refund_order"},
    "admin_read": {"get_order_status", "list_audit_events"},
}

Bind profiles to routes or handoff targets. Smaller catalogs improve tool selection quality.

Keeping domain code free of registry types

domain/orders.py          # refund_order_domain(order_id, tenant_id)
runtime/plugins/orders.py # FunctionPlugin wrapping domain
runtime/registry.py
runtime/loop.py

Domain functions remain callable from non-agent jobs and tests.

Gotchas

Autoloading untrusted plugins in production without a review process.
One global allowlist for every user role.
Letting the model pass user_id / tenant_id instead of injecting AuthContext.
Renaming tools casually and breaking prompts, caches, and evals.
No schema validation - handlers crash on missing keys mid-flight.
Registering write tools as risk=read and skipping approvals.
Giant descriptions that waste context; keep schemas tight.
Catching all exceptions and returning success-shaped data - models then invent happy paths.

Alternatives

Approach	When	Trade-off
Plugin registry (this page)	Custom runtime with growing tools	You maintain the contract
Decorator `@tool` on functions	Small codebases	Can hide allowlist/policy if overused
Framework tool base classes	Already on LangGraph/Crew/etc.	Framework coupling
MCP-only tool mesh	Many external tool servers	Operational overhead for simple apps
OpenAI Agents `function_tool`	Native SDK path	Provider-aligned packaging

FAQs

How many tools is too many on one agent?

When selection quality drops or descriptions no longer fit context budgets. Split profiles or use a router agent with small specialist catalogs.

Should plugins be async?

If your host is async and tools do I/O, yes. Keep the registry API consistent (async def handle or a thread offload policy) so the loop does not block.

How do I version a breaking tool change?

Introduce refund_order_v2, migrate prompts/evals, then remove v1 after a deprecation window. Do not silently change required args.

Where do human approvals plug in?

In the dispatcher: if risk is write/dangerous and policy requires approval, return error_type=needs_approval or pause the run before handle.

Can plugins live in another package/repo?

Yes. Publish an internal package that exports register(registry) and pin versions. Review supply chain like any dependency.

How do plugins interact with circuit breakers?

Use breaker_key metadata so the dispatcher wraps remote calls uniformly. See the circuit breakers article in this section.

Do I need a plugin system for three tools?

A tiny dict of callables is fine. Introduce a formal registry when allowlists, risk metadata, or multiple profiles appear.

How should errors be shown to the model?

Stable JSON with error_type and a short message. Avoid huge stack traces in the transcript.

Is this the same as dependency injection frameworks?

It is a narrow registry for tools, not a general DI container. Keep it small on purpose.

How do I test a plugin?

Call handle with a fake AuthContext and fixed args. Separately test registry policy denials without invoking real side effects.

Custom Agent Runtimes Basics - simple dispatcher before plugins
Designing a ReAct Loop from Scratch in FastAPI or Express - host wiring
Circuit Breakers and Observability Without a Third-Party Framework - dependency protection
Framework Lock-In Risks and How to Avoid Them - portable tool boundaries
Tool Design Rules for Safe Composable Agents - policy rules
Custom Agent Runtimes Best Practices - section checklist

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Plugin Architecture for Tools in a Custom Runtime

Search Documentation

Plugin Architecture for Tools in a Custom Runtime