CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-fastmcp

The fast, Pythonic way to build MCP servers and clients with minimal boilerplate code.

—

Pending
Overview
Eval results
Files

transports.mddocs/

Transport Layer

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

Capabilities

Base Transport Classes

Abstract base classes for implementing transport protocols.

class ClientTransport:
    """Base class for client transport implementations."""
    
    async def connect(self) -> None:
        """Establish connection to the server."""
    
    async def disconnect(self) -> None:
        """Close connection to the server."""
    
    async def send_request(self, request: dict) -> dict:
        """Send request and return response."""

Standard I/O Transports

Transport implementations for standard input/output communication with subprocesses.

class StdioTransport(ClientTransport):
    """Standard I/O transport for subprocess communication."""
    
    def __init__(
        self,
        command: str,
        args: list[str] | None = None,
        env: dict[str, str] | None = None,
        cwd: str | None = None
    ):
        """
        Initialize stdio transport.
        
        Parameters:
        - command: Command to execute
        - args: Command arguments
        - env: Environment variables
        - cwd: Working directory
        """

class PythonStdioTransport(StdioTransport):
    """Stdio transport specifically for Python scripts."""
    
    def __init__(
        self,
        script_path: str,
        python_executable: str = "python",
        args: list[str] | None = None
    ):
        """
        Initialize Python stdio transport.
        
        Parameters:
        - script_path: Path to Python script
        - python_executable: Python interpreter to use
        - args: Additional script arguments
        """

class NodeStdioTransport(StdioTransport):
    """Stdio transport for Node.js scripts."""
    
    def __init__(
        self,
        script_path: str,
        node_executable: str = "node",
        args: list[str] | None = None
    ):
        """
        Initialize Node.js stdio transport.
        
        Parameters:
        - script_path: Path to Node.js script
        - node_executable: Node.js interpreter to use
        - args: Additional script arguments
        """

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

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

class NpxStdioTransport(StdioTransport):
    """Stdio transport using npx for Node.js packages."""

Network Transports

Transport implementations for network-based communication.

class SSETransport(ClientTransport):
    """Server-Sent Events transport for HTTP streaming."""
    
    def __init__(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        timeout: float = 30.0
    ):
        """
        Initialize SSE transport.
        
        Parameters:
        - url: SSE endpoint URL
        - headers: Optional HTTP headers
        - timeout: Connection timeout
        """

class StreamableHttpTransport(ClientTransport):
    """Streamable HTTP transport for request/response."""
    
    def __init__(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        timeout: float = 30.0
    ):
        """
        Initialize streamable HTTP transport.
        
        Parameters:
        - url: HTTP endpoint URL
        - headers: Optional HTTP headers  
        - timeout: Request timeout
        """

class WSTransport(ClientTransport):
    """WebSocket transport for real-time communication."""
    
    def __init__(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        ping_interval: float = 30.0
    ):
        """
        Initialize WebSocket transport.
        
        Parameters:
        - url: WebSocket URL
        - headers: Optional connection headers
        - ping_interval: Ping interval for keepalive
        """

FastMCP-Specific Transport

In-memory transport for direct FastMCP server connection.

class FastMCPTransport(ClientTransport):
    """In-memory transport for direct FastMCP server connection."""
    
    def __init__(self, server: FastMCP):
        """
        Initialize FastMCP transport.
        
        Parameters:
        - server: FastMCP server instance to connect to
        """

Usage Examples

Basic Transport Usage

from fastmcp import Client
from fastmcp.client import (
    StdioTransport, 
    SSETransport,
    StreamableHttpTransport,
    WSTransport
)

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

async def sse_example():
    """Connect via Server-Sent Events."""
    transport = SSETransport(
        url="http://localhost:8000/sse",
        headers={"Authorization": "Bearer token"}
    )
    
    async with Client(transport=transport) as client:
        resources = await client.list_resources()
        return resources

async def http_example():
    """Connect via HTTP transport."""
    transport = StreamableHttpTransport(
        url="http://localhost:8000/mcp",
        timeout=60.0
    )
    
    async with Client(transport=transport) as client:
        prompts = await client.list_prompts()
        return prompts

