0
# Transport Mechanisms
1

2
Transport layer implementations for various connection types including stdio, HTTP, WebSocket, and streaming protocols. The MCP SDK supports multiple transport mechanisms with authentication and security features for different deployment scenarios.
3

4
## Capabilities
5

6
### Stdio Transport
7

8
Standard input/output transport for local process communication, ideal for command-line tools and local server execution.
9

10
```python { .api }
11
def stdio_client(server: StdioServerParameters) -> AsyncContextManager:
12
    """
13
    Create a stdio transport client connection.
14

15
    Parameters:
16
    - server: Stdio server parameters
17

18
    Returns:
19
    Async context manager yielding (read_stream, write_stream)
20
    """
21

22
def stdio_server() -> AsyncContextManager:
23
    """
24
    Create a stdio server transport.
25

26
    Returns:
27
    Async context manager yielding (read_stream, write_stream)
28
    """
29

30
class StdioServerParameters:
31
    def __init__(
32
        self,
33
        command: str | list[str],
34
        args: list[str] | None = None,
35
        env: dict[str, str] | None = None,
36
        cwd: str | None = None,
37
    ):
38
        """
39
        Parameters for stdio server connection.
40

41
        Parameters:
42
        - command: Server command to execute
43
        - args: Additional command arguments
44
        - env: Environment variables for server process
45
        - cwd: Working directory for server process
46
        """
47
```
48

49
### WebSocket Transport
50

51
Real-time bidirectional communication using WebSocket protocol for web applications and real-time scenarios.
52

53
```python { .api }
54
def websocket_client(
55
    url: str,
56
    headers: dict[str, str] | None = None,
57
    **kwargs
58
) -> AsyncContextManager:
59
    """
60
    Create a WebSocket client connection.
61

62
    Parameters:
63
    - url: WebSocket server URL (ws:// or wss://)
64
    - headers: HTTP headers for WebSocket handshake
65
    - **kwargs: Additional connection options
66

67
    Returns:
68
    Async context manager yielding (read_stream, write_stream)
69
    """
70

71
def websocket_server(
72
    host: str = "127.0.0.1",
73
    port: int = 8000,
74
    **kwargs
75
) -> AsyncContextManager:
76
    """
77
    Create a WebSocket server.
78

79
    Parameters:
80
    - host: Server host address
81
    - port: Server port number
82
    - **kwargs: Additional server options
83

84
    Returns:
85
    Async context manager for server lifecycle
86
    """
87
```
88

89
### Server-Sent Events (SSE) Transport
90

91
HTTP-based transport using Server-Sent Events for web applications requiring unidirectional streaming.
92

93
```python { .api }
94
def sse_client(
95
    url: str,
96
    headers: dict[str, str] | None = None,
97
    **kwargs
98
) -> AsyncContextManager:
99
    """
100
    Create a Server-Sent Events client connection.
101

102
    Parameters:
103
    - url: SSE endpoint URL
104
    - headers: HTTP headers for connection
105
    - **kwargs: Additional connection options
106

107
    Returns:
108
    Async context manager yielding (read_stream, write_stream)
109
    """
110

111
def sse_server(
112
    host: str = "127.0.0.1",
113
    port: int = 8000,
114
    **kwargs
115
) -> AsyncContextManager:
116
    """
117
    Create a Server-Sent Events server.
118

119
    Parameters:
120
    - host: Server host address  
121
    - port: Server port number
122
    - **kwargs: Additional server options
123

124
    Returns:
125
    Async context manager for server lifecycle
126
    """
127

128
class SseServerParameters:
129
    def __init__(
130
        self,
131
        url: str,
132
        headers: dict[str, str] | None = None,
133
        **kwargs
134
    ):
135
        """
136
        Parameters for Server-Sent Events connection.
137

138
        Parameters:
139
        - url: SSE endpoint URL
140
        - headers: HTTP headers
141
        - **kwargs: Additional options
142
        """
143
```
144

145
### HTTP Streaming Transport
146

