Building Agent Plugins with OpenAI Agents SDK: Extensible Tool Architecture

Why Plugins Matter for Agent Systems

As your agent system grows, you will face a familiar software engineering problem: the monolith. All tools defined in one file. All logic coupled together. Every new capability requires modifying core agent code.

A plugin architecture solves this by letting you add, remove, and update agent tools without touching the core system. Third-party developers can contribute capabilities. Teams can work independently on different tool sets.

Defining the Plugin Interface

Start with a base class that every plugin must implement.

from abc import ABC, abstractmethod
from agents import FunctionTool, function_tool
from dataclasses import dataclass
from typing import Any


@dataclass
class PluginMetadata:
    name: str
    version: str
    description: str
    author: str


class AgentPlugin(ABC):
    """Base class for all agent plugins."""

    @abstractmethod
    def metadata(self) -> PluginMetadata:
        """Return plugin metadata."""
        ...

    @abstractmethod
    def get_tools(self) -> list[FunctionTool]:
        """Return the tools this plugin provides."""
        ...

    def on_load(self) -> None:
        """Called when the plugin is loaded. Override for setup logic."""
        pass

    def on_unload(self) -> None:
        """Called when the plugin is unloaded. Override for cleanup."""
        pass

Implementing a Concrete Plugin

Here is a weather plugin that provides two tools — current weather and forecast.

import httpx
from agents import function_tool


class WeatherPlugin(AgentPlugin):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client: httpx.AsyncClient | None = None

    def metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="weather",
            version="1.2.0",
            description="Current weather and forecasts",
            author="internal-team",
        )

    def on_load(self) -> None:
        self.client = httpx.AsyncClient(
            base_url="https://api.weatherapi.com/v1",
            params={"key": self.api_key},
            timeout=10.0,
        )

    def on_unload(self) -> None:
        if self.client:
            import asyncio
            asyncio.get_event_loop().run_until_complete(self.client.aclose())

    def get_tools(self) -> list:
        @function_tool
        async def get_current_weather(location: str) -> str:
            """Get current weather for a location."""
            resp = await self.client.get("/current.json", params={"q": location})
            data = resp.json()
            current = data["current"]
            return f"{current['temp_c']}C, {current['condition']['text']} in {location}"

        @function_tool
        async def get_forecast(location: str, days: int = 3) -> str:
            """Get weather forecast for a location."""
            resp = await self.client.get("/forecast.json", params={"q": location, "days": days})
            data = resp.json()
            forecasts = []
            for day in data["forecast"]["forecastday"]:
                forecasts.append(f"{day['date']}: {day['day']['condition']['text']}, {day['day']['avgtemp_c']}C")
            return "\n".join(forecasts)

        return [get_current_weather, get_forecast]

Building the Plugin Registry

The registry manages plugin lifecycle — discovery, loading, and tool aggregation.

import importlib
import os
from pathlib import Path


class PluginRegistry:
    def __init__(self):
        self._plugins: dict[str, AgentPlugin] = {}

    def register(self, plugin: AgentPlugin) -> None:
        meta = plugin.metadata()
        if meta.name in self._plugins:
            self.unregister(meta.name)
        plugin.on_load()
        self._plugins[meta.name] = plugin
        print(f"Loaded plugin: {meta.name} v{meta.version}")

    def unregister(self, name: str) -> None:
        if name in self._plugins:
            self._plugins[name].on_unload()
            del self._plugins[name]
            print(f"Unloaded plugin: {name}")

    def get_all_tools(self) -> list:
        tools = []
        for plugin in self._plugins.values():
            tools.extend(plugin.get_tools())
        return tools

    def list_plugins(self) -> list[PluginMetadata]:
        return [p.metadata() for p in self._plugins.values()]

    def load_from_directory(self, plugin_dir: str) -> None:
        """Auto-discover and load plugins from a directory."""
        for file_path in Path(plugin_dir).glob("*.py"):
            if file_path.name.startswith("_"):
                continue
            module_name = file_path.stem
            spec = importlib.util.spec_from_file_location(module_name, file_path)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            # Find all AgentPlugin subclasses in the module
            for attr_name in dir(module):
                attr = getattr(module, attr_name)
                if isinstance(attr, type) and issubclass(attr, AgentPlugin) and attr is not AgentPlugin:
                    instance = attr()
                    self.register(instance)

Wiring Plugins into an Agent

from agents import Agent, Runner
import asyncio

registry = PluginRegistry()
registry.register(WeatherPlugin(api_key=os.environ["WEATHER_API_KEY"]))

# Dynamically build agent with all plugin tools
agent = Agent(
    name="plugin_powered_assistant",
    instructions="You are a helpful assistant. Use your tools to answer questions.",
    tools=registry.get_all_tools(),
)

async def main():
    result = await Runner.run(agent, input="What is the weather in Tokyo?")
    print(result.final_output)

asyncio.run(main())

Hot-Reloading Plugins in Development

For development, you can watch the plugin directory and reload when files change.

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class PluginReloader(FileSystemEventHandler):
    def __init__(self, registry: PluginRegistry, plugin_dir: str):
        self.registry = registry
        self.plugin_dir = plugin_dir

    def on_modified(self, event):
        if event.src_path.endswith(".py"):
            print(f"Plugin changed: {event.src_path}, reloading...")
            self.registry.load_from_directory(self.plugin_dir)

def start_watcher(registry: PluginRegistry, plugin_dir: str):
    observer = Observer()
    observer.schedule(PluginReloader(registry, plugin_dir), plugin_dir)
    observer.start()
    return observer

FAQ

How do I isolate plugins so a buggy one does not crash the whole system?

Wrap each plugin's get_tools and lifecycle methods in try/except blocks within the registry. If a plugin raises an exception during loading, log the error and skip it. For tool execution, the SDK's runner already handles tool errors gracefully — a failed tool call returns an error message to the agent rather than crashing the process.

Can plugins define their own guardrails?

Yes. Extend the AgentPlugin base class with a get_guardrails method that returns a list of guardrail instances. In the registry, aggregate guardrails alongside tools and pass both to the agent constructor.

How do I version plugins for backward compatibility?

Use semantic versioning in the PluginMetadata. The registry can enforce version constraints — for example, only loading plugins with a major version matching the host system. Store version requirements in a manifest file alongside the plugin directory.

---

#OpenAIAgentsSDK #Plugins #ToolArchitecture #Extensibility #Python #SoftwareDesign #AgenticAI #LearnAI #AIEngineering