> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# MCP Transports

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

# MCP Transports

<Info>**Protocol Revision**: 2025-11-25</Info>

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

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    subgraph "PraisonAI Agent"
        A[🤖 Agent]
    end
    
    subgraph "MCP Module"
        MCP[📡 MCP Class]
    end
    
    subgraph "Transport Layer"
        STDIO[📟 stdio]
        HTTP[🌐 Streamable HTTP]
        WS[🔌 WebSocket]
        SSE[📡 SSE Legacy]
    end
    
    subgraph "MCP Servers"
        S1[Local Server]
        S2[Remote Server]
        S3[Cloud Server]
    end
    
    A --> MCP
    MCP --> STDIO
    MCP --> HTTP
    MCP --> WS
    MCP --> SSE
    
    STDIO --> S1
    HTTP --> S2
    WS --> S3
    SSE --> S2
    
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef transport fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef server fill:#2E8B57,stroke:#7C90A0,color:#fff
    
    class A agent
    class MCP,STDIO,HTTP,WS,SSE transport
    class S1,S2,S3 server
```

## Transport Types

<CardGroup cols={2}>
  <Card title="stdio" icon="terminal">
    **Local subprocess communication**

    * Best for local development
    * NPX packages, Python scripts
    * Newline-delimited JSON-RPC
  </Card>

  <Card title="Streamable HTTP" icon="globe">
    **HTTP POST/GET with SSE streaming**

    * Production deployments
    * Session management
    * Resumable connections
  </Card>

  <Card title="WebSocket" icon="plug">
    **Bidirectional real-time**

    * Long-lived connections
    * Cloud-native (hibernatable)
    * Lower protocol overhead
  </Card>

  <Card title="SSE (Legacy)" icon="tower-broadcast">
    **Server-Sent Events**

    * Backward compatibility
    * Protocol version 2024-11-05
    * URLs ending in `/sse`
  </Card>
</CardGroup>

## Quick Start

<CodeGroup>
  ```python stdio Transport theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  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")
  )
  ```

  ```python Streamable HTTP theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, MCP

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

  # With session management
  agent = Agent(
      tools=MCP(
          "https://api.example.com/mcp",
          session=True
      )
  )
  ```

  ```python WebSocket theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  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"
      )
  )
  ```

  ```python SSE Legacy theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, MCP

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

## 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://` or `https://`          | Streamable HTTP | `https://api.example.com/mcp` |
| Command string                   | stdio           | `npx @mcp/server-memory`      |

***

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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant Agent
    participant MCPRunner
    participant Queue1 as Response Queue 1
    participant Queue2 as Response Queue 2
    participant MCPWorker as MCP Worker Thread
    participant MCPServer as MCP Server
    
    Agent->>MCPRunner: call_tool("search", {"query": "X"})
    Agent->>MCPRunner: call_tool("search", {"query": "Y"})
    
    MCPRunner->>Queue1: create queue.Queue(maxsize=1)
    MCPRunner->>Queue2: create queue.Queue(maxsize=1)
    
    MCPRunner->>MCPWorker: submit("search", {"query": "X"}, Queue1)
    MCPRunner->>MCPWorker: submit("search", {"query": "Y"}, Queue2)
    
    MCPWorker->>MCPServer: search X
    MCPWorker->>MCPServer: search Y
    
    MCPServer-->>MCPWorker: result X
    MCPServer-->>MCPWorker: result Y
    
    MCPWorker->>Queue1: put(result X)
    MCPWorker->>Queue2: put(result Y)
    
    Queue1-->>MCPRunner: get(timeout=60) → result X
    Queue2-->>MCPRunner: get(timeout=60) → result Y
    
    MCPRunner-->>Agent: return result X
    MCPRunner-->>Agent: return result Y
```

***

## Streamable HTTP Transport

<Info>This is the current standard transport (Protocol Revision 2025-11-25)</Info>

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

### Features

* ✅ Single MCP endpoint for all communication
* ✅ Session management via `Mcp-Session-Id` header
* ✅ Protocol versioning via `Mcp-Protocol-Version` header
* ✅ SSE streaming with resumability
* ✅ `Last-Event-ID` for connection recovery
* ✅ Session termination via HTTP DELETE

### Usage

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, MCP

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

# With all options
agent = Agent(
    tools=MCP(
        "https://api.example.com/mcp",
        timeout=60,
        debug=True,
        headers={"Authorization": "Bearer token"},
        session=True,           # Enable session management
        resumability=True       # Enable SSE resumability
    )
)
```

### Session Management

Sessions allow stateful interactions with MCP servers:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import MCP, SessionManager

# Session is automatically managed
mcp = MCP("https://api.example.com/mcp")

# Access session info
print(f"Session ID: {mcp.http_stream_client.transport.session_id}")

# Session is terminated on cleanup or explicitly
# mcp.http_stream_client.transport.terminate_session()
```

### Protocol Version

The client automatically includes `Mcp-Protocol-Version` header:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Default version: 2025-03-26 (for backward compatibility)
mcp = MCP("https://api.example.com/mcp")

# Specify version explicitly
mcp = MCP(
    "https://api.example.com/mcp",
    protocol_version="2025-11-25"
)
```

***

## WebSocket Transport

<Note>Based on SEP-1288 (in review)</Note>

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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)

<Warning>This transport is deprecated. Use Streamable HTTP for new implementations.</Warning>

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

### Usage

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

<CardGroup cols={2}>
  <Card title="Transport Selection" icon="route">
    * **stdio**: Local development, NPX packages
    * **Streamable HTTP**: Production web services
    * **WebSocket**: Real-time, cloud-native apps
    * **SSE**: Legacy server compatibility
  </Card>

  <Card title="Session Management" icon="key">
    * Enable sessions for stateful operations
    * Handle HTTP 404 (session expired)
    * Terminate sessions explicitly when done
  </Card>

  <Card title="Resumability" icon="rotate">
    * Enable for long-running operations
    * Track Last-Event-ID for recovery
    * Respect server retry delays
  </Card>

  <Card title="Security" icon="shield">
    * Validate Origin headers
    * Use HTTPS/WSS in production
    * Implement proper authentication
    * Bind to localhost for local servers
  </Card>
</CardGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="MCP Tools" icon="wrench" href="/docs/mcp/mcp-tools">
    Learn about MCP tool integration
  </Card>

  <Card title="MCP Servers" icon="server" href="/docs/mcp/mcp-server">
    Explore available MCP servers
  </Card>

  <Card title="SSE Transport Details" icon="tower-broadcast" href="/docs/mcp/sse-transport">
    Deep dive into SSE transport
  </Card>

  <Card title="Custom MCP Servers" icon="code" href="/docs/mcp/custom">
    Build your own MCP servers
  </Card>
</CardGroup>