147
Efficient HTTP streaming connections for high-throughput data transfer scenarios.
148

149
```python { .api }
150
def streamablehttp_client(
151
    url: str,
152
    headers: dict[str, str] | None = None,
153
    **kwargs
154
) -> AsyncContextManager:
155
    """
156
    Create a streaming HTTP client connection.
157

158
    Parameters:
159
    - url: HTTP endpoint URL
160
    - headers: HTTP headers for connection
161
    - **kwargs: Additional connection options
162

163
    Returns:
164
    Async context manager yielding (read_stream, write_stream)
165
    """
166

167
class StreamableHttpParameters:
168
    def __init__(
169
        self,
170
        url: str,
171
        headers: dict[str, str] | None = None,
172
        method: str = "POST",
173
        **kwargs
174
    ):
175
        """
176
        Parameters for streaming HTTP connection.
177

178
        Parameters:
179
        - url: HTTP endpoint URL
180
        - headers: HTTP headers
181
        - method: HTTP method
182
        - **kwargs: Additional options
183
        """
184
```
185

186
### Transport Security
187

188
Security settings and utilities for protecting transport connections.
189

190
```python { .api }
191
class TransportSecuritySettings:
192
    def __init__(
193
        self,
194
        dns_rebinding_protection: bool = True,
195
        allowed_origins: list[str] | None = None,
196
        **kwargs
197
    ):
198
        """
199
        Transport security configuration.
200

201
        Parameters:
202
        - dns_rebinding_protection: Enable DNS rebinding protection
203
        - allowed_origins: List of allowed origins for CORS
204
        - **kwargs: Additional security options
205
        """
206
```
207

208
## Usage Examples
209

210
### Stdio Transport
211

212
```python
213
import asyncio
214
from mcp import ClientSession, stdio_client, StdioServerParameters
215

216
async def stdio_example():
217
    # Client connecting to stdio server
218
    server_params = StdioServerParameters(
219
        command=["python", "my_server.py"],
220
        env={"DEBUG": "1"},
221
        cwd="/path/to/server"
222
    )
223
    
224
    async with stdio_client(server_params) as (read, write):
225
        async with ClientSession(read, write) as session:
226
            await session.initialize()
227
            tools = await session.list_tools()
228
            print(f"Tools: {[t.name for t in tools.tools]}")
229

230
# Server using stdio transport
231
async def stdio_server_example():
232
    from mcp.server import FastMCP
233
    
234
    app = FastMCP("stdio-server")
235
    
236
    @app.tool()
237
    async def hello() -> str:
238
        return "Hello from stdio server!"
239
    
240
    # Run with stdio transport
241
    app.run_stdio()
242

243
asyncio.run(stdio_example())
244
```
245

246
### WebSocket Transport
247

248
```python
249
import asyncio
250
from mcp import ClientSession, websocket_client
251

252
async def websocket_example():
253
    # Client connecting to WebSocket server
254
    async with websocket_client("ws://localhost:8000/mcp") as (read, write):
255
        async with ClientSession(read, write) as session:
256
            await session.initialize()
257
            resources = await session.list_resources()
258
            print(f"Resources: {[r.name for r in resources.resources]}")
259

260
# Server using WebSocket transport
261
async def websocket_server_example():
262
    from mcp.server import FastMCP
263
    
264
    app = FastMCP("websocket-server")
265
    
266
    @app.resource("ws://data")
267
    async def get_data() -> str:
268
        return "WebSocket server data"
269
    
270
    # Run with WebSocket transport
271
    await app.run_websocket(host="0.0.0.0", port=8000)
272

273
asyncio.run(websocket_example())
274
```
275

276
### SSE Transport
277

