dynamic_activation_mcp ====================== .. py:module:: dynamic_activation_mcp .. autoapi-nested-parse:: Dynamic Activation MCP Server Implementation. This module provides MCP (Model Context Protocol) integration for the Dynamic Activation Pattern, allowing MCP servers to dynamically discover and activate tools based on client requests. Based on: - @project_docs/active/patterns/dynamic_activation_pattern.md - MCP protocol specification - Existing haive-mcp infrastructure Classes ------- .. autoapisummary:: dynamic_activation_mcp.DynamicActivationMCPServer dynamic_activation_mcp.DynamicMCPRegistry dynamic_activation_mcp.DynamicMCPState dynamic_activation_mcp.MCPTool Module Contents --------------- .. py:class:: DynamicActivationMCPServer(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` MCP server with dynamic tool activation capabilities. This server implements the MCP protocol with dynamic tool discovery and activation. It can discover tools from documentation and activate them based on client requests. Key Features: - Dynamic tool discovery from documentation - Tool activation on demand - MCP protocol compliance - MetaStateSchema integration for tracking - Session management :param name: Server name :param discovery_source: Source for tool discovery :param tool_registry: Registry for managing tools :param discovery_config: Configuration for discovery :param meta_state: MetaStateSchema for tracking :param state: DynamicMCPState for session management .. rubric:: Examples Create MCP server: .. code-block:: python server = DynamicActivationMCPServer( name="dynamic_mcp_server", discovery_source="@haive-tools" ) # Start server await server.start() # Handle client requests result = await server.handle_tool_request({ "tool": "calculator", "input": {"expression": "2 + 2"} }) With custom discovery: .. code-block:: python server = DynamicActivationMCPServer( name="custom_mcp_server", discovery_source="/path/to/tools", discovery_config={ "auto_discover": True, "max_tools": 50, "cache_ttl": 3600 } ) .. py:method:: get_available_tools() -> list[dict[str, Any]] Get list of available tools for MCP clients. :returns: List of tool descriptions for MCP protocol .. rubric:: Examples Get tools for MCP registration: .. code-block:: python tools = server.get_available_tools() for tool in tools: print(f"{tool['name']}: {tool['description']}") .. py:method:: get_server_stats() -> dict[str, Any] Get server statistics. :returns: Dictionary with server statistics .. rubric:: Examples Get server status: .. code-block:: python stats = server.get_server_stats() print(f"Tools: {stats['total_tools']}") print(f"Active: {stats['active_tools']}") print(f"Clients: {stats['connected_clients']}") .. py:method:: handle_client_connect(client_id: str, client_info: dict[str, Any]) -> dict[str, Any] :async: Handle MCP client connection. :param client_id: ID of connecting client :param client_info: Client information :returns: Connection response .. rubric:: Examples Handle client connection: .. code-block:: python response = await server.handle_client_connect("client_123", { "name": "My Client", "version": "1.0" }) .. py:method:: handle_client_disconnect(client_id: str) -> None :async: Handle MCP client disconnection. :param client_id: ID of disconnecting client .. rubric:: Examples Handle client disconnect: .. code-block:: python await server.handle_client_disconnect("client_123") .. py:method:: handle_tool_request(request: dict[str, Any]) -> dict[str, Any] :async: Handle dynamic tool requests from MCP clients. :param request: MCP tool request :returns: Tool execution result .. rubric:: Examples Handle tool request: .. code-block:: python request = { "tool": "calculator", "input": {"expression": "2 + 2"}, "client_id": "client_123" } result = await server.handle_tool_request(request) print(f"Result: {result}") .. py:method:: setup_mcp_server() -> DynamicActivationMCPServer Initialize MCP server components. This validator: 1. Sets up discovery configuration 2. Initializes discovery agent 3. Wraps in MetaStateSchema 4. Configures MCP protocol settings .. py:method:: start() -> None :async: Start the MCP server. .. rubric:: Examples Start server: .. code-block:: python await server.start() print("MCP server started") .. py:method:: stop() -> None :async: Stop the MCP server. .. rubric:: Examples Stop server: .. code-block:: python await server.stop() print("MCP server stopped") .. py:attribute:: discovery_config :type: dict[str, Any] :value: None .. py:attribute:: discovery_source :type: str :value: None .. py:attribute:: meta_state :type: haive.core.schema.prebuilt.meta_state.MetaStateSchema | None :value: None .. py:attribute:: model_config Configuration for the model, should be a dictionary conforming to [`ConfigDict`][pydantic.config.ConfigDict]. .. py:attribute:: name :type: str :value: None .. py:attribute:: state :type: DynamicMCPState :value: None .. py:attribute:: tool_registry :type: DynamicMCPRegistry :value: None .. py:class:: DynamicMCPRegistry Bases: :py:obj:`haive.core.registry.DynamicRegistry`\ [\ :py:obj:`MCPTool`\ ] MCP-specific registry for dynamic tool activation. Extends DynamicRegistry with MCP-specific functionality for tool registration and activation with MCP servers. .. rubric:: Examples Create MCP registry: .. code-block:: python registry = DynamicMCPRegistry() # Register MCP tool tool = MCPTool( name="search", description="Web search", input_schema={"type": "object", "properties": {"query": {"type": "string"}}}, handler=search_handler ) item = RegistryItem( id="search_001", name="Web Search", description="Search the web", component=tool ) registry.register(item) .. py:method:: activate_mcp_tool(tool_id: str, mcp_server: Any | None = None) -> MCPTool | None :async: Activate an MCP tool and register with server. :param tool_id: ID of tool to activate :param mcp_server: MCP server to register with (optional) :returns: Activated MCPTool or None if activation failed .. rubric:: Examples Activate MCP tool: .. code-block:: python tool = await registry.activate_mcp_tool("search_001", server) if tool: print(f"Activated: {tool.name}") .. py:method:: get_tool_schemas() -> dict[str, dict[str, Any]] Get input schemas for all registered tools. :returns: Dictionary of tool name to input schema .. rubric:: Examples Get schemas for MCP registration: .. code-block:: python schemas = registry.get_tool_schemas() for tool_name, schema in schemas.items(): print(f"{tool_name}: {schema}") .. py:class:: DynamicMCPState(/, **data: Any) Bases: :py:obj:`haive.core.schema.prebuilt.dynamic_activation_state.DynamicActivationState` MCP-specific state for dynamic activation. Extends DynamicActivationState with MCP protocol specific fields and functionality for handling MCP client requests. :param mcp_client_id: ID of connected MCP client :param mcp_session_id: Current MCP session ID :param mcp_protocol_version: MCP protocol version :param tool_call_history: History of MCP tool calls .. rubric:: Examples Create MCP state: .. code-block:: python state = DynamicMCPState( mcp_client_id="client_123", mcp_session_id="session_456", mcp_protocol_version="1.0" ) # Track tool calls state.tool_call_history.append({ "tool": "calculator", "input": {"expression": "2 + 2"}, "timestamp": "2025-01-15T10:30:00" }) .. py:method:: track_tool_call(tool_name: str, input_data: dict[str, Any], result: Any) -> None Track an MCP tool call. :param tool_name: Name of tool called :param input_data: Input data for tool :param result: Result from tool execution .. rubric:: Examples Track tool call: .. code-block:: python state.track_tool_call( tool_name="calculator", input_data={"expression": "2 + 2"}, result=4 ) .. py:attribute:: mcp_client_id :type: str | None :value: None .. py:attribute:: mcp_protocol_version :type: str :value: None .. py:attribute:: mcp_session_id :type: str | None :value: None .. py:attribute:: tool_call_history :type: list[dict[str, Any]] :value: None .. py:class:: MCPTool(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` MCP tool representation for dynamic activation. Represents a tool that can be activated and used via MCP protocol. :param name: Tool name :param description: Tool description :param input_schema: JSON schema for tool input :param handler: Function to handle tool execution :param metadata: Additional metadata .. rubric:: Examples Create MCP tool: .. code-block:: python tool = MCPTool( name="calculator", description="Mathematical calculations", input_schema={"type": "object", "properties": {"expression": {"type": "string"}}}, handler=calculator_handler ) .. py:attribute:: description :type: str :value: None .. py:attribute:: handler :type: Any :value: None .. py:attribute:: input_schema :type: dict[str, Any] :value: None .. py:attribute:: metadata :type: dict[str, Any] :value: None .. py:attribute:: model_config Configuration for the model, should be a dictionary conforming to [`ConfigDict`][pydantic.config.ConfigDict]. .. py:attribute:: name :type: str :value: None