MCP Transports

Complete guide to MCP transport mechanisms - stdio, Streamable HTTP, WebSocket, and SSE

MCP Transports

Protocol Revision: 2025-11-25

PraisonAI Agents supports all MCP transport mechanisms for connecting to MCP servers, providing flexibility for different deployment scenarios.

Transport Types

stdio

Local subprocess communication

• Best for local development
• NPX packages, Python scripts
• Newline-delimited JSON-RPC

Streamable HTTP

HTTP POST/GET with SSE streaming

• Production deployments
• Official SDK streamablehttp_client
• Auth via custom headers

WebSocket

Bidirectional real-time

• Long-lived connections
• Cloud-native (hibernatable)
• Lower protocol overhead

SSE (Legacy)

Server-Sent Events

• Backward compatibility
• Protocol version 2024-11-05
• URLs ending in /sse

Quick Start

stdio Transport

from praisonaiagents import Agent, MCP

# Local NPX server
agent = Agent(
    name="Assistant",
    tools=MCP("npx @modelcontextprotocol/server-memory")
)

# Python script
agent = Agent(
    tools=MCP("python3 mcp_server.py")
)

from praisonaiagents import Agent, MCP

# HTTP endpoint (auto-detected)
agent = Agent(
    tools=MCP("https://api.example.com/mcp")
)

# With auth headers and timeout
agent = Agent(
    tools=MCP(
        "https://api.example.com/mcp",
        headers={"Authorization": "Bearer YOUR_TOKEN"},
        timeout=60
    )
)

from praisonaiagents import Agent, MCP

# WebSocket endpoint (auto-detected via ws://)
agent = Agent(
    tools=MCP("ws://localhost:8080/mcp")
)

# Secure WebSocket
agent = Agent(
    tools=MCP(
        "wss://api.example.com/mcp",
        auth_token="your-token"
    )
)

from praisonaiagents import Agent, MCP

# Legacy SSE endpoint (auto-detected via /sse suffix)
agent = Agent(
    tools=MCP("http://localhost:8080/sse")
)

Transport Auto-Detection

The MCP class automatically selects the appropriate transport based on the URL:

URL Pattern	Transport	Example
ws:// or wss://	WebSocket	ws://localhost:8080/mcp
http(s)://... ending in /sse	SSE (Legacy)	http://localhost:8080/sse
http(s)://<host><path> (path required)	Streamable HTTP	https://api.example.com/mcp
Command string	stdio	npx @mcp/server-memory

Streamable HTTP: the URL is used verbatim. Include whatever endpoint path your server exposes (commonly /mcp); a bare-host URL will POST to / and fail with Session terminated.

---

stdio Transport

The stdio transport runs MCP servers as subprocesses, communicating via standard input/output.

Features

• ✅ JSON-RPC messages over stdin/stdout
• ✅ Newline-delimited messages
• ✅ UTF-8 encoding enforced
• ✅ Environment variable support

Usage

from praisonaiagents import Agent, MCP

# Simple command string
agent = Agent(
    tools=MCP("npx @modelcontextprotocol/server-filesystem")
)

# With separate command and args
agent = Agent(
    tools=MCP(
        command="python3",
        args=["server.py", "--config", "prod.json"]
    )
)

# With environment variables
agent = Agent(
    tools=MCP(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_TOKEN": "ghp_xxx"}
    )
)

# With timeout and debug
agent = Agent(
    tools=MCP(
        "npx @modelcontextprotocol/server-memory",
        timeout=120,
        debug=True
    )
)

Parallel Tool Calls

MCP stdio tool calls are safe to run in parallel — each call gets its own response channel, so results never cross.

from praisonaiagents import Agent, MCP
from praisonaiagents.config.feature_configs import ToolConfig

agent = Agent(
    name="Researcher",
    instructions="Use the MCP tools in parallel when helpful.",
    tools=MCP("npx @modelcontextprotocol/server-memory", timeout=60),
    tool_config=ToolConfig(parallel=True),
)

agent.start("Look up X and Y at the same time")

The timeout parameter on MCP() controls the per-call wait time (default 60s). On timeout, the tool result is the string "Error: MCP tool call timed out after {N} seconds" — the agent sees this as a tool error and can retry. The timeout also bounds the initial MCP handshake.

