FastMCP - Build MCP Servers in Python

FastMCP is a Python framework for building Model Context Protocol (MCP) servers that expose tools, resources, and prompts to Large Language Models like Claude. This skill provides production-tested patterns, error prevention, and deployment strategies for building robust MCP servers.

Quick Start

Installation

pip install fastmcp
# or
uv pip install fastmcp

Minimal Server

from fastmcp import FastMCP

# MUST be at module level for FastMCP Cloud
mcp = FastMCP("My Server")

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

if __name__ == "__main__":
    mcp.run()

Run it:

# Local development
python server.py

# With FastMCP CLI
fastmcp dev server.py

# HTTP mode
python server.py --transport http --port 8000

Core Concepts

1. Tools

Tools are functions that LLMs can call to perform actions:

@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
    """Perform mathematical operations.

    Args:
        operation: add, subtract, multiply, or divide
        a: First number
        b: Second number

    Returns:
        Result of the operation
    """
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else None
    }
    return operations.get(operation, lambda x, y: None)(a, b)

Best Practices:

• Clear, descriptive function names
• Comprehensive docstrings (LLMs read these!)
• Strong type hints (Pydantic validates automatically)
• Return structured data (dicts/lists)
• Handle errors gracefully

Sync vs Async:

# Sync tool (for non-blocking operations)
@mcp.tool()
def sync_tool(param: str) -> dict:
    return {"result": param.upper()}

# Async tool (for I/O operations, API calls)
@mcp.tool()
async def async_tool(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

2. Resources

Resources expose static or dynamic data to LLMs:

# Static resource
@mcp.resource("data://config")
def get_config() -> dict:
    """Provide application configuration."""
    return {
        "version": "1.0.0",
        "features": ["auth", "api", "cache"]
    }

# Dynamic resource
@mcp.resource("info://status")
async def server_status() -> dict:
    """Get current server status."""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "api_configured": bool(os.getenv("API_KEY"))
    }

Resource URI Schemes:

• data:// - Generic data
• file:// - File resources
• resource:// - General resources
• info:// - Information/metadata
• api:// - API endpoints
• Custom schemes allowed

3. Resource Templates

Dynamic resources with parameters in the URI:

# Single parameter
@mcp.resource("user://{user_id}/profile")
async def get_user_profile(user_id: str) -> dict:
    """Get user profile by ID."""
    user = await fetch_user_from_db(user_id)
    return {
        "id": user_id,
        "name": user.name,
        "email": user.email
    }

# Multiple parameters
@mcp.resource("org://{org_id}/team/{team_id}/members")
async def get_team_members(org_id: str, team_id: str) -> list:
    """Get team members with org context."""
    return await db.query(
        "SELECT * FROM members WHERE org_id = ? AND team_id = ?",
        [org_id, team_id]
    )

Critical: Parameter names must match exactly between URI template and function signature.

4. Prompts

Pre-configured prompts for LLMs:

@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
    """Generate analysis prompt."""
    return f"""
    Analyze {topic} considering:
    1. Current state
    2. Challenges
    3. Opportunities
    4. Recommendations

    Use available tools to gather data.
    """

@mcp.prompt("help")
def help_prompt() -> str:
    """Generate help text for server."""
    return """
    Welcome to My Server!

    Available tools:
    - search: Search for items
    - process: Process data

    Available resources:
    - info://status: Server status
    """

Context Features

FastMCP provides advanced features through context injection:

1. Elicitation (User Input)

Request user input during tool execution:

from fastmcp import Context

@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
    """Perform action with user confirmation."""
    # Request confirmation from user
    confirmed = await context.request_elicitation(
        prompt=f"Confirm {action}? (yes/no)",
        response_type=str
    )

    if confirmed.lower() == "yes":
        result = await perform_action(action)
        return {"status": "completed", "action": action}
    else:
        return {"status": "cancelled", "action": action}

2. Progress Tracking

Report progress for long-running operations:

@mcp.tool()
async def batch_import(file_path: str, context: Context) -> dict:
    """Import data with progress updates."""
    data = await read_file(file_path)
    total = len(data)

    imported = []
    for i, item in enumerate(data):
        # Report progress
        await context.report_progress(
            progress=i + 1,
            total=total,
            message=f"Importing item {i + 1}/{total}"
        )

        result = await import_item(item)
        imported.append(result)

    return {"imported": len(imported), "total": total}

