1
# Transport Layer
2

3
Multiple transport protocols for flexible server deployment and client connections. FastMCP supports various transport mechanisms to accommodate different deployment scenarios and client environments.
4

5
## Capabilities
6

7
### Base Transport Classes
8

9
Abstract base classes for implementing transport protocols.
10

11
```python { .api }
12
class ClientTransport:
13
    """Base class for client transport implementations."""
14
    
15
    async def connect(self) -> None:
16
        """Establish connection to the server."""
17
    
18
    async def disconnect(self) -> None:
19
        """Close connection to the server."""
20
    
21
    async def send_request(self, request: dict) -> dict:
22
        """Send request and return response."""
23
```
24

25
### Standard I/O Transports
26

27
Transport implementations for standard input/output communication with subprocesses.
28

29
```python { .api }
30
class StdioTransport(ClientTransport):
31
    """Standard I/O transport for subprocess communication."""
32
    
33
    def __init__(
34
        self,
35
        command: str,
36
        args: list[str] | None = None,
37
        env: dict[str, str] | None = None,
38
        cwd: str | None = None
39
    ):
40
        """
41
        Initialize stdio transport.
42
        
43
        Parameters:
44
        - command: Command to execute
45
        - args: Command arguments
46
        - env: Environment variables
47
        - cwd: Working directory
48
        """
49

50
class PythonStdioTransport(StdioTransport):
51
    """Stdio transport specifically for Python scripts."""
52
    
53
    def __init__(
54
        self,
55
        script_path: str,
56
        python_executable: str = "python",
57
        args: list[str] | None = None
58
    ):
59
        """
60
        Initialize Python stdio transport.
61
        
62
        Parameters:
63
        - script_path: Path to Python script
64
        - python_executable: Python interpreter to use
65
        - args: Additional script arguments
66
        """
67

68
class NodeStdioTransport(StdioTransport):
69
    """Stdio transport for Node.js scripts."""
70
    
71
    def __init__(
72
        self,
73
        script_path: str,
74
        node_executable: str = "node",
75
        args: list[str] | None = None
76
    ):
77
        """
78
        Initialize Node.js stdio transport.
79
        
80
        Parameters:
81
        - script_path: Path to Node.js script
82
        - node_executable: Node.js interpreter to use
83
        - args: Additional script arguments
84
        """
85

86
class UvStdioTransport(StdioTransport):
87
    """Stdio transport using uv Python package manager."""
88

89
class UvxStdioTransport(StdioTransport):
90
    """Stdio transport using uvx for package execution."""
91

92
class NpxStdioTransport(StdioTransport):
93
    """Stdio transport using npx for Node.js packages."""
94
```
95

96
### Network Transports
97

98
Transport implementations for network-based communication.
99

100
```python { .api }
101
class SSETransport(ClientTransport):
102
    """Server-Sent Events transport for HTTP streaming."""
103
    
104
    def __init__(
105
        self,
106
        url: str,
107
        headers: dict[str, str] | None = None,
108
        timeout: float = 30.0
109
    ):
110
        """
111
        Initialize SSE transport.
112
        
113
        Parameters:
114
        - url: SSE endpoint URL
115
        - headers: Optional HTTP headers
116
        - timeout: Connection timeout
117
        """
118

119
class StreamableHttpTransport(ClientTransport):
120
    """Streamable HTTP transport for request/response."""
121
    
122
    def __init__(
123
        self,
124
        url: str,
125
        headers: dict[str, str] | None = None,
126
        timeout: float = 30.0
127
    ):
128
        """
129
        Initialize streamable HTTP transport.
130
        
131
        Parameters:
132
        - url: HTTP endpoint URL
133
        - headers: Optional HTTP headers  
134
        - timeout: Request timeout
135
        """
136

137
class WSTransport(ClientTransport):
138
    """WebSocket transport for real-time communication."""
139
    
140
    def __init__(
141
        self,
142
        url: str,
143
        headers: dict[str, str] | None = None,
144
        ping_interval: float = 30.0
145
    ):
146
        """
147
        Initialize WebSocket transport.
148
        
149
        Parameters:
150
        - url: WebSocket URL
151
        - headers: Optional connection headers
152
        - ping_interval: Ping interval for keepalive
153
        """
154
```
155

