haive.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”}

)

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 haive.mcp.manager.MCPBulkInstaller(/, **data)

Bases: pydantic.BaseModel

Handles bulk installation of MCP servers with progress tracking.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

async bulk_install_servers(server_packages, operation_id=None)

Install multiple MCP servers in parallel with progress tracking.

Parameters:

• server_packages (list[str])

• operation_id (str | None)

Return type:

MCPBulkOperation

get_operation_status(operation_id)

Get status of a bulk operation.

Parameters:

operation_id (str)

Return type:

MCPBulkOperation | None

class haive.mcp.manager.MCPBulkOperation(/, **data)

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.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

property progress_percentage: float

Get completion percentage.

Return type:

float

property success_rate: float

Get success rate percentage.

Return type:

float

class haive.mcp.manager.MCPHealthStatus(/, **data)

Bases: pydantic.BaseModel

Health status information for an MCP server.

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

Parameters:

data (Any)

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:

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

class haive.mcp.manager.MCPManager(/, **data)

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

Parameters:

data (Any)

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

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

add_custom_category(category)

Add a custom server category.

Parameters:

category (MCPServerCategory)

Return type:

None

async add_server(server_name, config, connect_immediately=True)

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 (str) – Unique name for the server

• config (haive.mcp.config.MCPServerConfig) – Complete server configuration

• connect_immediately (bool) – Whether to attempt connection immediately

Returns:

Result of the registration attempt

Return type:

MCPRegistrationResult

Examples

Adding a filesystem server:

async bulk_health_check()

Perform health check on all connected servers.

Returns:

Dict with health status for all servers

Return type:

dict[str, Any]

async bulk_install_category(category_name, max_concurrent=5)

Install all servers in a category.

Parameters:

• category_name (str) – Name of the category to install

• max_concurrent (int) – Maximum concurrent installations

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async bulk_install_servers(server_packages, add_to_manager=True, max_concurrent=5)

Install multiple MCP servers in parallel.

Parameters:

• server_packages (list[str]) – List of npm package names to install

• add_to_manager (bool) – Whether to add installed servers to the manager

• max_concurrent (int) – Maximum concurrent installations

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async bulk_remove_servers(server_names)

Remove multiple servers from the manager.

Parameters:

server_names (list[str]) – List of server names to remove

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async bulk_update_servers()

Update all installed MCP servers to their latest versions.

Returns:

Operation tracking object

Return type:

MCPBulkOperation

async call_tool(tool_name, arguments)

Call a tool from any connected server.

Parameters:

• tool_name (str) – Name of the tool to call

• arguments (dict[str, Any]) – Arguments for the tool

Returns:

Result of the tool call

Return type:

Any

get_all_server_status()

Get status information for all servers.

Returns:

Status information for all servers

Return type:

Dict[str, Dict[str, Any]]

async get_all_tools(refresh=False)

Get all tools from all connected servers.

Parameters:

refresh (bool) – Whether to refresh the tool list from servers

Returns:

List of all available tools

Return type:

List[Any]

get_available_categories()

Get all available server categories for bulk operations.

Return type:

dict[str, MCPServerCategory]

get_bulk_operation_status(operation_id)

Get the status of a bulk operation.

Parameters:

operation_id (str)

Return type:

MCPBulkOperation | None

async get_prompts(server_name=None)

Get available prompts from MCP servers.

Parameters:

server_name (str | None) – Optional specific server to query, otherwise gets from all

Returns:

List of available prompts

Return type:

list[Any]

async get_resources(server_name=None)

Get available resources from MCP servers.

Parameters:

server_name (str | None) – Optional specific server to query, otherwise gets from all

Returns:

List of available resources

Return type:

list[Any]

get_server_status(server_name)

Get the status of a specific server.

Parameters:

server_name (str) – Name of the server

Returns:

Server status or None if not found

Return type:

Optional[MCPServerStatus]

model_post_init(__context)

Initialize the MCP manager after model creation.

Return type:

None

async refresh_tools()

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.

Return type:

None

async reload_server(server_name)

Reload a specific MCP server.

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

Parameters:

server_name (str) – Name of the server to reload

Returns:

Result of the reload operation

Return type:

MCPRegistrationResult

async remove_server(server_name)

Remove an MCP server from the manager.

Parameters:

server_name (str) – Name of the server to remove

Returns:

True if server was removed successfully

Return type:

bool

async retry_failed_servers()

Retry connection to all failed servers.

Returns:

Results of retry attempts

Return type:

List[MCPRegistrationResult]

async shutdown()

Shutdown the MCP manager and close all connections.

Return type:

None

class haive.mcp.manager.MCPRegistrationResult(/, **data)

Bases: pydantic.BaseModel

Result of MCP server registration.

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

Parameters:

data (Any)

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

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

class haive.mcp.manager.MCPServerCategory(/, **data)

Bases: pydantic.BaseModel

Represents a category of MCP servers for bulk operations.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class haive.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

Initialize self. See help(type(self)) for accurate signature.