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.

Attributes¶

logger

Classes¶

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

Module Contents¶

class mcp.client.connection.ConnectionInfo¶

Information about an MCP connection.

client: mcp.client.mcp_client.MCPClient | None = None¶

health_score: float = 1.0¶

last_connected: float | None = None¶

last_error: str | None = None¶

metadata: Dict[str, Any]¶

name: str¶

reconnect_attempts: int = 0¶

status: ConnectionStatus¶

transport: mcp.client.transport.MCPTransport¶

class mcp.client.connection.ConnectionStatus¶

Bases: str, enum.Enum

Connection status states.

CONNECTED = 'connected'¶

CONNECTING = 'connecting'¶

DISCONNECTED = 'disconnected'¶

ERROR = 'error'¶

RECONNECTING = 'reconnecting'¶

class mcp.client.connection.MCPConnection(name: str, transport: mcp.client.transport.MCPTransport, auto_reconnect: bool = True, max_reconnect_attempts: int = 5, reconnect_delay: float = 1.0, max_reconnect_delay: float = 60.0, health_check_interval: float = 30.0, connection_timeout: float = 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"})

async __aenter__()¶: Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)¶: Async context manager exit.

async connect(timeout: float | None = None) → mcp.client.mcp_client.MCPClient¶

Connect to the MCP server.

Parameters:: timeout – Connection timeout (uses default if None)
Returns:: Connected MCP client
Raises:: MCPConnectionError – If connection fails

async disconnect() → None¶: Disconnect from the MCP server.

get_client() → mcp.client.mcp_client.MCPClient¶

Get the MCP client if connected.

Returns:: MCP client instance
Raises:: MCPConnectionError – If not connected

get_info() → ConnectionInfo¶

Get connection information.

Returns:: ConnectionInfo object with current state

async health_check() → Dict[str, Any]¶

Perform a health check on the connection.

Returns:: Health check results

is_connected() → bool¶

Check if connection is active.

Returns:: True if connected, False otherwise

is_healthy(threshold: float = 0.7) → bool¶

Check if connection is healthy.

Parameters:: threshold – Health score threshold (0.0-1.0)
Returns:: True if health score is above threshold

async reconnect() → mcp.client.mcp_client.MCPClient | None¶

Attempt to reconnect to the server.

Returns:: Connected client if successful, None if failed

async start_monitoring() → None¶: Start background health monitoring.

async stop_monitoring() → None¶: Stop background health monitoring.

auto_reconnect = True¶

client: mcp.client.mcp_client.MCPClient | None = None¶

connection_count = 0¶

connection_timeout = 30.0¶

error_count = 0¶

health_check_interval = 30.0¶

health_score = 1.0¶

last_connected: float | None = None¶

last_error: str | None = None¶

last_operation_time = 0.0¶

max_reconnect_attempts = 5¶

max_reconnect_delay = 60.0¶

name¶

reconnect_attempts = 0¶

reconnect_delay = 1.0¶

status¶

total_uptime = 0.0¶

transport¶

mcp.client.connection.logger¶