156
### FastMCP-Specific Transport
157

158
In-memory transport for direct FastMCP server connection.
159

160
```python { .api }
161
class FastMCPTransport(ClientTransport):
162
    """In-memory transport for direct FastMCP server connection."""
163
    
164
    def __init__(self, server: FastMCP):
165
        """
166
        Initialize FastMCP transport.
167
        
168
        Parameters:
169
        - server: FastMCP server instance to connect to
170
        """
171
```
172

173
## Usage Examples
174

175
### Basic Transport Usage
176

177
```python
178
from fastmcp import Client
179
from fastmcp.client import (
180
    StdioTransport, 
181
    SSETransport,
182
    StreamableHttpTransport,
183
    WSTransport
184
)
185

186
async def stdio_example():
187
    """Connect via stdio transport."""
188
    transport = StdioTransport("python", ["server.py"])
189
    
190
    async with Client(transport=transport) as client:
191
        tools = await client.list_tools()
192
        result = await client.call_tool("hello", {"name": "World"})
193
        return result.text
194

195
async def sse_example():
196
    """Connect via Server-Sent Events."""
197
    transport = SSETransport(
198
        url="http://localhost:8000/sse",
199
        headers={"Authorization": "Bearer token"}
200
    )
201
    
202
    async with Client(transport=transport) as client:
203
        resources = await client.list_resources()
204
        return resources
205

206
async def http_example():
207
    """Connect via HTTP transport."""
208
    transport = StreamableHttpTransport(
209
        url="http://localhost:8000/mcp",
210
        timeout=60.0
211
    )
212
    
213
    async with Client(transport=transport) as client:
214
        prompts = await client.list_prompts()
215
        return prompts
216

217
async def websocket_example():
218
    """Connect via WebSocket."""
219
    transport = WSTransport(
220
        url="ws://localhost:8000/ws",
221
        ping_interval=20.0
222
    )
223
    
224
    async with Client(transport=transport) as client:
225
        result = await client.call_tool("process_data", {"data": "example"})
226
        return result.text
227
```
228

229
### Auto-Detection Examples
230

231
```python
232
from fastmcp import Client
233

234
async def auto_detection_examples():
235
    """Client automatically detects appropriate transport."""
236
    
237
    # Auto-detects stdio transport
238
    async with Client("python server.py") as client:
239
        result1 = await client.call_tool("test", {})
240
    
241
    # Auto-detects SSE transport
242
    async with Client("http://localhost:8000/sse") as client:
243
        result2 = await client.call_tool("test", {})
244
    
245
    # Auto-detects HTTP transport  
246
    async with Client("http://localhost:8000/mcp") as client:
247
        result3 = await client.call_tool("test", {})
248
    
249
    # Auto-detects WebSocket transport
250
    async with Client("ws://localhost:8000/ws") as client:
251
        result4 = await client.call_tool("test", {})
252
    
253
    return [result1, result2, result3, result4]
254
```
255

256
### Language-Specific Transports
257

258
```python
259
from fastmcp import Client
260
from fastmcp.client import (
261
    PythonStdioTransport,
262
    NodeStdioTransport,
263
    UvStdioTransport,
264
    NpxStdioTransport
265
)
266

267
async def python_transport():
268
    """Connect to Python MCP server."""
269
    transport = PythonStdioTransport(
270
        script_path="./servers/python_server.py",
271
        python_executable="python3.11"
272
    )
273
    
274
    async with Client(transport=transport) as client:
275
        return await client.list_tools()
276

277
async def node_transport():
278
    """Connect to Node.js MCP server."""
279
    transport = NodeStdioTransport(
280
        script_path="./servers/node_server.js",
281
        node_executable="node"
282
    )
283
    
284
    async with Client(transport=transport) as client:
285
        return await client.list_tools()
286

287
async def uv_transport():
288
    """Connect using uv package manager."""
289
    transport = UvStdioTransport(
290
        command="uv",
291
        args=["run", "python", "server.py"]
292
    )
293
    
294
    async with Client(transport=transport) as client:
295
        return await client.list_tools()
296

297
async def npx_transport():
298
    """Connect using npx package runner."""
299
    transport = NpxStdioTransport(
300
        command="npx",
301
        args=["@modelcontextprotocol/server-example"]
302
    )
303
    
304
    async with Client(transport=transport) as client:
305
        return await client.list_tools()
306
```
307