async def websocket_example():
    """Connect via WebSocket."""
    transport = WSTransport(
        url="ws://localhost:8000/ws",
        ping_interval=20.0
    )
    
    async with Client(transport=transport) as client:
        result = await client.call_tool("process_data", {"data": "example"})
        return result.text

Auto-Detection Examples

from fastmcp import Client

async def auto_detection_examples():
    """Client automatically detects appropriate transport."""
    
    # Auto-detects stdio transport
    async with Client("python server.py") as client:
        result1 = await client.call_tool("test", {})
    
    # Auto-detects SSE transport
    async with Client("http://localhost:8000/sse") as client:
        result2 = await client.call_tool("test", {})
    
    # Auto-detects HTTP transport  
    async with Client("http://localhost:8000/mcp") as client:
        result3 = await client.call_tool("test", {})
    
    # Auto-detects WebSocket transport
    async with Client("ws://localhost:8000/ws") as client:
        result4 = await client.call_tool("test", {})
    
    return [result1, result2, result3, result4]

Language-Specific Transports

from fastmcp import Client
from fastmcp.client import (
    PythonStdioTransport,
    NodeStdioTransport,
    UvStdioTransport,
    NpxStdioTransport
)

async def python_transport():
    """Connect to Python MCP server."""
    transport = PythonStdioTransport(
        script_path="./servers/python_server.py",
        python_executable="python3.11"
    )
    
    async with Client(transport=transport) as client:
        return await client.list_tools()

async def node_transport():
    """Connect to Node.js MCP server."""
    transport = NodeStdioTransport(
        script_path="./servers/node_server.js",
        node_executable="node"
    )
    
    async with Client(transport=transport) as client:
        return await client.list_tools()

async def uv_transport():
    """Connect using uv package manager."""
    transport = UvStdioTransport(
        command="uv",
        args=["run", "python", "server.py"]
    )
    
    async with Client(transport=transport) as client:
        return await client.list_tools()

async def npx_transport():
    """Connect using npx package runner."""
    transport = NpxStdioTransport(
        command="npx",
        args=["@modelcontextprotocol/server-example"]
    )
    
    async with Client(transport=transport) as client:
        return await client.list_tools()

Advanced Transport Configuration

from fastmcp import Client
from fastmcp.client import SSETransport, StreamableHttpTransport
from fastmcp.client.auth import BearerAuth

async def authenticated_transport():
    """Transport with authentication."""
    auth = BearerAuth("your-access-token")
    
    transport = SSETransport(
        url="https://secure-api.example.com/sse",
        headers={
            "User-Agent": "MyApp/1.0",
            "X-Client-Version": "2.0"
        },
        timeout=120.0
    )
    
    async with Client(transport=transport, auth=auth) as client:
        return await client.call_tool("secure_operation", {})

