Files
2025-11-30 08:24:49 +08:00

575 lines
22 KiB
Markdown

---
name: fastmcp
description: |
Build MCP servers in Python with FastMCP framework to expose tools, resources, and prompts to LLMs. Supports
storage backends (memory/disk/Redis), middleware, OAuth Proxy, OpenAPI integration, and FastMCP Cloud deployment.
Use when: creating MCP servers, defining tools or resources, implementing OAuth authentication, configuring
storage backends for tokens/cache, adding middleware for logging/rate limiting, deploying to FastMCP Cloud,
or troubleshooting module-level server, storage, lifespan, middleware order, circular imports, or OAuth errors.
Keywords: FastMCP, MCP server Python, Model Context Protocol Python, fastmcp framework, mcp tools, mcp resources, mcp prompts, fastmcp storage, fastmcp memory storage, fastmcp disk storage, fastmcp redis, fastmcp dynamodb, fastmcp lifespan, fastmcp middleware, fastmcp oauth proxy, server composition mcp, fastmcp import, fastmcp mount, fastmcp cloud, fastmcp deployment, mcp authentication, fastmcp icons, openapi mcp, claude mcp server, fastmcp testing, storage misconfiguration, lifespan issues, middleware order, circular imports, module-level server, async await mcp
license: MIT
metadata:
version: "2.0.0"
package_version: "fastmcp>=2.13.1"
python_version: ">=3.10"
token_savings: "90-95%"
errors_prevented: 25
production_tested: true
last_updated: "2025-11-25"
---
# 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
```bash
pip install fastmcp
# or
uv pip install fastmcp
```
### Minimal Server
```python
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:**
```bash
# Local development
python server.py
# With FastMCP CLI
fastmcp dev server.py
# HTTP mode
python server.py --transport http --port 8000
```
## What's New in v2.13.1 (November 2025)
**Meta Parameter Support:**
- ToolResult can return metadata alongside results (enables OpenAI Apps SDK integration)
- Client-sent meta parameters now supported
**Authentication Improvements:**
- `DebugTokenVerifier` for custom token validation during development
- OCI (Oracle Cloud Infrastructure) authentication provider added
- Enhanced OAuth error handling and messaging
- Improved CSP policies for OAuth consent screens
**Utilities & Developer Experience:**
- `Image.to_data_uri()` method added for easier icon embedding
- Manual Client initialization control (defer connection until needed)
- 20+ bug fixes: URL encoding in Cursor deeplinks, OAuth metadata endpoint handling, Windows test timeouts, token cache expiration
**Security Update:**
- **CVE-2025-61920**: authlib updated to 1.6.5
- Safer Windows API validation for Cursor deeplink URLs
**Known Compatibility:**
- MCP SDK 1.21.1 excluded due to integration test failures (use 1.21.0 or 1.22.0+)
---
## Core Concepts
### Tools
Functions LLMs can call. Best practices: Clear names, comprehensive docstrings (LLMs read these!), strong type hints (Pydantic validates), structured returns, error handling.
```python
@mcp.tool()
async def async_tool(url: str) -> dict: # Use async for I/O
async with httpx.AsyncClient() as client:
return (await client.get(url)).json()
```
### Resources
Expose data to LLMs. URI schemes: `data://`, `file://`, `resource://`, `info://`, `api://`, or custom.
```python
@mcp.resource("user://{user_id}/profile") # Template with parameters
async def get_user(user_id: str) -> dict: # CRITICAL: param names must match
return await fetch_user_from_db(user_id)
```
### Prompts
Pre-configured prompts with parameters.
```python
@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
return f"Analyze {topic} considering: state, challenges, opportunities, recommendations."
```
## Context Features
Inject `Context` parameter (with type hint!) for advanced features:
**Elicitation (User Input):**
```python
from fastmcp import Context
@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
confirmed = await context.request_elicitation(prompt=f"Confirm {action}?", response_type=str)
return {"status": "completed" if confirmed.lower() == "yes" else "cancelled"}
```
**Progress Tracking:**
```python
@mcp.tool()
async def batch_import(file_path: str, context: Context) -> dict:
data = await read_file(file_path)
for i, item in enumerate(data):
await context.report_progress(i + 1, len(data), f"Importing {i + 1}/{len(data)}")
await import_item(item)
return {"imported": len(data)}
```
**Sampling (LLM calls from tools):**
```python
@mcp.tool()
async def enhance_text(text: str, context: Context) -> str:
response = await context.request_sampling(
messages=[{"role": "user", "content": f"Enhance: {text}"}],
temperature=0.7
)
return response["content"]
```
## Storage Backends
Built on `py-key-value-aio` for OAuth tokens, response caching, persistent state.
**Available Backends:**
- **Memory** (default): Ephemeral, fast, dev-only
- **Disk**: Persistent, encrypted with `FernetEncryptionWrapper`, platform-aware (Mac/Windows default)
- **Redis**: Distributed, production, multi-instance
- **Others**: DynamoDB, MongoDB, Elasticsearch, Memcached, RocksDB, Valkey
**Basic Usage:**
```python
from key_value.stores import DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Disk (persistent, single instance)
mcp = FastMCP("Server", storage=DiskStore(path="/app/data/storage"))
# Redis (distributed, production)
mcp = FastMCP("Server", storage=RedisStore(
host=os.getenv("REDIS_HOST"), password=os.getenv("REDIS_PASSWORD")
))
# Encrypted storage (recommended)
mcp = FastMCP("Server", storage=FernetEncryptionWrapper(
key_value=DiskStore(path="/app/data"),
fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
))
```
**Platform Defaults:** Mac/Windows use Disk, Linux uses Memory. Override with `storage` parameter.
## Server Lifespans
**⚠️ Breaking Change in v2.13.0**: Lifespan behavior changed from per-session to per-server-instance.
Initialize/cleanup resources once per server (NOT per session) - critical for DB connections, API clients.
```python
from contextlib import asynccontextmanager
from dataclasses import dataclass
@dataclass
class AppContext:
db: Database
api_client: httpx.AsyncClient
@asynccontextmanager
async def app_lifespan(server: FastMCP):
"""Runs ONCE per server instance."""
db = await Database.connect(os.getenv("DATABASE_URL"))
api_client = httpx.AsyncClient(base_url=os.getenv("API_BASE_URL"), timeout=30.0)
try:
yield AppContext(db=db, api_client=api_client)
finally:
await db.disconnect()
await api_client.aclose()
mcp = FastMCP("Server", lifespan=app_lifespan)
# Access in tools
@mcp.tool()
async def query_db(sql: str, context: Context) -> list:
app_ctx = context.fastmcp_context.lifespan_context
return await app_ctx.db.query(sql)
```
**ASGI Integration (FastAPI/Starlette):**
```python
mcp = FastMCP("Server", lifespan=mcp_lifespan)
app = FastAPI(lifespan=mcp.lifespan) # ✅ MUST pass lifespan!
```
**State Management:**
```python
context.fastmcp_context.set_state(key, value) # Store
context.fastmcp_context.get_state(key, default=None) # Retrieve
```
## Middleware System
**8 Built-in Types:** TimingMiddleware, ResponseCachingMiddleware, LoggingMiddleware, RateLimitingMiddleware, ErrorHandlingMiddleware, ToolInjectionMiddleware, PromptToolMiddleware, ResourceToolMiddleware
**Execution Order (order matters!):**
```
Request Flow:
→ ErrorHandlingMiddleware (catches errors)
→ TimingMiddleware (starts timer)
→ LoggingMiddleware (logs request)
→ RateLimitingMiddleware (checks rate limit)
→ ResponseCachingMiddleware (checks cache)
→ Tool/Resource Handler
```
**Basic Usage:**
```python
from fastmcp.middleware import ErrorHandlingMiddleware, TimingMiddleware, LoggingMiddleware
mcp.add_middleware(ErrorHandlingMiddleware()) # First: catch errors
mcp.add_middleware(TimingMiddleware()) # Second: time requests
mcp.add_middleware(LoggingMiddleware(level="INFO"))
mcp.add_middleware(RateLimitingMiddleware(max_requests=100, window_seconds=60))
mcp.add_middleware(ResponseCachingMiddleware(ttl_seconds=300, storage=RedisStore()))
```
**Custom Middleware:**
```python
from fastmcp.middleware import BaseMiddleware
class AccessControlMiddleware(BaseMiddleware):
async def on_call_tool(self, tool_name, arguments, context):
user = context.fastmcp_context.get_state("user_id")
if user not in self.allowed_users:
raise PermissionError(f"User not authorized")
return await self.next(tool_name, arguments, context)
```
**Hook Hierarchy:** `on_message` (all) → `on_request`/`on_notification``on_call_tool`/`on_read_resource`/`on_get_prompt``on_list_*` (list operations)
## Server Composition
**Two Strategies:**
1. **`import_server()`** - Static snapshot: One-time copy at import, changes don't propagate, fast (no runtime delegation). Use for: Finalized component bundles.
2. **`mount()`** - Dynamic link: Live runtime link, changes immediately visible, runtime delegation (slower). Use for: Modular runtime composition.
**Basic Usage:**
```python
# Import (static)
main_server.import_server(api_server) # One-time copy
# Mount (dynamic)
main_server.mount(api_server, prefix="api") # Tools: api.fetch_data
main_server.mount(db_server, prefix="db") # Resources: resource://db/path
```
**Tag Filtering:**
```python
@api_server.tool(tags=["public"])
def public_api(): pass
main_server.import_server(api_server, include_tags=["public"]) # Only public
main_server.mount(api_server, prefix="api", exclude_tags=["admin"]) # No admin
```
**Resource Prefix Formats:**
- **Path** (default since v2.4.0): `resource://prefix/path`
- **Protocol** (legacy): `prefix+resource://path`
```python
main_server.mount(subserver, prefix="api", resource_prefix_format="path")
```
## OAuth & Authentication
**4 Authentication Patterns:**
1. **Token Validation** (`JWTVerifier`): Validate external tokens
2. **External Identity Providers** (`RemoteAuthProvider`): OAuth 2.0/OIDC with DCR
3. **OAuth Proxy** (`OAuthProxy`): Bridge to providers without DCR (GitHub, Google, Azure, AWS, Discord, Facebook)
4. **Full OAuth** (`OAuthProvider`): Complete authorization server
**Pattern 1: Token Validation**
```python
from fastmcp.auth import JWTVerifier
auth = JWTVerifier(issuer="https://auth.example.com", audience="my-server",
public_key=os.getenv("JWT_PUBLIC_KEY"))
mcp = FastMCP("Server", auth=auth)
```
**Pattern 3: OAuth Proxy (Production)**
```python
from fastmcp.auth import OAuthProxy
from key_value.stores import RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
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://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"),
enable_consent_screen=True # CRITICAL: Prevents confused deputy attacks
)
mcp = FastMCP("GitHub Auth", auth=auth)
```
**OAuth Proxy Features:** Token factory pattern (issues own JWTs), consent screens (prevents bypass), PKCE support, RFC 7662 token introspection
**Supported Providers:** GitHub, Google, Azure, AWS Cognito, Discord, Facebook, WorkOS, AuthKit, Descope, Scalekit, OCI (v2.13.1)
## Icons, API Integration, Cloud Deployment
**Icons:** Add to servers, tools, resources, prompts. Use `Icon(url, size)`, data URIs via `Icon.from_file()` or `Image.to_data_uri()` (v2.13.1).
**API Integration (3 Patterns):**
1. **Manual**: `httpx.AsyncClient` with base_url/headers/timeout
2. **OpenAPI Auto-Gen**: `FastMCP.from_openapi(spec, client, route_maps)` - GET→Resources/Templates, POST/PUT/DELETE→Tools
3. **FastAPI Conversion**: `FastMCP.from_fastapi(app, httpx_client_kwargs)`
**Cloud Deployment Critical Requirements:**
1.**Module-level server** named `mcp`, `server`, or `app`
2. **PyPI dependencies only** in requirements.txt
3. **Public GitHub repo** (or accessible)
4. **Environment variables** for config
```python
# ✅ CORRECT: Module-level export
mcp = FastMCP("server") # At module level!
# ❌ WRONG: Function-wrapped
def create_server():
return FastMCP("server") # Too late for cloud!
```
**Deployment:** https://fastmcp.cloud → Sign in → Create Project → Select repo → Deploy
**Client Config (Claude Desktop):**
```json
{"mcpServers": {"my-server": {"url": "https://project.fastmcp.app/mcp", "transport": "http"}}}
```
## 25 Common Errors (With Solutions)
### Error 1: Missing Server Object
**Error:** `RuntimeError: No server object found at module level`
**Cause:** Server not exported at module level (FastMCP Cloud requirement)
**Solution:** `mcp = FastMCP("server")` at module level, not inside functions
### Error 2: Async/Await Confusion
**Error:** `RuntimeError: no running event loop`, `TypeError: object coroutine can't be used in 'await'`
**Cause:** Mixing sync/async incorrectly
**Solution:** Use `async def` for tools with `await`, sync `def` for non-async code
### Error 3: Context Not Injected
**Error:** `TypeError: missing 1 required positional argument: 'context'`
**Cause:** Missing `Context` type annotation
**Solution:** `async def tool(context: Context)` - type hint required!
### Error 4: Resource URI Syntax
**Error:** `ValueError: Invalid resource URI: missing scheme`
**Cause:** Resource URI missing scheme prefix
**Solution:** Use `@mcp.resource("data://config")` not `@mcp.resource("config")`
### Error 5: Resource Template Parameter Mismatch
**Error:** `TypeError: get_user() missing 1 required positional argument`
**Cause:** Function parameter names don't match URI template
**Solution:** `@mcp.resource("user://{user_id}/profile")``def get_user(user_id: str)` - names must match exactly
---
### Error 6: Pydantic Validation Error
**Error:** `ValidationError: value is not a valid integer`
**Cause:** Type hints don't match provided data
**Solution:** Use Pydantic models: `class Params(BaseModel): query: str = Field(min_length=1)`
### Error 7: Transport/Protocol Mismatch
**Error:** `ConnectionError: Server using different transport`
**Cause:** Client and server using incompatible transports
**Solution:** Match transports - stdio: `mcp.run()` + `{"command": "python", "args": ["server.py"]}`, HTTP: `mcp.run(transport="http", port=8000)` + `{"url": "http://localhost:8000/mcp", "transport": "http"}`
### Error 8: Import Errors (Editable Package)
**Error:** `ModuleNotFoundError: No module named 'my_package'`
**Cause:** Package not properly installed
**Solution:** `pip install -e .` or use absolute imports or `export PYTHONPATH="/path/to/project"`
### Error 9: Deprecation Warnings
**Error:** `DeprecationWarning: 'mcp.settings' is deprecated`
**Cause:** Using old FastMCP v1 API
**Solution:** Use `os.getenv("API_KEY")` instead of `mcp.settings.get("API_KEY")`
### Error 10: Port Already in Use
**Error:** `OSError: [Errno 48] Address already in use`
**Cause:** Port 8000 already occupied
**Solution:** Use different port `--port 8001` or kill process `lsof -ti:8000 | xargs kill -9`
### Error 11: Schema Generation Failures
**Error:** `TypeError: Object of type 'ndarray' is not JSON serializable`
**Cause:** Unsupported type hints (NumPy arrays, custom classes)
**Solution:** Return JSON-compatible types: `list[float]` or convert: `{"values": np_array.tolist()}`
### Error 12: JSON Serialization
**Error:** `TypeError: Object of type 'datetime' is not JSON serializable`
**Cause:** Returning non-JSON-serializable objects
**Solution:** Convert: `datetime.now().isoformat()`, bytes: `.decode('utf-8')`
### Error 13: Circular Import Errors
**Error:** `ImportError: cannot import name 'X' from partially initialized module`
**Cause:** Circular dependency (common in cloud deployment)
**Solution:** Use direct imports in `__init__.py`: `from .api_client import APIClient` or lazy imports in functions
### Error 14: Python Version Compatibility
**Error:** `DeprecationWarning: datetime.utcnow() is deprecated`
**Cause:** Using deprecated Python 3.12+ methods
**Solution:** Use `datetime.now(timezone.utc)` instead of `datetime.utcnow()`
### Error 15: Import-Time Execution
**Error:** `RuntimeError: Event loop is closed`
**Cause:** Creating async resources at module import time
**Solution:** Use lazy initialization - create connection class with async `connect()` method, call when needed in tools
---
### Error 16: Storage Backend Not Configured
**Error:** `RuntimeError: OAuth tokens lost on restart`, `ValueError: Cache not persisting`
**Cause:** Using default memory storage in production without persistence
**Solution:** Use encrypted DiskStore (single instance) or RedisStore (multi-instance) with `FernetEncryptionWrapper`
### Error 17: Lifespan Not Passed to ASGI App
**Error:** `RuntimeError: Database connection never initialized`, `Warning: MCP lifespan hooks not running`
**Cause:** FastMCP with FastAPI/Starlette without passing lifespan (v2.13.0 requirement)
**Solution:** `app = FastAPI(lifespan=mcp.lifespan)` - MUST pass lifespan!
### Error 18: Middleware Execution Order Error
**Error:** `RuntimeError: Rate limit not checked before caching`
**Cause:** Incorrect middleware ordering (order matters!)
**Solution:** ErrorHandling → Timing → Logging → RateLimiting → ResponseCaching (this order)
### Error 19: Circular Middleware Dependencies
**Error:** `RecursionError: maximum recursion depth exceeded`
**Cause:** Middleware not calling `self.next()` or calling incorrectly
**Solution:** Always call `result = await self.next(tool_name, arguments, context)` in middleware hooks
### Error 20: Import vs Mount Confusion
**Error:** `RuntimeError: Subserver changes not reflected`, `ValueError: Unexpected tool namespacing`
**Cause:** Using `import_server()` when `mount()` was needed (or vice versa)
**Solution:** `import_server()` for static bundles (one-time copy), `mount()` for dynamic composition (live link)
### Error 21: Resource Prefix Format Mismatch
**Error:** `ValueError: Resource not found: resource://api/users`
**Cause:** Using wrong resource prefix format
**Solution:** Path format (default v2.4.0+): `resource://prefix/path`, Protocol (legacy): `prefix+resource://path` - set with `resource_prefix_format="path"`
### Error 22: OAuth Proxy Without Consent Screen
**Error:** `SecurityWarning: Authorization bypass possible`
**Cause:** OAuth Proxy without consent screen (security vulnerability)
**Solution:** Always set `enable_consent_screen=True` - prevents confused deputy attacks (CRITICAL)
### Error 23: Missing JWT Signing Key in Production
**Error:** `ValueError: JWT signing key required for OAuth Proxy`
**Cause:** OAuth Proxy missing `jwt_signing_key`
**Solution:** Generate: `secrets.token_urlsafe(32)`, store in `FASTMCP_JWT_SIGNING_KEY` env var, pass to `OAuthProxy(jwt_signing_key=...)`
### Error 24: Icon Data URI Format Error
**Error:** `ValueError: Invalid data URI format`
**Cause:** Incorrectly formatted data URI for icons
**Solution:** Use `Icon.from_file("/path/icon.png", size="medium")` or `Image.to_data_uri()` (v2.13.1) - don't manually format
### Error 25: Lifespan Behavior Change (v2.13.0)
**Error:** `Warning: Lifespan runs per-server, not per-session`
**Cause:** Expecting v2.12 behavior (per-session) in v2.13.0+ (per-server)
**Solution:** v2.13.0+ lifespans run ONCE per server, not per session - use middleware for per-session logic
---
## Production Patterns, Testing, CLI
**4 Production Patterns:**
1. **Utils Module**: Single `utils.py` with Config class, format_success/error helpers
2. **Connection Pooling**: Singleton `httpx.AsyncClient` with `get_client()` class method
3. **Retry with Backoff**: `retry_with_backoff(func, max_retries=3, initial_delay=1.0, exponential_base=2.0)`
4. **Time-Based Caching**: `TimeBasedCache(ttl=300)` with `.get()` and `.set()` methods
**Testing:**
- Unit: `pytest` + `create_test_client(test_server)` + `await client.call_tool()`
- Integration: `Client("server.py")` + `list_tools()` + `call_tool()` + `list_resources()`
**CLI Commands:**
```bash
fastmcp dev server.py # Run with inspector
fastmcp install server.py # Install to Claude Desktop
FASTMCP_LOG_LEVEL=DEBUG fastmcp dev # Debug logging
```
**Best Practices:** Factory pattern with module-level export, environment config with validation, comprehensive docstrings (LLMs read these!), health check resources
**Project Structure:**
- Simple: `server.py`, `requirements.txt`, `.env`, `README.md`
- Production: `src/` (server.py, utils.py, tools/, resources/, prompts/), `tests/`, `pyproject.toml`
---
## References & Summary
**Official:** https://github.com/jlowin/fastmcp, https://fastmcp.cloud, https://modelcontextprotocol.io, Context7: `/jlowin/fastmcp`
**Related Skills:** openai-api, claude-api, cloudflare-worker-base
**Package Versions:** fastmcp>=2.13.1, Python>=3.10, httpx, pydantic, py-key-value-aio, cryptography
**15 Key Takeaways:**
1. Module-level server export (FastMCP Cloud)
2. Persistent storage (Disk/Redis) for OAuth/caching
3. Server lifespans for resource management
4. Middleware order: errors → timing → logging → rate limiting → caching
5. Composition: `import_server()` (static) vs `mount()` (dynamic)
6. OAuth security: consent screens + encrypted storage + JWT signing
7. Async/await properly (don't block event loop)
8. Structured error handling
9. Avoid circular imports
10. Test locally (`fastmcp dev`)
11. Environment variables (never hardcode secrets)
12. Comprehensive docstrings (LLMs read!)
13. Production patterns (utils, pooling, retry, caching)
14. OpenAPI auto-generation
15. Health checks + monitoring
**Production Readiness:** Encrypted storage, 4 auth patterns, 8 middleware types, modular composition, OAuth security (consent screens, PKCE, RFC 7662), response caching, connection pooling, timing middleware
**Prevents 25 errors. 90-95% token savings.**