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

Classes¶

MCPClient

High-level MCP client for communicating with MCP servers.

Module Contents¶

class haive.mcp.client.mcp_client.MCPClient(transport, timeout=30.0, client_info=None, auto_reconnect=False, max_reconnect_attempts=3)[source]¶

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:

Connection establishment
Capability negotiation
Tool/resource/prompt discovery
Operation execution
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}")

Initialize MCP client.

Parameters:

transport (haive.mcp.client.transport.MCPTransport) – Transport implementation (STDIO, HTTP, etc.)
timeout (float) – Default timeout for operations
client_info (Optional[Dict[str, Any]]) – Client information for server handshake
auto_reconnect (bool) – Whether to automatically reconnect on failures
max_reconnect_attempts (int) – Maximum reconnection attempts

add_notification_handler(method, handler)[source]¶

Add a handler for server notifications.

Parameters:

method (str) – Notification method name
handler (Callable[[Dict[str, Any]], Awaitable[None]]) – Async handler function

Return type:

None

async call_tool(name, arguments=None, timeout=None)[source]¶

Call a tool on the server.

Parameters:

name (str) – Tool name to call
arguments (Optional[Dict[str, Any]]) – Tool arguments
timeout (Optional[float]) – Call timeout (uses default if None)

Returns:

Tool execution result

Raises:

MCPToolError – If tool execution fails
MCPProtocolError – If request fails

Return type:

Any

async connect()[source]¶

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:

MCPConnectionError – If connection fails
MCPProtocolError – If protocol handshake fails

Return type:

Dict[str, Any]

async disconnect()[source]¶

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.

Return type:: None

async get_capabilities()[source]¶

Get server capabilities.

Returns:: List of server capabilities
Raises:: MCPConnectionError – If not connected
Return type:: List[haive.mcp.client.protocol.MCPCapability]

async get_prompt(name, arguments=None)[source]¶

Get a prompt from the server.

Parameters:

name (str) – Prompt name
arguments (Optional[Dict[str, Any]]) – Prompt arguments

Returns:

Prompt content and metadata

Return type:

Dict[str, Any]

async get_server_info()[source]¶

Get server information from initialization.

Returns:: Server information dictionary
Raises:: MCPConnectionError – If not connected
Return type:: Dict[str, Any]

async health_check()[source]¶

Perform a health check on the MCP connection.

Returns:: Health check results including connectivity and capabilities
Return type:: Dict[str, Any]

async is_connected()[source]¶

Check if client is currently connected.

Returns:: True if connected, False otherwise
Return type:: bool

async list_prompts(use_cache=True)[source]¶

List available prompts from the server.

Parameters:: use_cache (bool) – Whether to use cached results
Returns:: List of available prompts
Raises:: MCPCapabilityError – If prompts capability not supported
Return type:: List[haive.mcp.client.protocol.MCPPrompt]

async list_resources(use_cache=True)[source]¶

List available resources from the server.

Parameters:: use_cache (bool) – Whether to use cached results
Returns:: List of available resources
Return type:: List[haive.mcp.client.protocol.MCPResource]

async list_tools(use_cache=True)[source]¶

List available tools from the server.

Parameters:

use_cache (bool) – Whether to use cached results if available

Returns:

List of available tools

Raises:

MCPCapabilityError – If tools capability not supported
MCPProtocolError – If request fails

Return type:

List[haive.mcp.client.protocol.MCPTool]

async read_resource(uri)[source]¶

Read a resource from the server.

Parameters:: uri (str) – Resource URI to read
Returns:: Resource content and metadata
Return type:: Dict[str, Any]

async refresh_cache()[source]¶

Refresh all cached server information.

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

Return type:: None

remove_notification_handler(method, handler)[source]¶

Remove a notification handler.

Parameters:

method (str) – Notification method name
handler (Callable[[Dict[str, Any]], Awaitable[None]]) – Handler function to remove

Return type:

None