---

Streamable HTTP Transport

This is the current standard transport (Protocol Revision 2025-11-25)

The Streamable HTTP transport uses HTTP POST for sending messages and supports SSE streaming for responses.

Requirements

The Streamable HTTP transport is powered by the official MCP SDK’s streamablehttp_client (httpx-based). It requires an up-to-date mcp package.

pip install -U 'mcp'

If the installed mcp package is too old, HTTPStreamMCPClient raises at construction time:

The installed 'mcp' package does not provide the Streamable HTTP client
(mcp.client.streamable_http.streamablehttp_client).
Upgrade it with: pip install -U 'mcp'

Fix it by upgrading: pip install -U 'mcp'.

Features

• ✅ Single MCP endpoint for all communication
• ✅ Powered by the official mcp.client.streamable_http.streamablehttp_client (httpx, not aiohttp)
• ✅ Session lifecycle managed by the official SDK client
• ✅ Custom headers for authentication
• ✅ Clean context unwinding — a failed initialize() handshake unwinds already-entered contexts, so no httpx/anyio connection or half-open remote session is leaked

Usage

The URL is used exactly as provided; there is no client-side rewriting. Pass the full endpoint your server exposes (commonly .../mcp).

from praisonaiagents import Agent, MCP

# Basic HTTP endpoint
agent = Agent(
    tools=MCP("https://api.example.com/mcp")
)

# With auth headers and timeout
agent = Agent(
    tools=MCP(
        "https://api.example.com/mcp",
        headers={"Authorization": "Bearer YOUR_TOKEN"},
        timeout=60,
        debug=True
    )
)

Supported Options

The HTTP-Stream transport reads only these two options:

Option	Type	Default	Description
headers	dict[str, str]	None	Extra headers forwarded to streamablehttp_client (e.g. {"Authorization": "Bearer ..."})
timeout	int	60	Passed to streamablehttp_client and used as the per-call bound

session, resumability, cors, and responseMode are still accepted as keyword arguments but are no-ops on this transport — session and resumability are fully managed by the official SDK client. Use headers and timeout only.

Troubleshooting

ImportError: streamablehttp_client not found

Your mcp package predates the Streamable HTTP client. Upgrade it:

pip install -U 'mcp'

session=True / resumability=True have no effect

These options are no-ops on the HTTP-Stream transport. Session and resumability are managed internally by the official streamablehttp_client. Remove them — pass headers and timeout instead.

AttributeError on http_stream_client.transport

The internal transport attribute is now None; session management is owned by the official SDK client. Do not access http_stream_client.transport.session_id or .terminate_session().

McpError: Session terminated on initialize

