mcp.client.protocol

MCP Protocol Implementation.

This module implements the Model Context Protocol (MCP) JSON-RPC based communication protocol. It handles the protocol-level details including message framing, request/response matching, and capability negotiation.

The protocol implementation is transport-agnostic and works with any transport that implements the MCPTransport interface.

Attributes

logger

Classes

MCPCapability

Standard MCP capabilities.

MCPMessage

Base MCP message.

MCPMessageType

MCP message types.

MCPMethod

Standard MCP methods.

MCPPrompt

MCP prompt definition.

MCPProtocol

MCP protocol implementation.

MCPProtocolVersion

Supported MCP protocol versions.

MCPResource

MCP resource definition.

MCPTool

MCP tool definition.

Module Contents

class mcp.client.protocol.MCPCapability

Bases: str, enum.Enum

Standard MCP capabilities.

LOGGING = 'logging'
PROMPTS = 'prompts'
RESOURCES = 'resources'
SAMPLING = 'sampling'
TOOLS = 'tools'
class mcp.client.protocol.MCPMessage(/, **data: Any)

Bases: pydantic.BaseModel

Base MCP message.

error: Dict[str, Any] | None = None
id: str | int | None = None
jsonrpc: str = None
method: str | None = None
params: Dict[str, Any] | None = None
result: Any | None = None
class mcp.client.protocol.MCPMessageType

Bases: str, enum.Enum

MCP message types.

NOTIFICATION = 'notification'
REQUEST = 'request'
RESPONSE = 'response'
class mcp.client.protocol.MCPMethod

Bases: str, enum.Enum

Standard MCP methods.

CANCELLED = 'cancelled'
INITIALIZE = 'initialize'
INITIALIZED = 'initialized'
LOGGING_SET_LEVEL = 'logging/setLevel'
PROGRESS = 'progress'
PROMPTS_GET = 'prompts/get'
PROMPTS_LIST = 'prompts/list'
PROMPT_LIST_CHANGED = 'prompts/list_changed'
RESOURCES_LIST = 'resources/list'
RESOURCES_READ = 'resources/read'
RESOURCES_SUBSCRIBE = 'resources/subscribe'
RESOURCES_UNSUBSCRIBE = 'resources/unsubscribe'
RESOURCE_LIST_CHANGED = 'resources/list_changed'
RESOURCE_UPDATED = 'resources/updated'
TOOLS_CALL = 'tools/call'
TOOLS_LIST = 'tools/list'
TOOL_LIST_CHANGED = 'tools/list_changed'
class mcp.client.protocol.MCPPrompt(/, **data: Any)

Bases: pydantic.BaseModel

MCP prompt definition.

arguments: List[Dict[str, Any]] | None = None
description: str = None
name: str = None
class mcp.client.protocol.MCPProtocol(transport, timeout: float = 30.0, client_info: Dict[str, Any] | None = None)

MCP protocol implementation.

This class handles the MCP protocol layer, including:

  • Message serialization/deserialization

  • Request/response matching

  • Capability negotiation

  • Protocol state management

  • Error handling

The protocol is transport-agnostic and works with any MCPTransport implementation. It provides a clean async API for MCP operations.

Examples

Basic protocol usage:

from haive.mcp.client.transport import StdioTransport
from haive.mcp.client.protocol import MCPProtocol

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

await protocol.initialize()
tools = await protocol.list_tools()
result = await protocol.call_tool("read_file", {"path": "/etc/hosts"})
await protocol.shutdown()

With context manager:

async with MCPProtocol(transport) as protocol:
    tools = await protocol.list_tools()
    result = await protocol.call_tool("tool_name", args)
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 notifications.

Parameters:

  • method – Notification method name

  • handler – Async handler function

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

Call a tool on the server.

Parameters:

  • name – Tool name to call

  • arguments – Tool arguments

Returns:

Tool execution result

Raises:
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 initialize() → Dict[str, Any]

Initialize the MCP connection.

This performs the MCP initialization handshake, including capability negotiation and protocol version agreement.

Returns:

Server information and capabilities

Raises:
async list_prompts() → List[MCPPrompt]

List available prompts from the server.

Returns:

List of available prompts

Raises:

MCPCapabilityError – If prompts capability not supported

async list_resources() → List[MCPResource]

List available resources from the server.

Returns:

List of available resources

async list_tools() → List[MCPTool]

List available tools from the server.

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

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

async shutdown() → None

Shutdown the MCP connection gracefully.

client_info
initialized = False
protocol_version: MCPProtocolVersion | None = None
server_capabilities: Set[MCPCapability]
server_info: Dict[str, Any] | None = None
timeout = 30.0
transport
class mcp.client.protocol.MCPProtocolVersion

Bases: str, enum.Enum

Supported MCP protocol versions.

V0_9 = '0.9'
V1_0 = '1.0'
class mcp.client.protocol.MCPResource(/, **data: Any)

Bases: pydantic.BaseModel

MCP resource definition.

description: str | None = None
mimeType: str | None = None
name: str = None
uri: str = None
class mcp.client.protocol.MCPTool(/, **data: Any)

Bases: pydantic.BaseModel

MCP tool definition.

description: str = None
inputSchema: Dict[str, Any] = None
name: str = None
mcp.client.protocol.logger