308
### Advanced Transport Configuration
309

310
```python
311
from fastmcp import Client
312
from fastmcp.client import SSETransport, StreamableHttpTransport
313
from fastmcp.client.auth import BearerAuth
314

315
async def authenticated_transport():
316
    """Transport with authentication."""
317
    auth = BearerAuth("your-access-token")
318
    
319
    transport = SSETransport(
320
        url="https://secure-api.example.com/sse",
321
        headers={
322
            "User-Agent": "MyApp/1.0",
323
            "X-Client-Version": "2.0"
324
        },
325
        timeout=120.0
326
    )
327
    
328
    async with Client(transport=transport, auth=auth) as client:
329
        return await client.call_tool("secure_operation", {})
330

331
async def custom_headers_transport():
332
    """Transport with custom headers."""
333
    transport = StreamableHttpTransport(
334
        url="http://api.example.com/mcp",
335
        headers={
336
            "Authorization": "Bearer custom-token",
337
            "X-API-Version": "v2",
338
            "Accept": "application/json",
339
            "Content-Type": "application/json"
340
        }
341
    )
342
    
343
    async with Client(transport=transport) as client:
344
        return await client.list_resources()
345

346
async def resilient_transport():
347
    """Transport with error handling and retries."""
348
    import asyncio
349
    from fastmcp.exceptions import ClientError
350
    
351
    transport = SSETransport(
352
        url="http://sometimes-unreliable.example.com/sse",
353
        timeout=10.0
354
    )
355
    
356
    max_retries = 3
357
    for attempt in range(max_retries):
358
        try:
359
            async with Client(transport=transport) as client:
360
                return await client.call_tool("operation", {})
361
        except ClientError as e:
362
            if attempt == max_retries - 1:
363
                raise
364
            await asyncio.sleep(2 ** attempt)  # Exponential backoff
365
```
366

367
### In-Memory Testing Transport
368

369
```python
370
from fastmcp import FastMCP, Client
371
from fastmcp.client import FastMCPTransport
372

373
async def in_memory_testing():
374
    """Test server using in-memory transport."""
375
    # Create server
376
    server = FastMCP("Test Server")
377
    
378
    @server.tool
379
    def add(a: int, b: int) -> int:
380
        """Add two numbers."""
381
        return a + b
382
    
383
    @server.resource("config://test")
384
    def get_config():
385
        """Get test configuration."""
386
        return {"test": True, "env": "development"}
387
    
388
    # Connect via in-memory transport (no subprocess)
389
    transport = FastMCPTransport(server)
390
    
391
    async with Client(transport=transport) as client:
392
        # Test tools
393
        tools = await client.list_tools()
394
        assert len(tools) == 1
395
        assert tools[0]["name"] == "add"
396
        
397
        result = await client.call_tool("add", {"a": 5, "b": 3})
398
        assert result.text == "8"
399
        
400
        # Test resources
401
        resources = await client.list_resources()
402
        assert len(resources) == 1
403
        
404
        config = await client.read_resource("config://test")
405
        assert "test" in config.content
406
        
407
        return "All tests passed"
408

409
# Usage in test suite
410
async def test_server_functionality():
411
    """Integration test using in-memory transport."""
412
    result = await in_memory_testing()
413
    print(result)
414
```
415

416
### Server Transport Configuration
417

