mcp.client.mcp_client

MCP Client Main Implementation.

This module provides the main MCPClient class that orchestrates the MCP protocol communication. It combines the transport layer, protocol layer, and provides a high-level interface for MCP operations.

The client handles:

  • Connection management and lifecycle

  • Tool discovery and execution

  • Resource access and management

  • Prompt retrieval and execution

  • Server capability discovery

  • Error handling and recovery

Attributes

logger

Classes

MCPClient

High-level MCP client for communicating with MCP servers.

Module Contents

class mcp.client.mcp_client.MCPClient(transport: mcp.client.transport.MCPTransport, timeout: float = 30.0, client_info: Dict[str, Any] | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 3)

High-level MCP client for communicating with MCP servers.

This is the main interface for interacting with MCP servers. It combines the transport and protocol layers to provide a clean, easy-to-use API for MCP operations.

The client handles the complete MCP lifecycle:

  1. Connection establishment

  2. Capability negotiation

  3. Tool/resource/prompt discovery

  4. Operation execution

  5. Connection cleanup

It supports multiple transport types and provides both sync-style and async context manager interfaces.

Examples

Basic usage with STDIO transport:

from haive.mcp.client import MCPClient, StdioTransport

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

await client.connect()
tools = await client.list_tools()
result = await client.call_tool("read_file", {"path": "/etc/hosts"})
await client.disconnect()

Using context manager (recommended):

async with MCPClient(transport) as client:
    tools = await client.list_tools()
    for tool in tools:
        print(f"Available tool: {tool.name}")

    result = await client.call_tool("tool_name", {"arg": "value"})

With HTTP transport:

from haive.mcp.client import HttpTransport

transport = HttpTransport("http://localhost:8080/mcp")
async with MCPClient(transport) as client:
    resources = await client.list_resources()
    content = await client.read_resource("file://config.json")

With notification handling:

def on_tool_list_changed(params):
    print("Tool list updated!")

client.add_notification_handler("tools/list_changed", on_tool_list_changed)

Error handling:

try:
    async with MCPClient(transport) as client:
        result = await client.call_tool("nonexistent", {})
except MCPToolError as e:
    print(f"Tool error: {e}")
except MCPConnectionError as e:
    print(f"Connection error: {e}")
async __aenter__()

Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)

Async context manager exit.

add_notification_handler(method: str, handler: Callable[[Dict[str, Any]], Awaitable[None]]) → None

Add a handler for server notifications.

Parameters:

  • method – Notification method name

  • handler – Async handler function

async call_tool(name: str, arguments: Dict[str, Any] | None = None, timeout: float | None = None) → Any

Call a tool on the server.

Parameters:

  • name – Tool name to call

  • arguments – Tool arguments

  • timeout – Call timeout (uses default if None)

Returns:

Tool execution result

Raises:
async connect() → Dict[str, Any]

Connect to the MCP server and perform initialization.

This method establishes the connection and performs the MCP initialization handshake, including capability negotiation.

Returns:

Server information and capabilities

Raises:
async disconnect() → None

Disconnect from the MCP server gracefully.

This method cleanly shuts down the MCP connection and cleans up all resources. It’s safe to call multiple times.

async get_capabilities() → List[mcp.client.protocol.MCPCapability]

Get server capabilities.

Returns:

List of server capabilities

Raises:

MCPConnectionError – If not connected

async get_prompt(name: str, arguments: Dict[str, Any] | None = None) → Dict[str, Any]

Get a prompt from the server.

Parameters:

  • name – Prompt name

  • arguments – Prompt arguments

Returns:

Prompt content and metadata

async get_server_info() → Dict[str, Any]

Get server information from initialization.

Returns:

Server information dictionary

Raises:

MCPConnectionError – If not connected

async health_check() → Dict[str, Any]

Perform a health check on the MCP connection.

Returns:

Health check results including connectivity and capabilities

async is_connected() → bool

Check if client is currently connected.

Returns:

True if connected, False otherwise

async list_prompts(use_cache: bool = True) → List[mcp.client.protocol.MCPPrompt]

List available prompts from the server.

Parameters:

use_cache – Whether to use cached results

Returns:

List of available prompts

Raises:

MCPCapabilityError – If prompts capability not supported

async list_resources(use_cache: bool = True) → List[mcp.client.protocol.MCPResource]

List available resources from the server.

Parameters:

use_cache – Whether to use cached results

Returns:

List of available resources

async list_tools(use_cache: bool = True) → List[mcp.client.protocol.MCPTool]

List available tools from the server.

Parameters:

use_cache – Whether to use cached results if available

Returns:

List of available tools

Raises:
async read_resource(uri: str) → Dict[str, Any]

Read a resource from the server.

Parameters:

uri – Resource URI to read

Returns:

Resource content and metadata

async refresh_cache() → None

Refresh all cached server information.

This re-fetches tools, prompts, and resources from the server and updates the local cache.

remove_notification_handler(method: str, handler: Callable[[Dict[str, Any]], Awaitable[None]]) → None

Remove a notification handler.

Parameters:

  • method – Notification method name

  • handler – Handler function to remove

auto_reconnect = False
max_reconnect_attempts = 3
protocol
timeout = 30.0
transport
mcp.client.mcp_client.logger