dynamic_activation_mcp

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

DynamicActivationMCPServer

MCP server with dynamic tool activation capabilities.

DynamicMCPRegistry

MCP-specific registry for dynamic tool activation.

DynamicMCPState

MCP-specific state for dynamic activation.

MCPTool

MCP tool representation for dynamic activation.

Module Contents

class dynamic_activation_mcp.DynamicActivationMCPServer(/, **data: Any)

Bases: 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

Parameters:

  • name – Server name

  • discovery_source – Source for tool discovery

  • tool_registry – Registry for managing tools

  • discovery_config – Configuration for discovery

  • meta_state – MetaStateSchema for tracking

  • state – DynamicMCPState for session management

Examples

Create MCP server:

With custom discovery:

server = DynamicActivationMCPServer(
name="custom_mcp_server",
discovery_source="/path/to/tools",
discovery_config={
    "auto_discover": True,
    "max_tools": 50,
    "cache_ttl": 3600
}

)

get_available_tools() → list[dict[str, Any]]

Get list of available tools for MCP clients.

Returns:

List of tool descriptions for MCP protocol

Examples

Get tools for MCP registration:

get_server_stats() → dict[str, Any]

Get server statistics.

Returns:

Dictionary with server statistics

Examples

Get server status:

async handle_client_connect(client_id: str, client_info: dict[str, Any]) → dict[str, Any]

Handle MCP client connection.

Parameters:

  • client_id – ID of connecting client

  • client_info – Client information

Returns:

Connection response

Examples

Handle client connection:

async handle_client_disconnect(client_id: str) → None

Handle MCP client disconnection.

Parameters:

client_id – ID of disconnecting client

Examples

Handle client disconnect:

async handle_tool_request(request: dict[str, Any]) → dict[str, Any]

Handle dynamic tool requests from MCP clients.

Parameters:

request – MCP tool request

Returns:

Tool execution result

Examples

Handle tool request:

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

async start() → None

Start the MCP server.

Examples

Start server:

async stop() → None

Stop the MCP server.

Examples

Stop server:

discovery_config: dict[str, Any] = None
discovery_source: str = None
meta_state: haive.core.schema.prebuilt.meta_state.MetaStateSchema | None = None
model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str = None
state: DynamicMCPState = None
tool_registry: DynamicMCPRegistry = None
class dynamic_activation_mcp.DynamicMCPRegistry

Bases: haive.core.registry.DynamicRegistry[MCPTool]

MCP-specific registry for dynamic tool activation.

Extends DynamicRegistry with MCP-specific functionality for tool registration and activation with MCP servers.

Examples

Create MCP registry:

async activate_mcp_tool(tool_id: str, mcp_server: Any | None = None) → MCPTool | None

Activate an MCP tool and register with server.

Parameters:

  • tool_id – ID of tool to activate

  • mcp_server – MCP server to register with (optional)

Returns:

Activated MCPTool or None if activation failed

Examples

Activate MCP tool:

get_tool_schemas() → dict[str, dict[str, Any]]

Get input schemas for all registered tools.

Returns:

Dictionary of tool name to input schema

Examples

Get schemas for MCP registration:

class dynamic_activation_mcp.DynamicMCPState(/, **data: Any)

Bases: 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.

Parameters:

  • mcp_client_id – ID of connected MCP client

  • mcp_session_id – Current MCP session ID

  • mcp_protocol_version – MCP protocol version

  • tool_call_history – History of MCP tool calls

Examples

Create MCP state:

track_tool_call(tool_name: str, input_data: dict[str, Any], result: Any) → None

Track an MCP tool call.

Parameters:

  • tool_name – Name of tool called

  • input_data – Input data for tool

  • result – Result from tool execution

Examples

Track tool call:

mcp_client_id: str | None = None
mcp_protocol_version: str = None
mcp_session_id: str | None = None
tool_call_history: list[dict[str, Any]] = None
class dynamic_activation_mcp.MCPTool(/, **data: Any)

Bases: pydantic.BaseModel

MCP tool representation for dynamic activation.

Represents a tool that can be activated and used via MCP protocol.

Parameters:

  • name – Tool name

  • description – Tool description

  • input_schema – JSON schema for tool input

  • handler – Function to handle tool execution

  • metadata – Additional metadata

Examples

Create MCP tool:

description: str = None
handler: Any = None
input_schema: dict[str, Any] = None
metadata: dict[str, Any] = None
model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str = None