278
```python
279
import asyncio
280
from mcp import ClientSession, sse_client
281

282
async def sse_example():
283
    # Client connecting to SSE server
284
    headers = {"Authorization": "Bearer token123"}
285
    
286
    async with sse_client("http://localhost:8080/mcp", headers=headers) as (read, write):
287
        async with ClientSession(read, write) as session:
288
            await session.initialize()
289
            prompts = await session.list_prompts()
290
            print(f"Prompts: {[p.name for p in prompts.prompts]}")
291

292
# Server using SSE transport
293
async def sse_server_example():
294
    from mcp.server import FastMCP
295
    
296
    app = FastMCP("sse-server")
297
    
298
    @app.prompt()
299
    async def generate_report(data_type: str) -> str:
300
        return f"Report for {data_type}: [generated content]"
301
    
302
    # Run with SSE transport
303
    await app.run_sse(host="localhost", port=8080)
304

305
asyncio.run(sse_example())
306
```
307

308
### Transport with Authentication
309

310
```python
311
import asyncio
312
from mcp import ClientSession, sse_client
313
from mcp.client.auth import ClientAuth
314

315
async def authenticated_transport():
316
    # Client with authentication
317
    auth = ClientAuth(token="bearer_token_here")
318
    headers = await auth.get_headers()
319
    
320
    async with sse_client("https://api.example.com/mcp", headers=headers) as (read, write):
321
        async with ClientSession(read, write) as session:
322
            await session.initialize()
323
            # Use authenticated session
324
            result = await session.call_tool("secure_tool", {"param": "value"})
325
            print(f"Secure result: {result}")
326

327
asyncio.run(authenticated_transport())
328
```
329

330
### Multi-Transport Server
331

332
```python
333
import asyncio
334
from mcp.server import FastMCP
335

336
app = FastMCP("multi-transport-server")
337

338
@app.tool()
339
async def universal_tool(message: str) -> str:
340
    """Tool available on all transports."""
341
    return f"Processed: {message}"
342

343
@app.resource("config://server")  
344
async def server_config() -> str:
345
    """Server configuration accessible via all transports."""
346
    return "transport=multi\nstatus=active"
347

348
async def run_multi_transport():
349
    """Run server on multiple transports simultaneously."""
350
    import asyncio
351
    
352
    # Create tasks for different transports
353
    tasks = [
354
        asyncio.create_task(app.run_sse(port=8080)),
355
        asyncio.create_task(app.run_websocket(port=8081)),
356
    ]
357
    
358
    # Also support stdio in separate process
359
    print("Server available on:")
360
    print("- SSE: http://localhost:8080/mcp") 
361
    print("- WebSocket: ws://localhost:8081/mcp")
362
    print("- Stdio: python server.py")
363
    
364
    await asyncio.gather(*tasks)
365

366
if __name__ == "__main__":
367
    # Choose transport via command line or environment
368
    import sys
369
    
370
    if len(sys.argv) > 1:
371
        transport = sys.argv[1]
372
        if transport == "stdio":
373
            app.run_stdio()
374
        elif transport == "sse":
375
            asyncio.run(app.run_sse(port=8080))
376
        elif transport == "websocket":
377
            asyncio.run(app.run_websocket(port=8081))
378
        elif transport == "multi":
379
            asyncio.run(run_multi_transport())
380
    else:
381
        # Default to stdio
382
        app.run_stdio()
383
```
384

385
### Custom Transport Configuration
386

387
```python
388
import asyncio
389
from mcp import ClientSession, streamablehttp_client
390
from mcp.client.streamable_http import StreamableHttpParameters
391

392
async def custom_transport():
393
    # Custom HTTP streaming transport
394
    params = StreamableHttpParameters(
395
        url="https://api.example.com/mcp/stream",
396
        headers={
397
            "Content-Type": "application/json",
398
            "User-Agent": "MCP-Client/1.0",
399
            "X-API-Key": "secret_key"
400
        },
401
        method="POST"
402
    )
403
    
404
    async with streamablehttp_client(params.url, headers=params.headers) as (read, write):
405
        async with ClientSession(read, write) as session:
406
            await session.initialize()
407
            
408
            # Use streaming connection
409
            for i in range(10):
410
                result = await session.call_tool("stream_processor", {"batch": i})
411
                print(f"Batch {i} result: {result}")
412

413
asyncio.run(custom_transport())
414
```