MCP Transports

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

MCP Transports

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

Transport Types

Quick Start

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:

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

# Multi-word command string with extra runtime args
agent = Agent(
    tools=MCP(
        "npx -y @modelcontextprotocol/server-filesystem",
        args=["/tmp"],
    )
)

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

Fail-closed on empty or timed-out servers

Since PraisonAI #4423, MCP(...) fails closed on the stdio path — a server that times out, errors on init, or lists zero tools raises TimeoutError or RuntimeError at construction time instead of returning a usable-but-empty object. See MCP → Errors on construction. SSE, Streamable HTTP, and WebSocket transports are unchanged.

Parallel Tool Calls

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

Concurrent calls to one stdio server also run concurrently on it: each request is dispatched as its own task on the MCP session, so a slow call never holds up the others and timeout= is charged from when a call starts, not from when it was queued. When the runner shuts down, in-flight calls are cancelled and return MCP <kind> call cancelled: runner shut down instead of blocking (PR #5796, fixes #5795).

from praisonaiagents import Agent, ExecutionConfig, MCP

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

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

Timeouts

The timeout parameter on MCP() bounds every call issued via the MCP session — tool calls, resource reads (read_resource), and prompt fetches (get_prompt). When you don’t pass an explicit value, the default is:

• 60s for URL transports (ws/http/sse) and non-launcher stdio commands,
• 180s for cold-start package launchers (npx, uvx, bunx, pnpm, yarn, including the Windows .cmd/.exe shims and full paths ending in one) — their first run downloads the server package before the MCP handshake can begin (PraisonAI #5110).

An explicit MCP(..., timeout=N) overrides both defaults.

Timed-out calls never run later. A timed-out call is also abandoned: the caller marks the queued request cancelled and the runner skips it, so the tool will not execute after you received the timeout error (PraisonAI #5772). Side effects like sending an email or calling an API will not fire after the timeout.

The timeout is enforced at two layers:

• Server-side: every session call is wrapped in asyncio.wait_for(...), so a misbehaving server can only stall one request. On timeout the response is "MCP {kind} call timed out after {N} seconds (server side)" where {kind} is tool, resource, or prompt.
• Caller-side: the response queue also uses get(timeout=...), so a caller that never receives a reply sees "Error: MCP tool call timed out after {N} seconds". The agent runtime normalizes this into {"error": "MCP tool call timed out after N seconds", "timeout": True} before returning it to the tool-execution loop, so MCP timeouts participate in retry, doom-loop detection, and circuit-breaking instead of silently looking like a successful string result.

As of PraisonAI PR #4634, this normalization applies on both the sync tool loop (agent.chat() / agent.start()) and the async tool loop (agent.achat() / agent.astart()). Earlier releases did not consult MCP on the async path at all — an MCP tool call over achat() hard-failed as “Function … not found in tools” and an MCP timeout on achat() never produced the normalized {"error": ..., "timeout": True} dict. The two paths now share a _resolve_mcp_tool_result / _normalize_mcp_result implementation, so classification is symmetric.

timeout= also bounds the initial MCP handshake — initialize() waits at most timeout seconds before failing with "Error: MCP initialization timed out after {N} seconds". That string is normalized to the same {"error": ..., "timeout": True} shape by the agent runtime.

Since PraisonAI #4423, the construction-time stdio handshake raises TimeoutError at the MCP.__init__ layer (before any normalization can happen), so callers who construct MCP() inline must catch it directly — see MCP → Errors on construction.

A first npx/uvx run downloads the package before the handshake, which is why the effective init timeout auto-extends to 180s.

If the cold download runs past 180s, construction raises with a pre-warm hint.

Streamable HTTP Transport

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 mcp >= 1.20.0, < 2. mcp 2.0.0 removed streamablehttp_client entirely, so do not upgrade to mcp 2.x — see PraisonAI #5706.

# Fresh install — the praisonaiagents[mcp] extra already pins mcp<2.
pip install "praisonaiagents[mcp]"

# Already on mcp 2.x? Downgrade:
pip install "mcp<2"

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
• ✅ Structured error dicts on timeout or transport failure (see Error handling below)
• ✅ 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:

Error handling

Failures return structured dicts, so retry, circuit breaker, and ON_RETRY engage with no config.

# Timeout
{"error": "MCP tool call timed out after <timeout> seconds", "timeout": True, "mcp_error": True}

# Other failure (message falls back to just <ExcType> if empty)
{"error": "MCP tool '<name>' failed: <ExcType>: <msg>", "mcp_error": True}

See MCP → remote tool failures.

Troubleshooting

pip install "mcp<2"

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

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

WebSocket Transport

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

Error handling

Failures return structured dicts, so retry, circuit breaker, and ON_RETRY engage with no config.

# Timeout
{"error": "MCP tool call timed out after <timeout> seconds", "timeout": True, "mcp_error": True}

# Other failure (message falls back to just <ExcType> if empty)
{"error": "MCP tool '<name>' failed: <ExcType>: <msg>", "mcp_error": True}

See MCP → remote tool failures.

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)

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

Error handling

Failures return structured dicts, so retry, circuit breaker, and ON_RETRY engage with no config.

# Timeout
{"error": "MCP tool call timed out after <timeout> seconds", "timeout": True, "mcp_error": True}

# Other failure (message falls back to just <ExcType> if empty)
{"error": "MCP tool '<name>' failed: <ExcType>: <msg>", "mcp_error": True}

See MCP → remote tool failures.

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

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

Next Steps