418
```python
419
from fastmcp import FastMCP
420

421
mcp = FastMCP("Multi-Transport Server")
422

423
@mcp.tool
424
def hello(name: str) -> str:
425
    """Say hello."""
426
    return f"Hello, {name}!"
427

428
# Run with different transports
429
if __name__ == "__main__":
430
    import sys
431
    
432
    if len(sys.argv) > 1:
433
        transport = sys.argv[1]
434
    else:
435
        transport = "stdio"
436
    
437
    if transport == "stdio":
438
        # Default stdio transport
439
        mcp.run()
440
    
441
    elif transport == "http":
442
        # HTTP transport
443
        mcp.run(
444
            transport="http",
445
            host="0.0.0.0",
446
            port=8080,
447
            path="/mcp"
448
        )
449
    
450
    elif transport == "sse":
451
        # Server-Sent Events transport
452
        mcp.run(
453
            transport="sse",
454
            host="0.0.0.0", 
455
            port=8080
456
        )
457
    
458
    else:
459
        print(f"Unknown transport: {transport}")
460
        sys.exit(1)
461
```
462

463
### Transport Error Handling
464

465
```python
466
from fastmcp import Client
467
from fastmcp.client import SSETransport
468
from fastmcp.exceptions import ClientError, ConnectionError
469
import asyncio
470
import logging
471

472
logging.basicConfig(level=logging.INFO)
473
logger = logging.getLogger(__name__)
474

475
async def robust_client_connection():
476
    """Robust client with comprehensive error handling."""
477
    transport = SSETransport(
478
        url="http://api.example.com/sse",
479
        timeout=30.0
480
    )
481
    
482
    try:
483
        async with Client(transport=transport) as client:
484
            # Test connection
485
            await client.list_tools()
486
            logger.info("Successfully connected to server")
487
            
488
            # Perform operations
489
            result = await client.call_tool("test_operation", {})
490
            return result.text
491
            
492
    except ConnectionError as e:
493
        logger.error(f"Failed to connect to server: {e}")
494
        return "Connection failed"
495
    
496
    except ClientError as e:
497
        logger.error(f"Client error: {e}")
498
        return "Client error occurred"
499
    
500
    except asyncio.TimeoutError:
501
        logger.error("Request timed out")
502
        return "Request timeout"
503
    
504
    except Exception as e:
505
        logger.error(f"Unexpected error: {e}")
506
        return "Unexpected error occurred"
507

508
async def connection_with_fallback():
509
    """Try multiple transports with fallback."""
510
    transports = [
511
        SSETransport("http://primary.example.com/sse"),
512
        SSETransport("http://backup.example.com/sse"),
513
        StreamableHttpTransport("http://fallback.example.com/mcp")
514
    ]
515
    
516
    for i, transport in enumerate(transports):
517
        try:
518
            logger.info(f"Trying transport {i+1}/{len(transports)}")
519
            
520
            async with Client(transport=transport) as client:
521
                result = await client.call_tool("health_check", {})
522
                logger.info(f"Successfully connected via transport {i+1}")
523
                return result.text
524
                
525
        except Exception as e:
526
            logger.warning(f"Transport {i+1} failed: {e}")
527
            if i == len(transports) - 1:
528
                logger.error("All transports failed")
529
                raise
530
    
531
    return "No successful connection"
532
```
533

534
## Transport Selection Guide
535

536
### When to Use Each Transport
537

538
**Stdio Transport:**
539
- Local development and testing
540
- Command-line tools and scripts
541
- Simple server deployment
542
- When subprocess management is acceptable
543

544
**HTTP/SSE Transport:**
545
- Web deployments and cloud services
546
- When you need standard HTTP semantics
547
- Load balancing and reverse proxy scenarios
548
- Production deployments with monitoring
549

550
**WebSocket Transport:**
551
- Real-time applications
552
- When you need bidirectional communication
553
- Long-lived connections
554
- Gaming or chat applications
555

556
**FastMCP Transport:**
557
- Unit testing and integration testing
558
- Embedded scenarios where no network is needed
559
- High-performance local communication
560
- Development and debugging