mcp.client.transport

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

logger

Classes

HttpTransport

HTTP transport for MCP communication.

MCPTransport

Abstract base class for MCP transports.

SseTransport

Server-Sent Events transport for MCP communication.

StdioTransport

STDIO transport for MCP communication.

WebSocketTransport

WebSocket transport for MCP communication.

Module Contents

class mcp.client.transport.HttpTransport(url: str, headers: Dict[str, str] | None = None, timeout: float = 30.0)

Bases: 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.

Examples

Basic HTTP transport:

transport = HttpTransport("http://localhost:8080/mcp")

With authentication:

transport = HttpTransport(
    "https://api.example.com/mcp",
    headers={"Authorization": "Bearer token123"}
)
async connect() → None

Create HTTP session.

async disconnect() → None

Close HTTP session.

async receive_message() → Dict[str, Any]

Send pending message and receive response.

async send_message(message: Dict[str, Any]) → None

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.

headers
session: aiohttp.ClientSession | None = None
url
class mcp.client.transport.MCPTransport(timeout: float = 30.0)

Bases: 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

async __aenter__()

Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)

Async context manager exit.

abstractmethod connect() → None
Async:

Establish connection to the MCP server.

Raises:
abstractmethod disconnect() → None
Async:

Close connection to the MCP server.

Should be idempotent and safe to call multiple times.

abstractmethod receive_message() → Dict[str, Any]
Async:

Receive a message from the server.

Returns:

JSON-RPC message from server

Raises:
abstractmethod send_message(message: Dict[str, Any]) → None
Async:

Send a message to the server.

Parameters:

message – JSON-RPC message to send

Raises:
connected = False
timeout = 30.0
class mcp.client.transport.SseTransport(sse_url: str, post_url: str, headers: Dict[str, str] | None = None, timeout: float = 30.0)

Bases: 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.

async connect() → None

Establish SSE connection.

async disconnect() → None

Close SSE connection.

async receive_message() → Dict[str, Any]

Receive message from SSE stream.

async send_message(message: Dict[str, Any]) → None

Send message via HTTP POST.

headers
post_url
session: aiohttp.ClientSession | None = None
sse_url
class mcp.client.transport.StdioTransport(command: str, args: List[str] | None = None, env: Dict[str, str] | None = None, cwd: str | None = None, timeout: float = 30.0)

Bases: 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

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"
)
async connect() → None

Start the MCP server process and establish stdio communication.

async disconnect() → None

Stop the MCP server process and cleanup resources.

async receive_message() → Dict[str, Any]

Receive JSON-RPC message from server via stdout.

async send_message(message: Dict[str, Any]) → None

Send JSON-RPC message to server via stdin.

args = []
command
cwd = None
env = None
process: asyncio.subprocess.Process | None = None
class mcp.client.transport.WebSocketTransport(url: str, headers: Dict[str, str] | None = None, timeout: float = 30.0)

Bases: MCPTransport

WebSocket transport for MCP communication.

This transport provides real-time bidirectional communication over WebSocket. Useful for interactive applications and real-time scenarios.

async connect() → None

Establish WebSocket connection.

async disconnect() → None

Close WebSocket connection.

async receive_message() → Dict[str, Any]

Receive message from WebSocket.

async send_message(message: Dict[str, Any]) → None

Send message over WebSocket.

headers
session: aiohttp.ClientSession | None = None
url
websocket: aiohttp.ClientWebSocketResponse | None = None
mcp.client.transport.logger