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

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 haive.mcp.client.protocol.MCPCapability[source]¶

Bases: str, enum.Enum

Standard MCP capabilities.

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

class haive.mcp.client.protocol.MCPMessage(/, **data)[source]¶

Bases: pydantic.BaseModel

Base MCP message.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class haive.mcp.client.protocol.MCPMessageType[source]¶

Bases: str, enum.Enum

MCP message types.

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

class haive.mcp.client.protocol.MCPMethod[source]¶

Bases: str, enum.Enum

Standard MCP methods.

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

class haive.mcp.client.protocol.MCPPrompt(/, **data)[source]¶

Bases: pydantic.BaseModel

MCP prompt definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class haive.mcp.client.protocol.MCPProtocol(transport, timeout=30.0, client_info=None)[source]¶

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)

Initialize MCP protocol.

Parameters:

transport – Transport implementation
timeout (float) – Default timeout for requests
client_info (Optional[Dict[str, Any]]) – Client information for initialization

add_notification_handler(method, handler)[source]¶

Add a handler for 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)[source]¶

Call a tool on the server.

Parameters:

name (str) – Tool name to call
arguments (Optional[Dict[str, Any]]) – Tool arguments

Returns:

Tool execution result

Raises:

MCPCapabilityError – If tools capability not supported
MCPToolError – If tool execution fails
MCPProtocolError – If request fails

Return type:

Any

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 initialize()[source]¶

Initialize the MCP connection.

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

Returns:

Server information and capabilities

Raises:

MCPProtocolError – If initialization fails
MCPCapabilityError – If capabilities are incompatible

Return type:

Dict[str, Any]

async list_prompts()[source]¶

List available prompts from the server.

Returns:: List of available prompts
Raises:: MCPCapabilityError – If prompts capability not supported
Return type:: List[MCPPrompt]

async list_resources()[source]¶

List available resources from the server.

Returns:: List of available resources
Return type:: List[MCPResource]

async list_tools()[source]¶

List available tools from the server.

Returns:

List of available tools

Raises:

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

Return type:

List[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]

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

async shutdown()[source]¶

Shutdown the MCP connection gracefully.

Return type:: None

class haive.mcp.client.protocol.MCPProtocolVersion[source]¶

Bases: str, enum.Enum

Supported MCP protocol versions.

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

class haive.mcp.client.protocol.MCPResource(/, **data)[source]¶

Bases: pydantic.BaseModel

MCP resource definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class haive.mcp.client.protocol.MCPTool(/, **data)[source]¶

Bases: pydantic.BaseModel

MCP tool definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)