mcp.manager =========== .. py:module:: mcp.manager .. autoapi-nested-parse:: Dynamic MCP Manager for procedural server addition and bulk operations. This module provides a comprehensive system for adding MCP servers procedurally, one by one, during runtime, plus industrial-strength bulk operations for managing hundreds of servers from the 1900+ available MCP server ecosystem. The manager enables: - Step-by-step MCP server addition - Bulk installation and management operations - Runtime configuration updates - Health monitoring and retry logic - Incremental capability discovery - Safe server removal and replacement - Progress tracking for bulk operations - Category-based server management Classes: MCPManager: Main manager for dynamic MCP operations MCPRegistrationResult: Result of server registration MCPHealthStatus: Health monitoring information .. rubric:: Examples Adding MCP servers procedurally: .. code-block:: python from haive.mcp.manager import MCPManager from haive.mcp.config import MCPServerConfig # Create manager manager = MCPManager() # Add servers one by one await manager.add_server("filesystem", MCPServerConfig( name="filesystem", transport="stdio", command="npx", args=["-y", "@modelcontextprotocol/server-filesystem"] )) await manager.add_server("github", MCPServerConfig( name="github", transport="stdio", command="npx", args=["-y", "@modelcontextprotocol/server-github"], env={"GITHUB_TOKEN": "your_token"} )) # Get all available tools tools = await manager.get_all_tools() Health monitoring example: .. code-block:: python # Check server health health = await manager.check_server_health("filesystem") if health.status == MCPServerStatus.UNHEALTHY: await manager.reconnect_server("filesystem") Tool execution example: .. code-block:: python # Execute a tool on specific server result = await manager.execute_tool( server="filesystem", tool="read_file", params={"path": "/path/to/file.txt"} ) Attributes ---------- .. autoapisummary:: mcp.manager.MCP_AVAILABLE mcp.manager.logger Classes ------- .. autoapisummary:: mcp.manager.MCPBulkInstaller mcp.manager.MCPBulkOperation mcp.manager.MCPHealthStatus mcp.manager.MCPManager mcp.manager.MCPRegistrationResult mcp.manager.MCPServerCategory mcp.manager.MCPServerStatus Module Contents --------------- .. py:class:: MCPBulkInstaller(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Handles bulk installation of MCP servers with progress tracking. .. py:method:: bulk_install_servers(server_packages: list[str], operation_id: str | None = None) -> MCPBulkOperation :async: Install multiple MCP servers in parallel with progress tracking. .. py:method:: get_operation_status(operation_id: str) -> MCPBulkOperation | None Get status of a bulk operation. .. py:attribute:: max_concurrent :type: int :value: None .. py:attribute:: retry_attempts :type: int :value: None .. py:attribute:: show_progress :type: bool :value: None .. py:attribute:: timeout_per_server :type: float :value: None .. py:class:: MCPBulkOperation(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Represents a bulk operation on multiple MCP servers. Tracks progress, results, and errors for bulk installation, removal, or update operations across multiple servers. .. py:attribute:: completed_at :type: datetime.datetime | None :value: None .. py:attribute:: completed_count :type: int :value: None .. py:attribute:: current_server :type: str | None :value: None .. py:attribute:: failed_count :type: int :value: None .. py:attribute:: failed_servers :type: list[dict] :value: None .. py:attribute:: is_complete :type: bool :value: None .. py:attribute:: operation_id :type: str :value: None .. py:attribute:: operation_type :type: str :value: None .. py:property:: progress_percentage :type: float Get completion percentage. .. py:attribute:: server_names :type: list[str] :value: None .. py:attribute:: started_at :type: datetime.datetime :value: None .. py:attribute:: succeeded_servers :type: list[str] :value: None .. py:attribute:: success_count :type: int :value: None .. py:property:: success_rate :type: float Get success rate percentage. .. py:attribute:: total_count :type: int :value: None .. py:class:: MCPHealthStatus(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Health status information for an MCP server. Tracks the health and performance metrics of an individual MCP server connection. .. attribute:: server_name Name of the server being monitored .. attribute:: status Current operational status .. attribute:: last_check Timestamp of the most recent health check .. attribute:: response_time Latest response time in seconds (None if failed) .. attribute:: consecutive_failures Count of consecutive failed health checks .. attribute:: total_requests Total number of requests made to this server .. attribute:: successful_requests Number of successful requests .. attribute:: error_details Details of the most recent error (if any) .. rubric:: Examples Health status after monitoring: .. code-block:: python status = MCPHealthStatus( server_name="filesystem", status=MCPServerStatus.CONNECTED, last_check=datetime.now(), response_time=0.125, consecutive_failures=0, total_requests=1000, successful_requests=998 ) .. py:attribute:: consecutive_failures :type: int :value: None .. py:attribute:: error_details :type: str | None :value: None .. py:attribute:: last_check :type: datetime.datetime :value: None .. py:attribute:: response_time :type: float | None :value: None .. py:attribute:: server_name :type: str :value: None .. py:attribute:: status :type: MCPServerStatus :value: None .. py:attribute:: successful_requests :type: int :value: None .. py:attribute:: total_requests :type: int :value: None .. py:class:: MCPManager(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Dynamic MCP manager for procedural server addition. Provides a high-level interface for managing MCP servers during runtime, allowing for incremental addition, health monitoring, and dynamic configuration updates without disrupting existing connections. The manager maintains: - Individual server configurations and status - Health monitoring for each server - Consolidated tool registry from all servers - Connection pooling and retry logic - Event callbacks for server state changes .. attribute:: enabled Whether MCP management is enabled .. attribute:: auto_health_check Whether to automatically monitor server health .. attribute:: health_check_interval Interval between health checks in seconds .. attribute:: max_retry_attempts Maximum retry attempts for failed connections .. attribute:: connection_timeout Timeout for server connections in seconds .. py:method:: add_custom_category(category: MCPServerCategory) -> None Add a custom server category. .. py:method:: add_server(server_name: str, config: haive.mcp.config.MCPServerConfig, connect_immediately: bool = True) -> MCPRegistrationResult :async: Add a new MCP server procedurally. Adds a single MCP server to the manager with optional immediate connection. This allows for step-by-step server addition during runtime without disrupting existing connections. :param server_name: Unique name for the server :param config: Complete server configuration :param connect_immediately: Whether to attempt connection immediately :returns: Result of the registration attempt :rtype: MCPRegistrationResult .. rubric:: Examples Adding a filesystem server: .. code-block:: python result = await manager.add_server("filesystem", MCPServerConfig( name="filesystem", transport="stdio", command="npx", args=["-y", "@modelcontextprotocol/server-filesystem"] )) if result.success: print(f"Added {result.tools_count} tools") .. py:method:: bulk_health_check() -> dict[str, Any] :async: Perform health check on all connected servers. :returns: Dict with health status for all servers .. py:method:: bulk_install_category(category_name: str, max_concurrent: int = 5) -> MCPBulkOperation :async: Install all servers in a category. :param category_name: Name of the category to install :param max_concurrent: Maximum concurrent installations :returns: Operation tracking object :rtype: MCPBulkOperation .. py:method:: bulk_install_servers(server_packages: list[str], add_to_manager: bool = True, max_concurrent: int = 5) -> MCPBulkOperation :async: Install multiple MCP servers in parallel. :param server_packages: List of npm package names to install :param add_to_manager: Whether to add installed servers to the manager :param max_concurrent: Maximum concurrent installations :returns: Operation tracking object :rtype: MCPBulkOperation .. py:method:: bulk_remove_servers(server_names: list[str]) -> MCPBulkOperation :async: Remove multiple servers from the manager. :param server_names: List of server names to remove :returns: Operation tracking object :rtype: MCPBulkOperation .. py:method:: bulk_update_servers() -> MCPBulkOperation :async: Update all installed MCP servers to their latest versions. :returns: Operation tracking object :rtype: MCPBulkOperation .. py:method:: call_tool(tool_name: str, arguments: dict[str, Any]) -> Any :async: Call a tool from any connected server. :param tool_name: Name of the tool to call :param arguments: Arguments for the tool :returns: Result of the tool call :rtype: Any .. py:method:: get_all_server_status() -> dict[str, dict[str, Any]] Get status information for all servers. :returns: Status information for all servers :rtype: Dict[str, Dict[str, Any]] .. py:method:: get_all_tools(refresh: bool = False) -> list[Any] :async: Get all tools from all connected servers. :param refresh: Whether to refresh the tool list from servers :returns: List of all available tools :rtype: List[Any] .. py:method:: get_available_categories() -> dict[str, MCPServerCategory] Get all available server categories for bulk operations. .. py:method:: get_bulk_operation_status(operation_id: str) -> MCPBulkOperation | None Get the status of a bulk operation. .. py:method:: get_prompts(server_name: str | None = None) -> list[Any] :async: Get available prompts from MCP servers. :param server_name: Optional specific server to query, otherwise gets from all :returns: List of available prompts .. py:method:: get_resources(server_name: str | None = None) -> list[Any] :async: Get available resources from MCP servers. :param server_name: Optional specific server to query, otherwise gets from all :returns: List of available resources .. py:method:: get_server_status(server_name: str) -> MCPServerStatus | None Get the status of a specific server. :param server_name: Name of the server :returns: Server status or None if not found :rtype: Optional[MCPServerStatus] .. py:method:: model_post_init(__context) -> None Initialize the MCP manager after model creation. .. py:method:: refresh_tools() -> None :async: Refresh the tool list from all connected servers. This method rebuilds the multi-client and forces a fresh discovery of all tools from connected servers. Call this after adding new servers or when tools may have changed. .. py:method:: reload_server(server_name: str) -> MCPRegistrationResult :async: Reload a specific MCP server. Disconnects and reconnects to the server, refreshing all tools, resources, and prompts. :param server_name: Name of the server to reload :returns: Result of the reload operation :rtype: MCPRegistrationResult .. py:method:: remove_server(server_name: str) -> bool :async: Remove an MCP server from the manager. :param server_name: Name of the server to remove :returns: True if server was removed successfully :rtype: bool .. py:method:: retry_failed_servers() -> list[MCPRegistrationResult] :async: Retry connection to all failed servers. :returns: Results of retry attempts :rtype: List[MCPRegistrationResult] .. py:method:: shutdown() -> None :async: Shutdown the MCP manager and close all connections. .. py:attribute:: auto_health_check :type: bool :value: None .. py:attribute:: connection_timeout :type: float :value: None .. py:attribute:: enabled :type: bool :value: None .. py:attribute:: health_check_interval :type: float :value: None .. py:attribute:: max_retry_attempts :type: int :value: None .. py:class:: MCPRegistrationResult(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Result of MCP server registration. Contains the outcome of attempting to register and connect to an MCP server. .. attribute:: server_name Name of the server that was registered .. attribute:: success Whether registration and connection succeeded .. attribute:: status Current status of the server connection .. attribute:: error Optional error message if registration failed .. attribute:: tools_discovered Number of tools discovered from this server .. attribute:: resources_discovered Number of resources discovered from this server .. attribute:: connection_time Time taken to establish connection in seconds .. py:attribute:: connection_time :type: float | None :value: None .. py:attribute:: error_message :type: str | None :value: None .. py:attribute:: server_name :type: str :value: None .. py:attribute:: status :type: MCPServerStatus :value: None .. py:attribute:: success :type: bool :value: None .. py:attribute:: tools :type: list[str] :value: None .. py:attribute:: tools_count :type: int :value: None .. py:class:: MCPServerCategory(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Represents a category of MCP servers for bulk operations. .. py:attribute:: description :type: str :value: None .. py:attribute:: name :type: str :value: None .. py:attribute:: popularity_threshold :type: int | None :value: None .. py:attribute:: servers :type: list[str] :value: None .. py:attribute:: tags :type: list[str] :value: None .. py:class:: MCPServerStatus Bases: :py:obj:`str`, :py:obj:`enum.Enum` Status of an MCP server. .. attribute:: PENDING Not yet attempted to connect .. attribute:: CONNECTING Connection in progress .. attribute:: CONNECTED Successfully connected and operational .. attribute:: FAILED Connection failed with error .. attribute:: DISCONNECTED Intentionally disconnected by user .. attribute:: UNHEALTHY Connected but health check failed .. py:attribute:: CONNECTED :value: 'connected' .. py:attribute:: CONNECTING :value: 'connecting' .. py:attribute:: DISCONNECTED :value: 'disconnected' .. py:attribute:: FAILED :value: 'failed' .. py:attribute:: PENDING :value: 'pending' .. py:attribute:: UNHEALTHY :value: 'unhealthy' .. py:data:: MCP_AVAILABLE :value: True .. py:data:: logger