By Sagar Shankaran, Founder of CallSphere
Learn how to create a plugin system for OpenAI Agents SDK that supports dynamic tool loading, hot-reloading during development, and isolated execution for third-party extensions.
Key takeaways
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.
Start with a base class that every plugin must implement.
Hear it before you finish reading
Talk to a live CallSphere AI voice agent in your browser — 60 seconds, no signup.
flowchart LR
INPUT(["User input"])
AGENT["Agent<br/>name plus instructions"]
HAND{"Handoff to<br/>another agent?"}
SUB["Sub-agent<br/>specialist"]
GUARD{"Guardrail<br/>passed?"}
TOOL["Tool call"]
SDK[("Tracing<br/>OpenAI dashboard")]
OUT(["Final output"])
INPUT --> AGENT --> HAND
HAND -->|Yes| SUB --> GUARD
HAND -->|No| GUARD
GUARD -->|Yes| TOOL --> AGENT
GUARD -->|Block| OUT
AGENT --> OUT
AGENT --> SDK
style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
style SDK fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
style OUT fill:#059669,stroke:#047857,color:#fff
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
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]
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)
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())
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
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.
Still reading? Stop comparing — try CallSphere live.
CallSphere ships complete AI voice agents per industry — 14 tools for healthcare, 10 agents for real estate, 4 specialists for salons. See how it actually handles a call before you book a demo.
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.
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

Written by
Sagar Shankaran· Founder, CallSphere
LinkedInSagar Shankaran is the founder of CallSphere, where he builds production AI voice and chat agents deployed across healthcare, hospitality, real estate, and home services. He writes about agentic AI, LLM engineering, and shipping voice agents that handle real calls in production.
See how AI voice agents work for your industry. Live demo available -- no signup required.
How we built a fault-tolerant HVAC emergency triage and tech-dispatch platform on Kubernetes — three-tier CQRS, 11 micro-agents on the OpenAI Agents SDK + LangGraph, NATS JetStream, DTMF/SMS/WebSocket acceptance, circuit breakers, and an evaluation pipeline that catches regressions before they wake a tech at 3 AM.
How to build a safety eval pipeline that runs known jailbreak corpora, prompt-injection attacks, and tool-misuse scenarios on every release — and gates merges on it.
OpenAI's parallel function calling can cut latency in half — or burn money on dependent calls. The architecture, code, and an eval that proves the win.
Honest principal-engineer comparison of the OpenAI Agents SDK and the legacy Assistants API, with a migration checklist and eval-parity strategy so you don't ship regressions.
Langgraph multi-agent supervisor handoffs docs: the supervisor pattern in LangGraph for coordinating specialist agents, with full code, an eval pipeline that scores routing accuracy, and the failure modes to watch for.
How to stream tokens, tool-call deltas, and intermediate steps from an agent — with code for both the OpenAI Agents SDK and LangChain — and the gotchas that bite in production.
© 2026 CallSphere Inc. All rights reserved.
Made within San Francisco
Watch how CallSphere handles real customer calls, schedules appointments, and processes payments — live.