10. Plugin API

Chapter 10 of 24 · 15 min

KEY INSIGHT

The Plugin API exposes agent capabilities to plugins through a well-defined interface, enabling rich integration without coupling plugins to internal implementation details. Plugins need access to agent capabilities to be useful. The Plugin API provides this access through a stable interface. Changes to internal implementation do not break plugins as long as the API remains compatible. Core API Components The API provides access to several agent subsystems. The tool registry allows plugins to register new tools. The memory system enables plugins to store and retrieve information. The event bus lets plugins subscribe to agent events. The configuration interface provides access to settings. ```python class AgentAPI: """Public API available to plugins.""" def __init__(self, agent: "OpenCLawAgent"): self._agent = agent @property def tools(self) -> "ToolRegistry": """Access the tool registry to register or lookup tools.""" return self._agent._tool_registry @property def memory(self) -> "MemorySystem": """Access the memory system for storage and retrieval.""" return self._agent._memory @property def events(self) -> "EventBus": """Subscribe to agent events.""" return self._agent._event_bus @property def config(self) -> "Config": """Access agent configuration.""" return self._agent._config async def send_message(self, content: str, role: str = "user") -> str: """Send a message to the agent and get the response.""" response = await self._agent.process_message(content, role) return response.content def get_user_context(self) -> dict: """Retrieve known information about the user.""" return self._agent._user_profile.snapshot() ``` Event System Plugins often need to react to agent events. The event system provides publish-subscribe messaging. Plugins subscribe to specific event types and receive notifications when those events occur. ```python class EventBus: def __init__(self): self._subscribers: dict[str, list[callable]] = {} def subscribe(self, event_type: str, handler: callable) -> None: """Subscribe to an event type.""" if event_type not in self._subscribers: self._subscribers[event_type] = [] self._subscribers[event_type].append(handler) async def publish(self, event_type: str, data: dict) -> None: """Publish an event to all subscribers.""" handlers = self._subscribers.get(event_type, []) for handler in handlers: try: if asyncio.iscoroutinefunction(handler): await handler(data) else: handler(data) except Exception as e: logging.error(f"Event handler error: {e}") # Example event types EVENT_MESSAGE_RECEIVED = "message.received" EVENT_TOOL_EXECUTED = "tool.executed" EVENT_MEMORY_CONSOLIDATED = "memory.consolidated" EVENT_USER_PRESENT = "user.present" ```

EXERCISE

Write a plugin that monitors the EVENT_MESSAGE_RECEIVED event, extracts action items from messages using a simple heuristic, and stores them in the memory system for later review.