pip install fastmcp
# or: uv pip install fastmcp

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

python server.py              # Local development
fastmcp dev server.py         # With FastMCP CLI
python server.py --transport http --port 8000  # HTTP mode

@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
    """Perform mathematical operations."""
    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)

@mcp.resource("data://config")
def get_config() -> dict:
    """Provide application configuration."""
    return {"version": "1.0.0", "features": ["auth", "api"]}
# Dynamic resource with parameters
@mcp.resource("user://{user_id}/profile")
async def get_user_profile(user_id: str) -> dict:
    """Get user profile by ID."""
    return {"id": user_id, "name": f"User {user_id}"}

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

from fastmcp import Context
@mcp.tool()
async def batch_process(items: list, context: Context) -> dict:
    """Process items with progress updates."""
    for i, item in enumerate(items):
        await context.report_progress(i + 1, len(items), f"Processing {item}")
        await process_item(item)
    return {"processed": len(items)}

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

from key_value.stores import DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Memory (default) - Development only
mcp = FastMCP("Dev Server")
# Disk - Single instance
mcp = FastMCP(
    "Production Server",
    storage=FernetEncryptionWrapper(
        key_value=DiskStore(path="/var/lib/mcp/storage"),
        fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
    )
)
# Redis - Multi-instance
mcp = FastMCP(
    "Production Server",
    storage=FernetEncryptionWrapper(
        key_value=RedisStore(
            host=os.getenv("REDIS_HOST"),
            password=os.getenv("REDIS_PASSWORD")
        ),
        fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
    )
)

from contextlib import asynccontextmanager
@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Runs ONCE when server starts (v2.13.0+)."""
    db = await Database.connect()
    print("Server starting")
    
    try:
        yield {"db": db}
    finally:
        await db.disconnect()
        print("Server stopping")
mcp = FastMCP("My Server", lifespan=app_lifespan)

from fastmcp.middleware import (
    LoggingMiddleware,
    TimingMiddleware,
    RateLimitingMiddleware,
    ResponseCachingMiddleware
)
# Order matters!
mcp.add_middleware(LoggingMiddleware())
mcp.add_middleware(TimingMiddleware())
mcp.add_middleware(RateLimitingMiddleware(max_requests=100, window_seconds=60))
mcp.add_middleware(ResponseCachingMiddleware(ttl_seconds=3600))

from fastmcp.middleware import BaseMiddleware
class CustomMiddleware(BaseMiddleware):
    async def on_call_tool(self, tool_name, arguments, context):
        print(f"Before: {tool_name}")
        result = await self.next(tool_name, arguments, context)  # MUST call next()
        print(f"After: {tool_name}")
        return result

main_server.import_server(vendor_server)  # Static bundle

main_server.mount(api_server, prefix="api")  # Changes appear immediately

# ✅ Cloud-ready pattern
mcp = FastMCP("My Server")  # Module level
@mcp.tool()
async def my_tool(): pass
if __name__ == "__main__":
    mcp.run()

fastmcp deploy server.py

# ❌ WRONG
def create_server():
    return FastMCP("server")
# ✅ CORRECT
mcp = FastMCP("server")  # At module level

# ❌ WRONG: Sync function calling async
@mcp.tool()
def bad_tool():
    result = await async_function()  # Error!
# ✅ CORRECT: Async tool
@mcp.tool()
async def good_tool():
    result = await async_function()
    return result

from fastmcp import Context
# ❌ WRONG: No type hint
@mcp.tool()
async def bad_tool(context):  # Missing type!
    await context.report_progress(...)
# ✅ CORRECT: Proper type hint
@mcp.tool()
async def good_tool(context: Context):
    await context.report_progress(0, 100, "Starting")

# ❌ WRONG: Factory function creating circular dependency
# shared/__init__.py
def get_client():
    from .api_client import APIClient  # Circular!
    return APIClient()
# ✅ CORRECT: Direct imports
# shared/__init__.py
from .api_client import APIClient
from .cache import CacheManager
# shared/monitoring.py
from .api_client import APIClient
client = APIClient()

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

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["server.py"]
    }
  }
}

fastmcp dev server.py              # Development mode with hot reload
fastmcp run server.py              # Production mode
fastmcp deploy server.py           # Deploy to FastMCP Cloud
fastmcp test server.py             # Run tests

{
  "dependencies": {
    "fastmcp": ">=2.13.0",
    "pydantic": ">=2.0.0"
  },
  "optionalDependencies": {
    "py-key-value-aio": ">=0.1.0",  // For storage backends
    "cryptography": ">=41.0.0",     // For encryption
    "redis": ">=5.0.0"              // For Redis storage
  }
}