0
# Transport Layer
1

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

4
## Capabilities
5

6
### Base Transport Classes
7

8
Abstract base classes for implementing transport protocols.
9

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

24
### Standard I/O Transports
25

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

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

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

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

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

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

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

95
### Network Transports
96

97
Transport implementations for network-based communication.
98

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

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

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

155
### FastMCP-Specific Transport
156

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

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

172
## Usage Examples
173

174
### Basic Transport Usage
175

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

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

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

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

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

228
### Auto-Detection Examples
229

230
```python
231
from fastmcp import Client
232

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

255
### Language-Specific Transports
256

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

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

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

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

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

307
### Advanced Transport Configuration
308

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

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

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

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

366
### In-Memory Testing Transport
367

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

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

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

415
### Server Transport Configuration
416

417
```python
418
from fastmcp import FastMCP
419

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

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

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

462
### Transport Error Handling
463

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

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

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

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

533
## Transport Selection Guide
534

535
### When to Use Each Transport
536

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

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

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

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