haive.mcp.client.connection

MCP Connection Management.

This module provides connection management utilities for MCP clients. It handles connection pooling, health monitoring, and connection lifecycle management for different transport types.

Classes

ConnectionInfo	Information about an MCP connection.
ConnectionStatus	Connection status states.
MCPConnection	MCP connection manager with health monitoring and auto-reconnection.

Module Contents

class haive.mcp.client.connection.ConnectionInfo

Information about an MCP connection.

class haive.mcp.client.connection.ConnectionStatus

Bases: str, enum.Enum

Connection status states.

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

class haive.mcp.client.connection.MCPConnection(name, transport, auto_reconnect=True, max_reconnect_attempts=5, reconnect_delay=1.0, max_reconnect_delay=60.0, health_check_interval=30.0, connection_timeout=30.0)

MCP connection manager with health monitoring and auto-reconnection.

This class manages individual MCP connections with features like:

• Health monitoring and scoring

• Automatic reconnection with backoff

• Connection lifecycle management

• Error tracking and recovery

• Performance metrics

Examples

Basic connection management:

from haive.mcp.client import MCPConnection, StdioTransport

transport = StdioTransport("npx", ["-y", "@modelcontextprotocol/server-filesystem"])
connection = MCPConnection("filesystem", transport)

await connection.connect()
client = connection.get_client()
tools = await client.list_tools()
await connection.disconnect()

With health monitoring:

connection = MCPConnection(
    "filesystem",
    transport,
    health_check_interval=30.0,
    auto_reconnect=True
)

await connection.start_monitoring()
# Connection will be monitored and auto-reconnected

Connection status checking:

if connection.is_healthy():
    client = connection.get_client()
    result = await client.call_tool("read_file", {"path": "/tmp/test"})

Initialize MCP connection manager.

Parameters:

• name (str) – Connection name/identifier

• transport (haive.mcp.client.transport.MCPTransport) – Transport implementation

• auto_reconnect (bool) – Enable automatic reconnection

• max_reconnect_attempts (int) – Maximum reconnection attempts

• reconnect_delay (float) – Initial reconnection delay (seconds)

• max_reconnect_delay (float) – Maximum reconnection delay (seconds)

• health_check_interval (float) – Health check interval (seconds)

• connection_timeout (float) – Connection timeout (seconds)

async connect(timeout=None)

Connect to the MCP server.

Parameters:

timeout (Optional[float]) – Connection timeout (uses default if None)

Returns:

Connected MCP client

Raises:

MCPConnectionError – If connection fails

Return type:

haive.mcp.client.mcp_client.MCPClient

async disconnect()

Disconnect from the MCP server.

Return type:

None

get_client()

Get the MCP client if connected.

Returns:

MCP client instance

Raises:

MCPConnectionError – If not connected

Return type:

haive.mcp.client.mcp_client.MCPClient

get_info()

Get connection information.

Returns:

ConnectionInfo object with current state

Return type:

ConnectionInfo

async health_check()

Perform a health check on the connection.

Returns:

Health check results

Return type:

Dict[str, Any]

is_connected()

Check if connection is active.

Returns:

True if connected, False otherwise

Return type:

bool

is_healthy(threshold=0.7)

Check if connection is healthy.

Parameters:

threshold (float) – Health score threshold (0.0-1.0)

Returns:

True if health score is above threshold

Return type:

bool

async reconnect()

Attempt to reconnect to the server.

Returns:

Connected client if successful, None if failed

Return type:

Optional[haive.mcp.client.mcp_client.MCPClient]

async start_monitoring()

Start background health monitoring.

Return type:

None

async stop_monitoring()

Stop background health monitoring.

Return type:

None