haive.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
Docker: Communication via stdin/stdout with a Docker container

The transport layer is responsible for:

Establishing and managing connections
Sending and receiving raw messages
Handling transport-specific encoding/decoding
Managing connection lifecycle

Classes¶

`DockerTransport`	Docker transport for MCP communication.
`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 haive.mcp.client.transport.DockerTransport(image, env=None, volumes=None, ports=None, network=None, docker_args=None, timeout=30.0)[source]¶

Bases: MCPTransport

Docker transport for MCP communication.

This transport runs an MCP server inside a Docker container and communicates via stdin/stdout, similar to StdioTransport but using docker run as the process wrapper. This provides isolation, reproducibility, and simplified dependency management for MCP servers.

The transport handles:

Docker container lifecycle management
Volume mounting for data access
Port mapping for network services
Environment variable passing to the container
Graceful container shutdown with docker stop

Examples

Basic Docker transport:

transport = DockerTransport(
    image="mcp/postgres",
    env={"POSTGRES_HOST": "host.docker.internal"},
)

With volumes and network:

transport = DockerTransport(
    image="mcp/filesystem",
    volumes=["/home/user/data:/data:ro"],
    network="host",
    env={"ALLOWED_DIRS": "/data"},
)

Initialize Docker transport.

Parameters:

image (str) – Docker image name (e.g., "mcp/postgres")
env (Optional[Dict[str, str]]) – Environment variables to pass to the container
volumes (Optional[List[str]]) – Volume mounts (e.g., ["/host/path:/container/path:ro"])
ports (Optional[List[str]]) – Port mappings (e.g., ["8080:8080"])
network (Optional[str]) – Docker network to attach to (e.g., "host")
docker_args (Optional[List[str]]) – Extra arguments to pass to docker run
timeout (float) – Timeout for operations in seconds

async connect()[source]¶

Start the Docker container and establish stdio communication.

Return type:: None

async disconnect()[source]¶

Stop the Docker container and clean up resources.

Return type:: None

async receive_message()[source]¶

Receive JSON-RPC message from the container via stdout.

Return type:: Dict[str, Any]

async send_message(message)[source]¶

Send JSON-RPC message to the container via stdin.

Parameters:: message (Dict[str, Any])
Return type:: None

class haive.mcp.client.transport.HttpTransport(url, headers=None, timeout=30.0)[source]¶

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"}
)

Initialize HTTP transport.

Parameters:

url (str) – Base URL for the MCP server
headers (Optional[Dict[str, str]]) – HTTP headers to include in requests
timeout (float) – Request timeout in seconds

async connect()[source]¶

Create HTTP session.

Return type:: None

async disconnect()[source]¶

Close HTTP session.

Return type:: None

async receive_message()[source]¶

Send pending message and receive response.

Return type:: Dict[str, Any]

async send_message(message)[source]¶

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.

Parameters:: message (Dict[str, Any])
Return type:: None

class haive.mcp.client.transport.MCPTransport(timeout=30.0)[source]¶

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

Initialize transport with timeout.

Parameters:: timeout (float) – Default timeout for operations in seconds

abstractmethod connect()[source]¶

Async:
Return type:: None

Establish connection to the MCP server.

Raises:

MCPConnectionError – If connection fails
MCPTimeoutError – If connection times out

Return type:

None

abstractmethod disconnect()[source]¶

Async:
Return type:: None

Close connection to the MCP server.

Should be idempotent and safe to call multiple times.

abstractmethod receive_message()[source]¶

Async:
Return type:: Dict[str, Any]

Receive a message from the server.

Returns:

JSON-RPC message from server

Raises:

MCPTransportError – If receiving fails
MCPConnectionError – If not connected
MCPTimeoutError – If receive times out

Return type:

Dict[str, Any]

abstractmethod send_message(message)[source]¶

Async:
Parameters:: message (Dict[str, Any])
Return type:: None

Send a message to the server.

Parameters:

message (Dict[str, Any]) – JSON-RPC message to send

Raises:

MCPTransportError – If sending fails
MCPConnectionError – If not connected

Return type:

None

class haive.mcp.client.transport.SseTransport(sse_url, post_url, headers=None, timeout=30.0)[source]¶

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.

Initialize SSE transport.

Parameters:

sse_url (str) – URL for SSE stream (receiving messages)
post_url (str) – URL for POST requests (sending messages)
headers (Optional[Dict[str, str]]) – HTTP headers for both requests
timeout (float) – Request timeout

async connect()[source]¶

Establish SSE connection.

Return type:: None

async disconnect()[source]¶

Close SSE connection.

Return type:: None

async receive_message()[source]¶

Receive message from SSE stream.

Return type:: Dict[str, Any]

async send_message(message)[source]¶

Send message via HTTP POST.

Parameters:: message (Dict[str, Any])
Return type:: None

class haive.mcp.client.transport.StdioTransport(command, args=None, env=None, cwd=None, timeout=30.0)[source]¶

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"
)

Initialize STDIO transport.

Parameters:

command (str) – Command to execute (e.g., “npx”, “python”)
args (Optional[List[str]]) – Command line arguments
env (Optional[Dict[str, str]]) – Environment variables for the process
cwd (Optional[str]) – Working directory for the process
timeout (float) – Timeout for operations

async connect()[source]¶

Start the MCP server process and establish stdio communication.

Return type:: None

async disconnect()[source]¶

Stop the MCP server process and cleanup resources.

Return type:: None

async receive_message()[source]¶

Receive JSON-RPC message from server via stdout.

Return type:: Dict[str, Any]

async send_message(message)[source]¶

Send JSON-RPC message to server via stdin.

Parameters:: message (Dict[str, Any])
Return type:: None

class haive.mcp.client.transport.WebSocketTransport(url, headers=None, timeout=30.0)[source]¶

Bases: MCPTransport

WebSocket transport for MCP communication.

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

Initialize WebSocket transport.

Parameters:

url (str) – WebSocket URL (ws:// or wss://)
headers (Optional[Dict[str, str]]) – WebSocket headers
timeout (float) – Connection timeout

async connect()[source]¶

Establish WebSocket connection.

Return type:: None

async disconnect()[source]¶

Close WebSocket connection.

Return type:: None

async receive_message()[source]¶

Receive message from WebSocket.

Return type:: Dict[str, Any]

async send_message(message)[source]¶

Send message over WebSocket.

Parameters:: message (Dict[str, Any])
Return type:: None