mcp.manager¶

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

Examples

Adding MCP servers procedurally:

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:

# Check server health
health = await manager.check_server_health(“filesystem”) if health.status == MCPServerStatus.UNHEALTHY:

await manager.reconnect_server(“filesystem”)

Tool execution example:

# Execute a tool on specific server
result = await manager.execute_tool(
server=”filesystem”, tool=”read_file”, params={“path”: “/path/to/file.txt”}

)

Attributes¶

`MCP_AVAILABLE`
`logger`

Classes¶

`MCPBulkInstaller`	Handles bulk installation of MCP servers with progress tracking.
`MCPBulkOperation`	Represents a bulk operation on multiple MCP servers.
`MCPHealthStatus`	Health status information for an MCP server.
`MCPManager`	Dynamic MCP manager for procedural server addition.
`MCPRegistrationResult`	Result of MCP server registration.
`MCPServerCategory`	Represents a category of MCP servers for bulk operations.
`MCPServerStatus`	Status of an MCP server.

Module Contents¶

class mcp.manager.MCPBulkInstaller(/, **data: Any)¶

Bases: pydantic.BaseModel

Handles bulk installation of MCP servers with progress tracking.

async bulk_install_servers(server_packages: list[str], operation_id: str | None = None) → MCPBulkOperation¶: Install multiple MCP servers in parallel with progress tracking.

get_operation_status(operation_id: str) → MCPBulkOperation | None¶: Get status of a bulk operation.

max_concurrent: int = None¶

retry_attempts: int = None¶

show_progress: bool = None¶

timeout_per_server: float = None¶

class mcp.manager.MCPBulkOperation(/, **data: Any)¶

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

completed_at: datetime.datetime | None = None¶

completed_count: int = None¶

current_server: str | None = None¶

failed_count: int = None¶

failed_servers: list[dict] = None¶

is_complete: bool = None¶

operation_id: str = None¶

operation_type: str = None¶

property progress_percentage: float¶: Get completion percentage.

server_names: list[str] = None¶

started_at: datetime.datetime = None¶

succeeded_servers: list[str] = None¶

success_count: int = None¶

property success_rate: float¶: Get success rate percentage.

total_count: int = None¶

class mcp.manager.MCPHealthStatus(/, **data: Any)¶

Bases: pydantic.BaseModel

Health status information for an MCP server.

Tracks the health and performance metrics of an individual MCP server connection.

server_name¶: Name of the server being monitored

status¶: Current operational status

last_check¶: Timestamp of the most recent health check

response_time¶: Latest response time in seconds (None if failed)

consecutive_failures¶: Count of consecutive failed health checks

total_requests¶: Total number of requests made to this server

successful_requests¶: Number of successful requests

error_details¶: Details of the most recent error (if any)

Examples

Health status after monitoring:

consecutive_failures: int = None¶

error_details: str | None = None¶

last_check: datetime.datetime = None¶

response_time: float | None = None¶

server_name: str = None¶

status: MCPServerStatus = None¶

successful_requests: int = None¶

total_requests: int = None¶

class mcp.manager.MCPManager(/, **data: Any)¶

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

enabled¶: Whether MCP management is enabled

auto_health_check¶: Whether to automatically monitor server health

health_check_interval¶: Interval between health checks in seconds

max_retry_attempts¶: Maximum retry attempts for failed connections

connection_timeout¶: Timeout for server connections in seconds

add_custom_category(category: MCPServerCategory) → None¶: Add a custom server category.

async add_server(server_name: str, config: haive.mcp.config.MCPServerConfig, connect_immediately: bool = True) → MCPRegistrationResult¶

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.

Parameters:

server_name – Unique name for the server
config – Complete server configuration
connect_immediately – Whether to attempt connection immediately

Returns:

Result of the registration attempt

Return type:

MCPRegistrationResult

Examples

Adding a filesystem server:

async bulk_health_check() → dict[str, Any]¶

Perform health check on all connected servers.

Returns:: Dict with health status for all servers

async bulk_install_category(category_name: str, max_concurrent: int = 5) → MCPBulkOperation¶

Install all servers in a category.

Parameters:

category_name – Name of the category to install
max_concurrent – Maximum concurrent installations

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async bulk_install_servers(server_packages: list[str], add_to_manager: bool = True, max_concurrent: int = 5) → MCPBulkOperation¶

Install multiple MCP servers in parallel.

Parameters:

server_packages – List of npm package names to install
add_to_manager – Whether to add installed servers to the manager
max_concurrent – Maximum concurrent installations

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async bulk_remove_servers(server_names: list[str]) → MCPBulkOperation¶