3. Sampling (LLM Integration)

Request LLM completions from within tools:

@mcp.tool()
async def enhance_text(text: str, context: Context) -> str:
    """Enhance text using LLM."""
    response = await context.request_sampling(
        messages=[{
            "role": "system",
            "content": "You are a professional copywriter."
        }, {
            "role": "user",
            "content": f"Enhance this text: {text}"
        }],
        temperature=0.7,
        max_tokens=500
    )

    return response["content"]

Storage Backends

FastMCP supports pluggable storage backends built on the py-key-value-aio library. Storage backends enable persistent state for OAuth tokens, response caching, and client-side token storage.

Available Backends

Memory Store (Default):

• Ephemeral storage (lost on restart)
• Fast, no configuration needed
• Good for development

Disk Store:

• Persistent storage on local filesystem
• Encrypted by default with FernetEncryptionWrapper
• Platform-aware defaults (Mac/Windows use disk, Linux uses memory)

Redis Store:

• Distributed storage for production
• Supports multi-instance deployments
• Ideal for response caching across servers

Other Supported:

• DynamoDB (AWS)
• MongoDB
• Elasticsearch
• Memcached
• RocksDB
• Valkey

Basic Usage

from fastmcp import FastMCP
from key_value.stores import MemoryStore, DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
import os

# Memory storage (default)
mcp = FastMCP("My Server")

# Disk storage (persistent)
from key_value.stores import DiskStore

mcp = FastMCP(
    "My Server",
    storage=DiskStore(path="/app/data/storage")
)

# Redis storage (production)
from key_value.stores import RedisStore

mcp = FastMCP(
    "My Server",
    storage=RedisStore(
        host=os.getenv("REDIS_HOST", "localhost"),
        port=int(os.getenv("REDIS_PORT", "6379")),
        password=os.getenv("REDIS_PASSWORD")
    )
)

Encrypted Storage

Storage backends support automatic encryption:

from cryptography.fernet import Fernet
from key_value.encryption import FernetEncryptionWrapper
from key_value.stores import DiskStore

# Generate encryption key (store in environment!)
# key = Fernet.generate_key()

# Use encrypted storage
encrypted_storage = FernetEncryptionWrapper(
    key_value=DiskStore(path="/app/data/storage"),
    fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
)

mcp = FastMCP("My Server", storage=encrypted_storage)

OAuth Token Storage

Storage backends automatically persist OAuth tokens:

from fastmcp.auth import OAuthProxy
from key_value.stores import RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

# Production OAuth with encrypted Redis storage
auth = OAuthProxy(
    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
    client_storage=FernetEncryptionWrapper(
        key_value=RedisStore(
            host=os.getenv("REDIS_HOST"),
            password=os.getenv("REDIS_PASSWORD")
        ),
        fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
    ),
    upstream_authorization_endpoint="https://provider.com/oauth/authorize",
    upstream_token_endpoint="https://provider.com/oauth/token",
    upstream_client_id=os.getenv("OAUTH_CLIENT_ID"),
    upstream_client_secret=os.getenv("OAUTH_CLIENT_SECRET")
)

mcp = FastMCP("OAuth Server", auth=auth)

Platform-Aware Defaults

FastMCP automatically chooses storage based on platform:

• Mac/Windows: Disk storage (persistent)
• Linux: Memory storage (ephemeral)
• Override: Set storage parameter explicitly

# Explicitly use disk storage on Linux
from key_value.stores import DiskStore

mcp = FastMCP(
    "My Server",
    storage=DiskStore(path="/var/lib/mcp/storage")
)

Server Lifespans

Server lifespans provide initialization and cleanup hooks that run once per server instance (NOT per client session). This is critical for managing database connections, API clients, and other resources.

⚠️ Breaking Change in v2.13.0: Lifespan behavior changed from per-session to per-server-instance.

Basic Pattern

from fastmcp import FastMCP
from contextlib import asynccontextmanager
from typing import AsyncIterator
from dataclasses import dataclass

