pip install fastmcp

undefined

from fastmcp import FastMCP

**Run it:**
```bash

undefined

@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)

undefined

undefined

undefined

**Resource URI Schemes:**
- `data://` - Generic data
- `file://` - File resources
- `resource://` - General resources
- `info://` - Information/metadata
- `api://` - API endpoints
- Custom schemes allowed

undefined

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

@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
    """

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}

@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}

@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"]

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

undefined

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

undefined

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

undefined

undefined

undefined

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")

undefined

from fastapi import FastAPI
from fastmcp import FastMCP

**❌ WRONG**: Not passing lifespan to parent app
```python
app = FastAPI()  # MCP lifespan won't run!

app = FastAPI(lifespan=mcp.lifespan)

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}

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

mcp = FastMCP("My Server")

undefined

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)

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)

undefined

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)

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

undefined

from fastmcp import FastMCP

undefined

from fastmcp import FastMCP

undefined

undefined

**Proxy Mounting**:
```python

undefined

undefined

undefined

resource://prefix/path/to/resource

prefix+resource://path/to/resource

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

from fastmcp import FastMCP
from fastmcp.auth import JWTVerifier

undefined

from fastmcp.auth import RemoteAuthProvider

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

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

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)

from fastmcp.auth import OAuthProvider

auth = OAuthProvider(
    issuer="https://my-auth-server.com",
    client_storage=RedisStore(host="localhost"),
    # Full OAuth 2.0 server implementation
)

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

export FASTMCP_SERVER_AUTH='{"type": "oauth_proxy", "upstream_authorization_endpoint": "...", ...}'

undefined

undefined

from fastmcp import FastMCP, Icon

mcp = FastMCP(
    name="Weather Service",
    website_url="https://weather.example.com",
    icons=[
        Icon(
            url="https://example.com/icon-small.png",
            size="small"
        ),
        Icon(
            url="https://example.com/icon-large.png",
            size="large"
        )
    ]
)

from fastmcp import Icon

@mcp.tool(icons=[
    Icon(url="https://example.com/tool-icon.png")
])
async def analyze_data(data: str) -> dict:
    """Analyze data with visual icon."""
    return {"result": "analyzed"}

@mcp.resource(
    "user://{user_id}/profile",
    icons=[Icon(url="https://example.com/user-icon.png")]
)
async def get_user(user_id: str) -> dict:
    """User profile with icon."""
    return {"id": user_id, "name": "Alice"}

@mcp.prompt(
    "analyze",
    icons=[Icon(url="https://example.com/prompt-icon.png")]
)
def analysis_prompt(topic: str) -> str:
    """Analysis prompt with icon."""
    return f"Analyze {topic}"

from fastmcp import Icon, Image

undefined

mcp = FastMCP(
    "Responsive Server",
    icons=[
        Icon(url="icon-16.png", size="small"),    # 16x16
        Icon(url="icon-32.png", size="medium"),   # 32x32
        Icon(url="icon-64.png", size="large"),    # 64x64
    ]
)

import httpx
import os

undefined

from fastmcp import FastMCP
from fastmcp.server.openapi import RouteMap, MCPType
import httpx

undefined

from fastapi import FastAPI
from fastmcp import FastMCP

undefined

undefined

# Your implementation
return {"status": "success", "data": data}

undefined

def create_server():
    mcp = FastMCP("my-server")
    return mcp

if __name__ == "__main__":
    server = create_server()  # Too late for cloud!
    server.run()

def create_server() -> FastMCP:
    mcp = FastMCP("my-server")
    # Complex setup logic
    return mcp

undefined

git init
git add .
git commit -m "Initial MCP server"
gh repo create my-mcp-server --public
git push -u origin main

{
  "mcpServers": {
    "my-server": {
      "url": "https://your-project.fastmcp.app/mcp",
      "transport": "http"
    }
  }
}

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"],
      "env": {
        "API_KEY": "your-key",
        "DATABASE_URL": "your-db-url"
      }
    }
  }
}

{
  "mcpServers": {
    "my-server": {
      "command": "uv",
      "args": ["run", "python", "/absolute/path/to/server.py"]
    }
  }
}

RuntimeError: No server object found at module level

undefined

**Source:** FastMCP Cloud documentation, deployment failures

---

RuntimeError: no running event loop
TypeError: object coroutine can't be used in 'await' expression

undefined

**Source:** GitHub issues #156, #203

---