Remove multiple servers from the manager.

Parameters:: server_names – List of server names to remove
Returns:: Operation tracking object
Return type:: MCPBulkOperation

async bulk_update_servers() → MCPBulkOperation¶

Update all installed MCP servers to their latest versions.

Returns:: Operation tracking object
Return type:: MCPBulkOperation

async call_tool(tool_name: str, arguments: dict[str, Any]) → Any¶

Call a tool from any connected server.

Parameters:

tool_name – Name of the tool to call
arguments – Arguments for the tool

Returns:

Result of the tool call

Return type:

Any

get_all_server_status() → dict[str, dict[str, Any]]¶

Get status information for all servers.

Returns:: Status information for all servers
Return type:: Dict[str, Dict[str, Any]]

async get_all_tools(refresh: bool = False) → list[Any]¶

Get all tools from all connected servers.

Parameters:: refresh – Whether to refresh the tool list from servers
Returns:: List of all available tools
Return type:: List[Any]

get_available_categories() → dict[str, MCPServerCategory]¶: Get all available server categories for bulk operations.

get_bulk_operation_status(operation_id: str) → MCPBulkOperation | None¶: Get the status of a bulk operation.

async get_prompts(server_name: str | None = None) → list[Any]¶

Get available prompts from MCP servers.

Parameters:: server_name – Optional specific server to query, otherwise gets from all
Returns:: List of available prompts

async get_resources(server_name: str | None = None) → list[Any]¶

Get available resources from MCP servers.

Parameters:: server_name – Optional specific server to query, otherwise gets from all
Returns:: List of available resources

get_server_status(server_name: str) → MCPServerStatus | None¶

Get the status of a specific server.

Parameters:: server_name – Name of the server
Returns:: Server status or None if not found
Return type:: Optional[MCPServerStatus]

model_post_init(__context) → None¶: Initialize the MCP manager after model creation.

async refresh_tools() → None¶

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.

async reload_server(server_name: str) → MCPRegistrationResult¶

Reload a specific MCP server.

Disconnects and reconnects to the server, refreshing all tools, resources, and prompts.

Parameters:: server_name – Name of the server to reload
Returns:: Result of the reload operation
Return type:: MCPRegistrationResult

async remove_server(server_name: str) → bool¶

Remove an MCP server from the manager.

Parameters:: server_name – Name of the server to remove
Returns:: True if server was removed successfully
Return type:: bool

async retry_failed_servers() → list[MCPRegistrationResult]¶

Retry connection to all failed servers.

Returns:: Results of retry attempts
Return type:: List[MCPRegistrationResult]

async shutdown() → None¶: Shutdown the MCP manager and close all connections.

auto_health_check: bool = None¶

connection_timeout: float = None¶

enabled: bool = None¶

health_check_interval: float = None¶

max_retry_attempts: int = None¶

class mcp.manager.MCPRegistrationResult(/, **data: Any)¶

Bases: pydantic.BaseModel

Result of MCP server registration.

Contains the outcome of attempting to register and connect to an MCP server.

server_name¶: Name of the server that was registered

success¶: Whether registration and connection succeeded

status¶: Current status of the server connection

error¶: Optional error message if registration failed

tools_discovered¶: Number of tools discovered from this server

resources_discovered¶: Number of resources discovered from this server

connection_time¶: Time taken to establish connection in seconds

connection_time: float | None = None¶

error_message: str | None = None¶

server_name: str = None¶

status: MCPServerStatus = None¶

success: bool = None¶

tools: list[str] = None¶

tools_count: int = None¶

class mcp.manager.MCPServerCategory(/, **data: Any)¶

Bases: pydantic.BaseModel

Represents a category of MCP servers for bulk operations.

description: str = None¶

name: str = None¶

popularity_threshold: int | None = None¶

servers: list[str] = None¶

tags: list[str] = None¶

class mcp.manager.MCPServerStatus¶

Bases: str, enum.Enum

Status of an MCP server.

PENDING¶: Not yet attempted to connect

CONNECTING¶: Connection in progress

CONNECTED¶: Successfully connected and operational

FAILED¶: Connection failed with error

DISCONNECTED¶: Intentionally disconnected by user

UNHEALTHY¶: Connected but health check failed

CONNECTED = 'connected'¶

CONNECTING = 'connecting'¶

DISCONNECTED = 'disconnected'¶

FAILED = 'failed'¶

PENDING = 'pending'¶

UNHEALTHY = 'unhealthy'¶

mcp.manager.MCP_AVAILABLE = True¶

mcp.manager.logger¶