@dataclass
class AppContext:
    """Shared application state."""
    db: Database
    api_client: httpx.AsyncClient

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """
    Initialize resources on startup, cleanup on shutdown.
    Runs ONCE per server instance, NOT per client session.
    """
    # Startup: Initialize resources
    db = await Database.connect(os.getenv("DATABASE_URL"))
    api_client = httpx.AsyncClient(
        base_url=os.getenv("API_BASE_URL"),
        headers={"Authorization": f"Bearer {os.getenv('API_KEY')}"},
        timeout=30.0
    )

    print("Server initialized")

    try:
        # Yield context to tools
        yield AppContext(db=db, api_client=api_client)
    finally:
        # Shutdown: Cleanup resources
        await db.disconnect()
        await api_client.aclose()
        print("Server shutdown complete")

# Create server with lifespan
mcp = FastMCP("My Server", lifespan=app_lifespan)

# Access context in tools
from fastmcp import Context

@mcp.tool()
async def query_database(sql: str, context: Context) -> list:
    """Query database using shared connection."""
    # Access lifespan context
    app_context: AppContext = context.fastmcp_context.lifespan_context
    return await app_context.db.query(sql)

@mcp.tool()
async def api_request(endpoint: str, context: Context) -> dict:
    """Make API request using shared client."""
    app_context: AppContext = context.fastmcp_context.lifespan_context
    response = await app_context.api_client.get(endpoint)
    return response.json()

ASGI Integration

When using FastMCP with ASGI apps (FastAPI, Starlette), you must pass the lifespan explicitly:

from fastapi import FastAPI
from fastmcp import FastMCP

# FastMCP lifespan
@asynccontextmanager
async def mcp_lifespan(server: FastMCP):
    print("MCP server starting")
    yield
    print("MCP server stopping")

mcp = FastMCP("My Server", lifespan=mcp_lifespan)

# FastAPI app MUST include MCP lifespan
app = FastAPI(lifespan=mcp.lifespan)

# Add routes
@app.get("/")
def root():
    return {"message": "Hello World"}

❌ WRONG: Not passing lifespan to parent app

app = FastAPI()  # MCP lifespan won't run!

✅ CORRECT: Pass MCP lifespan to parent app

app = FastAPI(lifespan=mcp.lifespan)

State Management

Store and retrieve state during server lifetime:

from fastmcp import Context

@mcp.tool()
async def set_config(key: str, value: str, context: Context) -> dict:
    """Store configuration value."""
    context.fastmcp_context.set_state(key, value)
    return {"status": "saved", "key": key}

@mcp.tool()
async def get_config(key: str, context: Context) -> dict:
    """Retrieve configuration value."""
    value = context.fastmcp_context.get_state(key, default=None)
    if value is None:
        return {"error": f"Key '{key}' not found"}
    return {"key": key, "value": value}

Middleware System

FastMCP provides an MCP-native middleware system for cross-cutting functionality like logging, rate limiting, caching, and error handling.

Built-in Middleware (8 Types)

1. TimingMiddleware - Performance monitoring
2. ResponseCachingMiddleware - TTL-based caching with pluggable storage
3. LoggingMiddleware - Human-readable and JSON-structured logging
4. RateLimitingMiddleware - Token bucket and sliding window algorithms
5. ErrorHandlingMiddleware - Consistent error management
6. ToolInjectionMiddleware - Dynamic tool injection
7. PromptToolMiddleware - Tool-based prompt access for limited clients
8. ResourceToolMiddleware - Tool-based resource access for limited clients

Basic Usage

from fastmcp import FastMCP
from fastmcp.middleware import (
    TimingMiddleware,
    LoggingMiddleware,
    RateLimitingMiddleware,
    ResponseCachingMiddleware,
    ErrorHandlingMiddleware
)

mcp = FastMCP("My Server")

# Add middleware (order matters!)
mcp.add_middleware(ErrorHandlingMiddleware())
mcp.add_middleware(TimingMiddleware())
mcp.add_middleware(LoggingMiddleware(level="INFO"))
mcp.add_middleware(RateLimitingMiddleware(
    max_requests=100,
    window_seconds=60,
    algorithm="token_bucket"
))
mcp.add_middleware(ResponseCachingMiddleware(
    ttl_seconds=300,
    storage=RedisStore(host="localhost")
))

Middleware Execution Order

Middleware executes in order added:

