mcp.client ========== .. py:module:: mcp.client .. autoapi-nested-parse:: MCP Client Implementation Package. This package provides native MCP protocol client implementation for connecting to and communicating with MCP servers according to the Model Context Protocol specification. Key Components -------------- **Core Classes** - :class:`MCPClient`: Main client for protocol communication - :class:`MCPTransport`: Transport layer abstraction - :class:`MCPConnection`: Connection management - :class:`MCPProtocol`: Protocol implementation **Transport Support** - **STDIO**: Communication via stdin/stdout - **HTTP**: RESTful communication - **SSE**: Server-sent events - **WebSocket**: Real-time bidirectional communication Usage Examples -------------- Basic Connection ~~~~~~~~~~~~~~~~ .. code-block:: python from haive.mcp.client import MCPClient, StdioTransport # Create transport and client transport = StdioTransport( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem"] ) client = MCPClient(transport) # Connect and use await client.connect() tools = await client.list_tools() result = await client.call_tool("read_file", {"path": "/etc/hosts"}) await client.disconnect() Context Manager Usage ~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python async with MCPClient(transport) as client: tools = await client.list_tools() result = await client.call_tool("tool_name", args) Available Classes ----------------- Transport Classes ~~~~~~~~~~~~~~~~~ - :class:`StdioTransport` - Standard I/O transport - :class:`HttpTransport` - HTTP-based transport - :class:`SseTransport` - Server-Sent Events transport - :class:`WebSocketTransport` - WebSocket transport Exception Classes ~~~~~~~~~~~~~~~~~ - :class:`MCPError` - Base MCP exception - :class:`MCPConnectionError` - Connection-related errors - :class:`MCPProtocolError` - Protocol violation errors - :class:`MCPTimeoutError` - Timeout-related errors - :class:`MCPTransportError` - Transport layer errors - :class:`MCPAuthenticationError` - Authentication failures - :class:`MCPCapabilityError` - Capability negotiation errors - :class:`MCPToolError` - Tool execution errors .. note:: This is a native implementation of the MCP protocol, designed to work with any MCP-compliant server. It handles the full protocol lifecycle including initialization, capability discovery, and tool execution. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/mcp/client/connection/index /autoapi/mcp/client/exceptions/index /autoapi/mcp/client/mcp_client/index /autoapi/mcp/client/protocol/index /autoapi/mcp/client/transport/index Exceptions ---------- .. autoapisummary:: mcp.client.MCPAuthenticationError mcp.client.MCPCapabilityError mcp.client.MCPConnectionError mcp.client.MCPError mcp.client.MCPProtocolError mcp.client.MCPTimeoutError mcp.client.MCPToolError mcp.client.MCPTransportError Classes ------- .. autoapisummary:: mcp.client.HttpTransport mcp.client.MCPClient mcp.client.MCPConnection mcp.client.MCPProtocol mcp.client.MCPTransport mcp.client.SseTransport mcp.client.StdioTransport mcp.client.WebSocketTransport Package Contents ---------------- .. py:exception:: MCPAuthenticationError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPError` Authentication failure with MCP server. Raised when authentication is required but fails, or when authorization is denied for a specific operation. This includes token validation failures and permission issues. .. rubric:: Examples Invalid token:: raise MCPAuthenticationError( "Invalid authentication token", error_code=401 ) Insufficient permissions:: raise MCPAuthenticationError( "Insufficient permissions for tool execution", details={"tool": "sensitive_operation", "required_scope": "admin"} ) .. py:exception:: MCPCapabilityError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPProtocolError` Error related to MCP capabilities. Raised when there are issues with capability negotiation, unsupported capabilities, or capability validation failures. This indicates a mismatch between client requirements and server capabilities. .. rubric:: Examples Unsupported capability:: raise MCPCapabilityError( "Server doesn't support required capability", details={"required": "tools", "supported": ["logging"]} ) .. py:exception:: MCPConnectionError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPError` Error during MCP connection establishment or management. Raised when there are issues connecting to, maintaining, or disconnecting from an MCP server. This includes transport-level failures, timeout issues, and connection state problems. .. rubric:: Examples Connection timeout:: raise MCPConnectionError("Connection timeout after 30s") Transport failure:: raise MCPConnectionError( "Failed to start server process", details={"command": ["npx", "server"], "exit_code": 1} ) .. py:exception:: MCPError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`Exception` Base exception for all MCP-related errors. This is the root exception class for all MCP client errors. It provides a consistent interface for error handling and includes optional error details and context information. :param message: Human-readable error description :param error_code: MCP protocol error code (if applicable) :param details: Additional error details or context .. rubric:: Examples Basic error:: raise MCPError("Connection failed") With error code:: raise MCPError("Invalid request", error_code=-32600) With details:: raise MCPError( "Tool execution failed", details={"tool": "filesystem", "args": {...}} ) .. py:method:: __str__() -> str Return formatted error message. .. py:attribute:: details .. py:attribute:: error_code :value: None .. py:attribute:: message .. py:exception:: MCPProtocolError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPError` Error in MCP protocol communication. Raised when there are protocol-level issues such as invalid messages, unsupported protocol versions, or malformed responses. This indicates a problem with the MCP protocol implementation or server compliance. .. rubric:: Examples Protocol version mismatch:: raise MCPProtocolError( "Unsupported protocol version", details={"client": "1.0", "server": "0.9"} ) Invalid message format:: raise MCPProtocolError( "Malformed JSON-RPC message", error_code=-32700, details={"raw_message": "invalid json"} ) .. py:exception:: MCPTimeoutError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPConnectionError` Timeout during MCP operation. Raised when an MCP operation (connection, tool call, etc.) exceeds the configured timeout period. This is a specialized connection error that specifically indicates timing issues. :param operation: Name of the operation that timed out :param timeout: Timeout value that was exceeded .. rubric:: Examples Tool execution timeout:: raise MCPTimeoutError( "Tool call timed out", details={"operation": "call_tool", "timeout": 30.0} ) .. py:exception:: MCPToolError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPError` Error during tool execution. Raised when there are issues executing tools on the MCP server. This includes tool not found errors, invalid arguments, and execution failures. .. rubric:: Examples Tool not found:: raise MCPToolError( "Tool not found", details={"tool_name": "nonexistent_tool"} ) Invalid arguments:: raise MCPToolError( "Invalid tool arguments", details={"tool": "read_file", "error": "path is required"} ) .. py:exception:: MCPTransportError(message: str, error_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None) Bases: :py:obj:`MCPConnectionError` Error in the transport layer. Raised when there are issues with the underlying transport mechanism (STDIO, HTTP, WebSocket, etc.). This includes process management failures, network issues, and transport-specific problems. .. rubric:: Examples Process died unexpectedly:: raise MCPTransportError( "Server process died", details={"pid": 12345, "exit_code": -9} ) Network error:: raise MCPTransportError( "HTTP connection failed", details={"url": "http://localhost:8080", "status": 500} ) .. py:class:: HttpTransport(url: str, headers: Optional[Dict[str, str]] = None, timeout: float = 30.0) Bases: :py:obj:`MCPTransport` HTTP transport for MCP communication. This transport communicates with an MCP server over HTTP using JSON-RPC over HTTP POST requests. This is useful for servers that expose HTTP APIs or for remote MCP servers. .. rubric:: Examples Basic HTTP transport:: transport = HttpTransport("http://localhost:8080/mcp") With authentication:: transport = HttpTransport( "https://api.example.com/mcp", headers={"Authorization": "Bearer token123"} ) .. py:method:: connect() -> None :async: Create HTTP session. .. py:method:: disconnect() -> None :async: Close HTTP session. .. py:method:: receive_message() -> Dict[str, Any] :async: Send pending message and receive response. .. py:method:: send_message(message: Dict[str, Any]) -> None :async: Send JSON-RPC message via HTTP POST. Note: For HTTP transport, sending and receiving are combined in a request-response cycle. This method stores the message for the next receive_message call. .. py:attribute:: headers .. py:attribute:: session :type: Optional[aiohttp.ClientSession] :value: None .. py:attribute:: url .. py:class:: MCPClient(transport: mcp.client.transport.MCPTransport, timeout: float = 30.0, client_info: Optional[Dict[str, Any]] = 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. .. rubric:: 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}") .. 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 server notifications. :param method: Notification method name :param handler: Async handler function .. py:method:: call_tool(name: str, arguments: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Any :async: Call a tool on the server. :param name: Tool name to call :param arguments: Tool arguments :param timeout: Call timeout (uses default if None) :returns: Tool execution result :raises MCPToolError: If tool execution fails :raises MCPProtocolError: If request fails .. py:method:: connect() -> Dict[str, Any] :async: 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 :raises MCPProtocolError: If protocol handshake fails .. py:method:: disconnect() -> None :async: 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. .. py:method:: get_capabilities() -> List[mcp.client.protocol.MCPCapability] :async: Get server capabilities. :returns: List of server capabilities :raises MCPConnectionError: If not connected .. 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:: get_server_info() -> Dict[str, Any] :async: Get server information from initialization. :returns: Server information dictionary :raises MCPConnectionError: If not connected .. py:method:: health_check() -> Dict[str, Any] :async: Perform a health check on the MCP connection. :returns: Health check results including connectivity and capabilities .. py:method:: is_connected() -> bool :async: Check if client is currently connected. :returns: True if connected, False otherwise .. py:method:: list_prompts(use_cache: bool = True) -> List[mcp.client.protocol.MCPPrompt] :async: List available prompts from the server. :param use_cache: Whether to use cached results :returns: List of available prompts :raises MCPCapabilityError: If prompts capability not supported .. py:method:: list_resources(use_cache: bool = True) -> List[mcp.client.protocol.MCPResource] :async: List available resources from the server. :param use_cache: Whether to use cached results :returns: List of available resources .. py:method:: list_tools(use_cache: bool = True) -> List[mcp.client.protocol.MCPTool] :async: List available tools from the server. :param use_cache: Whether to use cached results if available :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:: refresh_cache() -> None :async: Refresh all cached server information. This re-fetches tools, prompts, and resources from the server and updates the local cache. .. 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:attribute:: auto_reconnect :value: False .. py:attribute:: max_reconnect_attempts :value: 3 .. py:attribute:: protocol .. py:attribute:: timeout :value: 30.0 .. py:attribute:: transport .. py:class:: MCPConnection(name: str, transport: mcp.client.transport.MCPTransport, auto_reconnect: bool = True, max_reconnect_attempts: int = 5, reconnect_delay: float = 1.0, max_reconnect_delay: float = 60.0, health_check_interval: float = 30.0, connection_timeout: float = 30.0) MCP connection manager with health monitoring and auto-reconnection. This class manages individual MCP connections with features like: - Health monitoring and scoring - Automatic reconnection with backoff - Connection lifecycle management - Error tracking and recovery - Performance metrics .. rubric:: Examples Basic connection management:: from haive.mcp.client import MCPConnection, StdioTransport transport = StdioTransport("npx", ["-y", "@modelcontextprotocol/server-filesystem"]) connection = MCPConnection("filesystem", transport) await connection.connect() client = connection.get_client() tools = await client.list_tools() await connection.disconnect() With health monitoring:: connection = MCPConnection( "filesystem", transport, health_check_interval=30.0, auto_reconnect=True ) await connection.start_monitoring() # Connection will be monitored and auto-reconnected Connection status checking:: if connection.is_healthy(): client = connection.get_client() result = await client.call_tool("read_file", {"path": "/tmp/test"}) .. py:method:: __aenter__() :async: Async context manager entry. .. py:method:: __aexit__(exc_type, exc_val, exc_tb) :async: Async context manager exit. .. py:method:: connect(timeout: Optional[float] = None) -> mcp.client.mcp_client.MCPClient :async: Connect to the MCP server. :param timeout: Connection timeout (uses default if None) :returns: Connected MCP client :raises MCPConnectionError: If connection fails .. py:method:: disconnect() -> None :async: Disconnect from the MCP server. .. py:method:: get_client() -> mcp.client.mcp_client.MCPClient Get the MCP client if connected. :returns: MCP client instance :raises MCPConnectionError: If not connected .. py:method:: get_info() -> ConnectionInfo Get connection information. :returns: ConnectionInfo object with current state .. py:method:: health_check() -> Dict[str, Any] :async: Perform a health check on the connection. :returns: Health check results .. py:method:: is_connected() -> bool Check if connection is active. :returns: True if connected, False otherwise .. py:method:: is_healthy(threshold: float = 0.7) -> bool Check if connection is healthy. :param threshold: Health score threshold (0.0-1.0) :returns: True if health score is above threshold .. py:method:: reconnect() -> Optional[mcp.client.mcp_client.MCPClient] :async: Attempt to reconnect to the server. :returns: Connected client if successful, None if failed .. py:method:: start_monitoring() -> None :async: Start background health monitoring. .. py:method:: stop_monitoring() -> None :async: Stop background health monitoring. .. py:attribute:: auto_reconnect :value: True .. py:attribute:: client :type: Optional[mcp.client.mcp_client.MCPClient] :value: None .. py:attribute:: connection_count :value: 0 .. py:attribute:: connection_timeout :value: 30.0 .. py:attribute:: error_count :value: 0 .. py:attribute:: health_check_interval :value: 30.0 .. py:attribute:: health_score :value: 1.0 .. py:attribute:: last_connected :type: Optional[float] :value: None .. py:attribute:: last_error :type: Optional[str] :value: None .. py:attribute:: last_operation_time :value: 0.0 .. py:attribute:: max_reconnect_attempts :value: 5 .. py:attribute:: max_reconnect_delay :value: 60.0 .. py:attribute:: name .. py:attribute:: reconnect_attempts :value: 0 .. py:attribute:: reconnect_delay :value: 1.0 .. py:attribute:: status .. py:attribute:: total_uptime :value: 0.0 .. py:attribute:: transport .. 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:: MCPTransport(timeout: float = 30.0) Bases: :py:obj:`abc.ABC` Abstract base class for MCP transports. All MCP transports must implement this interface to provide a consistent API for the protocol layer. The transport handles the low-level communication while the protocol layer handles MCP message semantics. The transport is responsible for: - Connection establishment and teardown - Raw message sending and receiving - Transport-specific error handling - Connection state management .. py:method:: __aenter__() :async: Async context manager entry. .. py:method:: __aexit__(exc_type, exc_val, exc_tb) :async: Async context manager exit. .. py:method:: connect() -> None :abstractmethod: :async: Establish connection to the MCP server. :raises MCPConnectionError: If connection fails :raises MCPTimeoutError: If connection times out .. py:method:: disconnect() -> None :abstractmethod: :async: Close connection to the MCP server. Should be idempotent and safe to call multiple times. .. py:method:: receive_message() -> Dict[str, Any] :abstractmethod: :async: Receive a message from the server. :returns: JSON-RPC message from server :raises MCPTransportError: If receiving fails :raises MCPConnectionError: If not connected :raises MCPTimeoutError: If receive times out .. py:method:: send_message(message: Dict[str, Any]) -> None :abstractmethod: :async: Send a message to the server. :param message: JSON-RPC message to send :raises MCPTransportError: If sending fails :raises MCPConnectionError: If not connected .. py:attribute:: connected :value: False .. py:attribute:: timeout :value: 30.0 .. py:class:: SseTransport(sse_url: str, post_url: str, headers: Optional[Dict[str, str]] = None, timeout: float = 30.0) Bases: :py:obj:`MCPTransport` Server-Sent Events transport for MCP communication. This transport uses SSE for receiving messages and HTTP POST for sending. Useful for streaming scenarios where the server needs to push messages to the client. .. py:method:: connect() -> None :async: Establish SSE connection. .. py:method:: disconnect() -> None :async: Close SSE connection. .. py:method:: receive_message() -> Dict[str, Any] :async: Receive message from SSE stream. .. py:method:: send_message(message: Dict[str, Any]) -> None :async: Send message via HTTP POST. .. py:attribute:: headers .. py:attribute:: post_url .. py:attribute:: session :type: Optional[aiohttp.ClientSession] :value: None .. py:attribute:: sse_url .. py:class:: StdioTransport(command: str, args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, cwd: Optional[str] = None, timeout: float = 30.0) Bases: :py:obj:`MCPTransport` STDIO transport for MCP communication. This transport communicates with an MCP server via stdin/stdout of a subprocess. This is the most common transport for MCP servers that are designed to run as command-line tools. The transport handles: - Process lifecycle management - JSON-RPC message framing over stdio - Process cleanup on disconnection - Error handling for process failures .. rubric:: Examples Basic usage:: transport = StdioTransport( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"] ) With environment variables:: transport = StdioTransport( command="python", args=["-m", "my_mcp_server"], env={"API_KEY": "secret"}, cwd="/path/to/server" ) .. py:method:: connect() -> None :async: Start the MCP server process and establish stdio communication. .. py:method:: disconnect() -> None :async: Stop the MCP server process and cleanup resources. .. py:method:: receive_message() -> Dict[str, Any] :async: Receive JSON-RPC message from server via stdout. .. py:method:: send_message(message: Dict[str, Any]) -> None :async: Send JSON-RPC message to server via stdin. .. py:attribute:: args :value: [] .. py:attribute:: command .. py:attribute:: cwd :value: None .. py:attribute:: env :value: None .. py:attribute:: process :type: Optional[asyncio.subprocess.Process] :value: None .. py:class:: WebSocketTransport(url: str, headers: Optional[Dict[str, str]] = None, timeout: float = 30.0) Bases: :py:obj:`MCPTransport` WebSocket transport for MCP communication. This transport provides real-time bidirectional communication over WebSocket. Useful for interactive applications and real-time scenarios. .. py:method:: connect() -> None :async: Establish WebSocket connection. .. py:method:: disconnect() -> None :async: Close WebSocket connection. .. py:method:: receive_message() -> Dict[str, Any] :async: Receive message from WebSocket. .. py:method:: send_message(message: Dict[str, Any]) -> None :async: Send message over WebSocket. .. py:attribute:: headers .. py:attribute:: session :type: Optional[aiohttp.ClientSession] :value: None .. py:attribute:: url .. py:attribute:: websocket :type: Optional[aiohttp.ClientWebSocketResponse] :value: None