mcp.client.transport ==================== .. py:module:: mcp.client.transport .. autoapi-nested-parse:: MCP Transport Layer Implementation. This module provides transport layer implementations for the MCP protocol. Different transports handle the underlying communication mechanism between the MCP client and server. Supported Transports: - STDIO: Communication via stdin/stdout with a subprocess - HTTP: RESTful communication over HTTP - SSE: Server-sent events for streaming - WebSocket: Real-time bidirectional communication The transport layer is responsible for: - Establishing and managing connections - Sending and receiving raw messages - Handling transport-specific encoding/decoding - Managing connection lifecycle Attributes ---------- .. autoapisummary:: mcp.client.transport.logger Classes ------- .. autoapisummary:: mcp.client.transport.HttpTransport mcp.client.transport.MCPTransport mcp.client.transport.SseTransport mcp.client.transport.StdioTransport mcp.client.transport.WebSocketTransport Module Contents --------------- .. 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:: 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 .. py:data:: logger