mcp.client.protocol =================== .. py:module:: mcp.client.protocol .. autoapi-nested-parse:: 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 ---------- .. autoapisummary:: mcp.client.protocol.logger Classes ------- .. autoapisummary:: mcp.client.protocol.MCPCapability mcp.client.protocol.MCPMessage mcp.client.protocol.MCPMessageType mcp.client.protocol.MCPMethod mcp.client.protocol.MCPPrompt mcp.client.protocol.MCPProtocol mcp.client.protocol.MCPProtocolVersion mcp.client.protocol.MCPResource mcp.client.protocol.MCPTool Module Contents --------------- .. py:class:: MCPCapability Bases: :py:obj:`str`, :py:obj:`enum.Enum` Standard MCP capabilities. .. py:attribute:: LOGGING :value: 'logging' .. py:attribute:: PROMPTS :value: 'prompts' .. py:attribute:: RESOURCES :value: 'resources' .. py:attribute:: SAMPLING :value: 'sampling' .. py:attribute:: TOOLS :value: 'tools' .. py:class:: MCPMessage(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Base MCP message. .. py:attribute:: error :type: Optional[Dict[str, Any]] :value: None .. py:attribute:: id :type: Optional[Union[str, int]] :value: None .. py:attribute:: jsonrpc :type: str :value: None .. py:attribute:: method :type: Optional[str] :value: None .. py:attribute:: params :type: Optional[Dict[str, Any]] :value: None .. py:attribute:: result :type: Optional[Any] :value: None .. py:class:: MCPMessageType Bases: :py:obj:`str`, :py:obj:`enum.Enum` MCP message types. .. py:attribute:: NOTIFICATION :value: 'notification' .. py:attribute:: REQUEST :value: 'request' .. py:attribute:: RESPONSE :value: 'response' .. py:class:: MCPMethod Bases: :py:obj:`str`, :py:obj:`enum.Enum` Standard MCP methods. .. py:attribute:: CANCELLED :value: 'cancelled' .. py:attribute:: INITIALIZE :value: 'initialize' .. py:attribute:: INITIALIZED :value: 'initialized' .. py:attribute:: LOGGING_SET_LEVEL :value: 'logging/setLevel' .. py:attribute:: PROGRESS :value: 'progress' .. py:attribute:: PROMPTS_GET :value: 'prompts/get' .. py:attribute:: PROMPTS_LIST :value: 'prompts/list' .. py:attribute:: PROMPT_LIST_CHANGED :value: 'prompts/list_changed' .. py:attribute:: RESOURCES_LIST :value: 'resources/list' .. py:attribute:: RESOURCES_READ :value: 'resources/read' .. py:attribute:: RESOURCES_SUBSCRIBE :value: 'resources/subscribe' .. py:attribute:: RESOURCES_UNSUBSCRIBE :value: 'resources/unsubscribe' .. py:attribute:: RESOURCE_LIST_CHANGED :value: 'resources/list_changed' .. py:attribute:: RESOURCE_UPDATED :value: 'resources/updated' .. py:attribute:: TOOLS_CALL :value: 'tools/call' .. py:attribute:: TOOLS_LIST :value: 'tools/list' .. py:attribute:: TOOL_LIST_CHANGED :value: 'tools/list_changed' .. py:class:: MCPPrompt(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` MCP prompt definition. .. py:attribute:: arguments :type: Optional[List[Dict[str, Any]]] :value: None .. py:attribute:: description :type: str :value: None .. py:attribute:: name :type: str :value: None .. py:class:: MCPProtocol(transport, timeout: float = 30.0, client_info: Optional[Dict[str, Any]] = 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. .. rubric:: 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) .. py:method:: __aenter__() :async: Async context manager entry. .. py:method:: __aexit__(exc_type, exc_val, exc_tb) :async: Async context manager exit. .. py:method:: add_notification_handler(method: str, handler: Callable[[Dict[str, Any]], Awaitable[None]]) -> None Add a handler for notifications. :param method: Notification method name :param handler: Async handler function .. py:method:: call_tool(name: str, arguments: Optional[Dict[str, Any]] = None) -> Any :async: Call a tool on the server. :param name: Tool name to call :param arguments: Tool arguments :returns: Tool execution result :raises MCPCapabilityError: If tools capability not supported :raises MCPToolError: If tool execution fails :raises MCPProtocolError: If request fails .. py:method:: get_prompt(name: str, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any] :async: Get a prompt from the server. :param name: Prompt name :param arguments: Prompt arguments :returns: Prompt content and metadata .. py:method:: initialize() -> Dict[str, Any] :async: 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 :raises MCPCapabilityError: If capabilities are incompatible .. py:method:: list_prompts() -> List[MCPPrompt] :async: List available prompts from the server. :returns: List of available prompts :raises MCPCapabilityError: If prompts capability not supported .. py:method:: list_resources() -> List[MCPResource] :async: List available resources from the server. :returns: List of available resources .. py:method:: list_tools() -> List[MCPTool] :async: List available tools from the server. :returns: List of available tools :raises MCPCapabilityError: If tools capability not supported :raises MCPProtocolError: If request fails .. py:method:: read_resource(uri: str) -> Dict[str, Any] :async: Read a resource from the server. :param uri: Resource URI to read :returns: Resource content and metadata .. py:method:: remove_notification_handler(method: str, handler: Callable[[Dict[str, Any]], Awaitable[None]]) -> None Remove a notification handler. :param method: Notification method name :param handler: Handler function to remove .. py:method:: shutdown() -> None :async: Shutdown the MCP connection gracefully. .. py:attribute:: client_info .. py:attribute:: initialized :value: False .. py:attribute:: protocol_version :type: Optional[MCPProtocolVersion] :value: None .. py:attribute:: server_capabilities :type: Set[MCPCapability] .. py:attribute:: server_info :type: Optional[Dict[str, Any]] :value: None .. py:attribute:: timeout :value: 30.0 .. py:attribute:: transport .. py:class:: MCPProtocolVersion Bases: :py:obj:`str`, :py:obj:`enum.Enum` Supported MCP protocol versions. .. py:attribute:: V0_9 :value: '0.9' .. py:attribute:: V1_0 :value: '1.0' .. py:class:: MCPResource(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` MCP resource definition. .. py:attribute:: description :type: Optional[str] :value: None .. py:attribute:: mimeType :type: Optional[str] :value: None .. py:attribute:: name :type: str :value: None .. py:attribute:: uri :type: str :value: None .. py:class:: MCPTool(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` MCP tool definition. .. py:attribute:: description :type: str :value: None .. py:attribute:: inputSchema :type: Dict[str, Any] :value: None .. py:attribute:: name :type: str :value: None .. py:data:: logger