The Streamable HTTP transport posts to the URL exactly as you provided. It does not silently append /mcp to a bare-host URL any more (this convenience was removed in the fix for PraisonAI #3032). If your server exposes tools at https://server.example.com/mcp, pass that full URL — a bare https://server.example.com will POST to /, get HTTP 404, and the SDK will surface it as McpError: Session terminated during session.initialize().

# ❌ Old convenience — no longer works
MCP("https://server.example.com")

# ✅ Pass the full endpoint URL
MCP("https://server.example.com/mcp")

---

WebSocket Transport

Based on SEP-1288 (in review)

WebSocket transport provides bidirectional, long-lived connections ideal for cloud deployments.

Features

• ✅ ws:// and wss:// URL detection
• ✅ JSON-RPC message framing
• ✅ Session ID handling
• ✅ Reconnection with exponential backoff
• ✅ Authentication token support
• ✅ Ping/pong keepalive

Usage

from praisonaiagents import Agent, MCP

# Basic WebSocket
agent = Agent(
    tools=MCP("ws://localhost:8080/mcp")
)

# Secure WebSocket with authentication
agent = Agent(
    tools=MCP(
        "wss://api.example.com/mcp",
        auth_token="Bearer your-secret-token",
        timeout=60
    )
)

Advanced WebSocket Options

from praisonaiagents.mcp.mcp_websocket import WebSocketMCPClient

# Direct client usage with all options
client = WebSocketMCPClient(
    server_url="wss://api.example.com/mcp",
    auth_token="your-token",
    timeout=60,
    options={
        "ping_interval": 30,
        "ping_timeout": 10,
        "max_retries": 5
    }
)

# Use with agent
agent = Agent(tools=list(client.tools))

Reconnection

WebSocket transport automatically reconnects with exponential backoff:

from praisonaiagents.mcp.mcp_websocket import calculate_backoff

# Backoff calculation: base_delay * 2^attempt, capped at max_delay
# Attempt 0: 1s, Attempt 1: 2s, Attempt 2: 4s, ... max 60s

---

SSE Transport (Legacy)

This transport is deprecated. Use Streamable HTTP for new implementations.

The SSE transport is maintained for backward compatibility with servers using protocol version 2024-11-05.

Usage

from praisonaiagents import Agent, MCP

# URLs ending in /sse use legacy SSE transport
agent = Agent(
    tools=MCP("http://localhost:8080/sse")
)

---

Security

Origin Validation (DNS Rebinding Prevention)

from praisonaiagents.mcp.mcp_security import SecurityConfig, is_valid_origin

# Configure allowed origins
config = SecurityConfig(
    allowed_origins=["localhost", "127.0.0.1", "example.com"],
    validate_origin=True
)

# Validate origin header
is_valid = is_valid_origin(
    origin="https://example.com",
    allowed_origins=config.allowed_origins
)

Authentication

from praisonaiagents.mcp.mcp_security import create_auth_header

# Bearer token
headers = create_auth_header("your-token", auth_type="bearer")
# {"Authorization": "Bearer your-token"}

# Basic auth
headers = create_auth_header("user:pass", auth_type="basic")
# {"Authorization": "Basic dXNlcjpwYXNz"}

# Custom header
headers = create_auth_header("api-key", auth_type="custom", header_name="X-API-Key")
# {"X-API-Key": "api-key"}

Secure Session IDs

from praisonaiagents.mcp.mcp_security import generate_secure_session_id

# Generate cryptographically secure session ID
session_id = generate_secure_session_id(length=32)
# Returns URL-safe base64 string with visible ASCII only

---

Backward Compatibility

Supporting Older Servers

The MCP class automatically handles backward compatibility:

from praisonaiagents import MCP
from praisonaiagents.mcp.mcp_compat import detect_transport_support

# Auto-detection handles:
# 1. Try Streamable HTTP first (POST to endpoint)
# 2. On 400/404/405, fall back to legacy SSE
# 3. URLs ending in /sse use legacy transport directly

mcp = MCP("https://api.example.com/mcp")  # Auto-negotiates

Protocol Versions Supported

Version	Transport	Status
2024-11-05	HTTP+SSE	Legacy (supported)
2025-03-26	Streamable HTTP	Supported
2025-06-18	Streamable HTTP	Supported
2025-11-25	Streamable HTTP	Current

---

Custom Transports

You can implement custom transports by extending the BaseTransport class:

from praisonaiagents.mcp.mcp_transport import BaseTransport, TransportRegistry

class MyCustomTransport(BaseTransport):
    async def connect(self):
        """Establish connection"""
        pass
    
    async def send(self, message: dict):
        """Send JSON-RPC message"""
        pass
    
    async def receive(self) -> dict:
        """Receive JSON-RPC message"""
        pass
    
    async def close(self):
        """Close connection"""
        pass
    
    @property
    def is_connected(self) -> bool:
        return self._connected

# Register custom transport
registry = TransportRegistry()
registry.register("my_transport", MyCustomTransport)

---

Best Practices

Transport Selection

• stdio: Local development, NPX packages
• Streamable HTTP: Production web services
• WebSocket: Real-time, cloud-native apps
• SSE: Legacy server compatibility

Streamable HTTP

• Upgrade with pip install -U 'mcp'
• Pass headers for authentication
• Session lifecycle is handled by the SDK client

Timeouts

• Set timeout for slow servers
• The timeout bounds each tool call
• It also bounds the initial handshake

Security

• Validate Origin headers
• Use HTTPS/WSS in production
• Implement proper authentication
• Bind to localhost for local servers

Next Steps

MCP Tools

Learn about MCP tool integration

MCP Servers

Explore available MCP servers

SSE Transport Details

Deep dive into SSE transport

Custom MCP Servers

Build your own MCP servers