Request Flow:
  → ErrorHandlingMiddleware (catches errors)
    → TimingMiddleware (starts timer)
      → LoggingMiddleware (logs request)
        → RateLimitingMiddleware (checks rate limit)
          → ResponseCachingMiddleware (checks cache)
            → Tool/Resource Handler
          ← ResponseCachingMiddleware (stores in cache)
        ← RateLimitingMiddleware
      ← LoggingMiddleware (logs response)
    ← TimingMiddleware (stops timer, logs duration)
  ← ErrorHandlingMiddleware (returns error if any)

Custom Middleware

Create custom middleware using hooks:

from fastmcp.middleware import BaseMiddleware
from fastmcp import Context

class AccessControlMiddleware(BaseMiddleware):
    """Check authorization before tool execution."""

    def __init__(self, allowed_users: list[str]):
        self.allowed_users = allowed_users

    async def on_call_tool(self, tool_name: str, arguments: dict, context: Context):
        """Hook runs before tool execution."""
        # Get user from context (from auth)
        user = context.fastmcp_context.get_state("user_id")

        if user not in self.allowed_users:
            raise PermissionError(f"User '{user}' not authorized")

        # Continue to tool
        return await self.next(tool_name, arguments, context)

# Add to server
mcp.add_middleware(AccessControlMiddleware(
    allowed_users=["alice", "bob", "charlie"]
))

Hook Hierarchy

Middleware hooks from most general to most specific:

1. on_message - All messages (requests and notifications)
2. on_request / on_notification - By message type
3. on_call_tool, on_read_resource, on_get_prompt - Operation-specific
4. on_list_tools, on_list_resources, on_list_prompts, on_list_resource_templates - List operations

class ComprehensiveMiddleware(BaseMiddleware):
    async def on_message(self, message: dict, context: Context):
        """Runs for ALL messages."""
        print(f"Message: {message['method']}")
        return await self.next(message, context)

    async def on_call_tool(self, tool_name: str, arguments: dict, context: Context):
        """Runs only for tool calls."""
        print(f"Tool: {tool_name}")
        return await self.next(tool_name, arguments, context)

    async def on_read_resource(self, uri: str, context: Context):
        """Runs only for resource reads."""
        print(f"Resource: {uri}")
        return await self.next(uri, context)

Response Caching Middleware

Improve performance by caching expensive operations:

from fastmcp.middleware import ResponseCachingMiddleware
from key_value.stores import RedisStore

# Cache responses for 5 minutes
cache_middleware = ResponseCachingMiddleware(
    ttl_seconds=300,
    storage=RedisStore(host="localhost"),  # Shared across instances
    cache_tools=True,       # Cache tool calls
    cache_resources=True,   # Cache resource reads
    cache_prompts=False     # Don't cache prompts
)

mcp.add_middleware(cache_middleware)

# Tools/resources are automatically cached
@mcp.tool()
async def expensive_computation(data: str) -> dict:
    """This will be cached for 5 minutes."""
    import time
    time.sleep(5)  # Expensive operation
    return {"result": process(data)}

Server Composition

Organize tools, resources, and prompts into modular components using server composition.

Two Strategies

1. import_server() - Static Snapshot:

• One-time copy of components at import time
• Changes to subserver don't propagate
• Fast (no runtime delegation)
• Use for: Bundling finalized components

2. mount() - Dynamic Link:

• Live runtime link to subserver
• Changes to subserver immediately visible
• Runtime delegation (slower)
• Use for: Modular runtime composition

Import Server (Static)

from fastmcp import FastMCP

# Subserver with tools
api_server = FastMCP("API Server")

@api_server.tool()
def api_tool():
    return "API result"

@api_server.resource("api://status")
def api_status():
    return {"status": "ok"}

# Main server imports components
main_server = FastMCP("Main Server")

# Import all components from subserver
main_server.import_server(api_server)

# Now main_server has api_tool and api://status
# Changes to api_server won't affect main_server

Mount Server (Dynamic)

from fastmcp import FastMCP

# Create servers
api_server = FastMCP("API Server")
db_server = FastMCP("DB Server")

@api_server.tool()
def fetch_data():
    return "API data"

@db_server.tool()
def query_db():
    return "DB result"

# Main server mounts subservers
main_server = FastMCP("Main Server")

# Mount with prefix
main_server.mount(api_server, prefix="api")
main_server.mount(db_server, prefix="db")

# Tools are namespaced:
# - api.fetch_data
# - db.query_db

# Resources are prefixed:
# - resource://api/path/to/resource
# - resource://db/path/to/resource

Mounting Modes