async def custom_headers_transport():
    """Transport with custom headers."""
    transport = StreamableHttpTransport(
        url="http://api.example.com/mcp",
        headers={
            "Authorization": "Bearer custom-token",
            "X-API-Version": "v2",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
    )
    
    async with Client(transport=transport) as client:
        return await client.list_resources()

async def resilient_transport():
    """Transport with error handling and retries."""
    import asyncio
    from fastmcp.exceptions import ClientError
    
    transport = SSETransport(
        url="http://sometimes-unreliable.example.com/sse",
        timeout=10.0
    )
    
    max_retries = 3
    for attempt in range(max_retries):
        try:
            async with Client(transport=transport) as client:
                return await client.call_tool("operation", {})
        except ClientError as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

In-Memory Testing Transport

from fastmcp import FastMCP, Client
from fastmcp.client import FastMCPTransport

async def in_memory_testing():
    """Test server using in-memory transport."""
    # Create server
    server = FastMCP("Test Server")
    
    @server.tool
    def add(a: int, b: int) -> int:
        """Add two numbers."""
        return a + b
    
    @server.resource("config://test")
    def get_config():
        """Get test configuration."""
        return {"test": True, "env": "development"}
    
    # Connect via in-memory transport (no subprocess)
    transport = FastMCPTransport(server)
    
    async with Client(transport=transport) as client:
        # Test tools
        tools = await client.list_tools()
        assert len(tools) == 1
        assert tools[0]["name"] == "add"
        
        result = await client.call_tool("add", {"a": 5, "b": 3})
        assert result.text == "8"
        
        # Test resources
        resources = await client.list_resources()
        assert len(resources) == 1
        
        config = await client.read_resource("config://test")
        assert "test" in config.content
        
        return "All tests passed"

# Usage in test suite
async def test_server_functionality():
    """Integration test using in-memory transport."""
    result = await in_memory_testing()
    print(result)

Server Transport Configuration

from fastmcp import FastMCP

mcp = FastMCP("Multi-Transport Server")

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

# Run with different transports
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1:
        transport = sys.argv[1]
    else:
        transport = "stdio"
    
    if transport == "stdio":
        # Default stdio transport
        mcp.run()
    
    elif transport == "http":
        # HTTP transport
        mcp.run(
            transport="http",
            host="0.0.0.0",
            port=8080,
            path="/mcp"
        )
    
    elif transport == "sse":
        # Server-Sent Events transport
        mcp.run(
            transport="sse",
            host="0.0.0.0", 
            port=8080
        )
    
    else:
        print(f"Unknown transport: {transport}")
        sys.exit(1)

Transport Error Handling

from fastmcp import Client
from fastmcp.client import SSETransport
from fastmcp.exceptions import ClientError, ConnectionError
import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def robust_client_connection():
    """Robust client with comprehensive error handling."""
    transport = SSETransport(
        url="http://api.example.com/sse",
        timeout=30.0
    )
    
    try:
        async with Client(transport=transport) as client:
            # Test connection
            await client.list_tools()
            logger.info("Successfully connected to server")
            
            # Perform operations
            result = await client.call_tool("test_operation", {})
            return result.text
            
    except ConnectionError as e:
        logger.error(f"Failed to connect to server: {e}")
        return "Connection failed"
    
    except ClientError as e:
        logger.error(f"Client error: {e}")
        return "Client error occurred"
    
    except asyncio.TimeoutError:
        logger.error("Request timed out")
        return "Request timeout"
    
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return "Unexpected error occurred"

async def connection_with_fallback():
    """Try multiple transports with fallback."""
    transports = [
        SSETransport("http://primary.example.com/sse"),
        SSETransport("http://backup.example.com/sse"),
        StreamableHttpTransport("http://fallback.example.com/mcp")
    ]
    
    for i, transport in enumerate(transports):
        try:
            logger.info(f"Trying transport {i+1}/{len(transports)}")
            
            async with Client(transport=transport) as client:
                result = await client.call_tool("health_check", {})
                logger.info(f"Successfully connected via transport {i+1}")
                return result.text
                
        except Exception as e:
            logger.warning(f"Transport {i+1} failed: {e}")
            if i == len(transports) - 1:
                logger.error("All transports failed")
                raise
    
    return "No successful connection"

Transport Selection Guide

When to Use Each Transport

Stdio Transport:

  • Local development and testing
  • Command-line tools and scripts
  • Simple server deployment
  • When subprocess management is acceptable

HTTP/SSE Transport:

  • Web deployments and cloud services
  • When you need standard HTTP semantics
  • Load balancing and reverse proxy scenarios
  • Production deployments with monitoring

WebSocket Transport:

  • Real-time applications
  • When you need bidirectional communication
  • Long-lived connections
  • Gaming or chat applications

FastMCP Transport:

  • Unit testing and integration testing
  • Embedded scenarios where no network is needed
  • High-performance local communication
  • Development and debugging

Install with Tessl CLI

npx tessl i tessl/pypi-fastmcp

docs

authentication.md

client.md

context.md

index.md

prompts.md

resources.md

server.md

tools.md

transports.md

utilities.md

tile.json