Direct Mounting (Default):

# In-memory access, subserver runs in same process
main_server.mount(subserver, prefix="sub")

Proxy Mounting:

# Treats subserver as separate entity with own lifecycle
main_server.mount(
    subserver,
    prefix="sub",
    mode="proxy"
)

Tag Filtering

Filter components when importing/mounting:

# Tag subserver components
@api_server.tool(tags=["public"])
def public_api():
    return "Public"

@api_server.tool(tags=["admin"])
def admin_api():
    return "Admin only"

# Import only public tools
main_server.import_server(
    api_server,
    include_tags=["public"]
)

# Or exclude admin tools
main_server.import_server(
    api_server,
    exclude_tags=["admin"]
)

# Tag filtering is recursive with mount()
main_server.mount(
    api_server,
    prefix="api",
    include_tags=["public"]
)

Resource Prefix Formats

Path Format (Default since v2.4.0):

resource://prefix/path/to/resource

Protocol Format (Legacy):

prefix+resource://path/to/resource

Configure format:

main_server.mount(
    subserver,
    prefix="api",
    resource_prefix_format="path"  # or "protocol"
)

OAuth Proxy & Authentication

FastMCP provides comprehensive authentication support for HTTP-based transports, including an OAuth Proxy for providers that don't support Dynamic Client Registration (DCR).

Four Authentication Patterns

1. Token Validation (TokenVerifier/JWTVerifier) - Validate external tokens
2. External Identity Providers (RemoteAuthProvider) - OAuth 2.0/OIDC with DCR
3. OAuth Proxy (OAuthProxy) - Bridge to traditional OAuth providers
4. Full OAuth (OAuthProvider) - Complete authorization server

Pattern 1: Token Validation

Validate tokens issued by external systems:

from fastmcp import FastMCP
from fastmcp.auth import JWTVerifier

# JWT verification
auth = JWTVerifier(
    issuer="https://auth.example.com",
    audience="my-mcp-server",
    public_key=os.getenv("JWT_PUBLIC_KEY")
)

mcp = FastMCP("Secure Server", auth=auth)

@mcp.tool()
async def secure_operation(context: Context) -> dict:
    """Only accessible with valid JWT."""
    # Token validated automatically
    user = context.fastmcp_context.get_state("user_id")
    return {"user": user, "status": "authorized"}

Pattern 2: External Identity Providers

Use OAuth 2.0/OIDC providers with Dynamic Client Registration:

from fastmcp.auth import RemoteAuthProvider

auth = RemoteAuthProvider(
    issuer="https://auth.example.com",
    # Provider must support DCR
)

mcp = FastMCP("OAuth Server", auth=auth)

Pattern 3: OAuth Proxy (Recommended for Production)

Bridge to OAuth providers without DCR support (GitHub, Google, Azure, AWS, Discord, Facebook, etc.):

from fastmcp.auth import OAuthProxy
from key_value.stores import RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
import os

auth = OAuthProxy(
    # JWT signing for issued tokens
    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],

    # Encrypted storage for upstream tokens
    client_storage=FernetEncryptionWrapper(
        key_value=RedisStore(
            host=os.getenv("REDIS_HOST"),
            password=os.getenv("REDIS_PASSWORD")
        ),
        fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
    ),

    # Upstream OAuth provider
    upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
    upstream_token_endpoint="https://github.com/login/oauth/access_token",
    upstream_client_id=os.getenv("GITHUB_CLIENT_ID"),
    upstream_client_secret=os.getenv("GITHUB_CLIENT_SECRET"),

    # Scopes
    upstream_scope="read:user user:email",

    # Security: Enable consent screen (prevents confused deputy attacks)
    enable_consent_screen=True
)

mcp = FastMCP("GitHub Auth Server", auth=auth)

OAuth Proxy Features

Token Factory Pattern:

• Proxy issues its own JWTs (not forwarding upstream tokens)
• Upstream tokens stored encrypted
• Proxy tokens can have custom claims

Consent Screens:

• Prevents authorization bypass attacks
• Shows user what permissions are being granted
• Required for security compliance

PKCE Support:

• End-to-end validation from client to upstream
• Protects against authorization code interception

RFC 7662 Token Introspection:

• Validate tokens with upstream provider
• Check revocation status

Pattern 4: Full OAuth Provider

Run complete authorization server: