Build MCP servers in Python with FastMCP to expose tools, resources, and prompts to LLMs. Supports storage backends, middleware, OAuth Proxy, OpenAPI integration, and FastMCP Cloud deployment. Prevents 25+ errors. Use when: creating MCP servers, or troubleshooting module-level server, storage, lifespan, middleware, or OAuth errors.
/plugin marketplace add jezweb/claude-skills/plugin install jezweb-tooling-skills@jezweb/claude-skillsThis skill inherits all available tools. When active, it can use any tool Claude has access to.
README.mdagents/mcp-scaffold.mdreferences/cli-commands.mdreferences/cloud-deployment.mdreferences/common-errors.mdreferences/context-features.mdreferences/integration-patterns.mdreferences/production-patterns.mdrules/fastmcp.mdscripts/check-versions.shscripts/deploy-cloud.shscripts/test-server.shtemplates/api-client-pattern.pytemplates/basic-server.pytemplates/client-example.pytemplates/error-handling.pytemplates/openapi-integration.pytemplates/prompts-examples.pytemplates/pyproject.tomltemplates/requirements.txtFastMCP 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.
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()
Run it:
# Local development
python server.py
# With FastMCP CLI
fastmcp dev server.py
# HTTP mode
python server.py --transport http --port 8000
auth_route parameter$ref resolution, OAuth Proxy validation, OpenAPI 3.1 supportctx.sample() now accepts tools for agentic workflowsAnthropicSamplingHandler promoted from experimentalctx.sample_step() for single LLM call returning SampleSteptask=True to async decorators; progress tracking without blocking⚠️ Breaking Changes (v2.14.0):
BearerAuthProvider module removed (use JWTVerifier or OAuthProxy)Context.get_http_request() method removedfastmcp.Image top-level import removed (use from fastmcp.utilities import Image)enable_docket, enable_tasks settings removed (always enabled)run_streamable_http_async(), sse_app(), streamable_http_app(), run_sse_async() methods removeddependencies parameter removed from decoratorsoutput_schema=False support eliminatedFASTMCP_SERVER_ environment variable prefix deprecatedKnown Compatibility:
Functions LLMs can call. Best practices: Clear names, comprehensive docstrings (LLMs read these!), strong type hints (Pydantic validates), structured returns, error handling.
@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()
Expose data to LLMs. URI schemes: data://, file://, resource://, info://, api://, or custom.
@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)
Pre-configured prompts with parameters.
@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
return f"Analyze {topic} considering: state, challenges, opportunities, recommendations."
Inject Context parameter (with type hint!) for advanced features:
Elicitation (User Input):
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:
@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):
@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"]
Long-running operations that report progress without blocking clients. Uses Docket task scheduler (always enabled in v2.14.0+).
Basic Usage:
@mcp.tool(task=True) # Enable background task mode
async def analyze_large_dataset(dataset_id: str, context: Context) -> dict:
"""Analyze large dataset with progress tracking."""
data = await fetch_dataset(dataset_id)
for i, chunk in enumerate(data.chunks):
# Report progress to client
await context.report_progress(
current=i + 1,
total=len(data.chunks),
message=f"Processing chunk {i + 1}/{len(data.chunks)}"
)
await process_chunk(chunk)
return {"status": "complete", "records_processed": len(data)}
Task States: pending → running → completed / failed / cancelled
When to Use:
Important: Tasks execute through Docket scheduler. Cannot execute tasks through proxies (will raise error).
Servers can pass tools to ctx.sample() for agentic workflows where the LLM can call tools during sampling.
Agentic Sampling:
from fastmcp import Context
from fastmcp.sampling import AnthropicSamplingHandler
# Configure sampling handler
mcp = FastMCP("Agent Server")
mcp.add_sampling_handler(AnthropicSamplingHandler(api_key=os.getenv("ANTHROPIC_API_KEY")))
@mcp.tool()
async def research_topic(topic: str, context: Context) -> dict:
"""Research a topic using agentic sampling with tools."""
# Define tools available during sampling
research_tools = [
{
"name": "search_web",
"description": "Search the web for information",
"inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}
},
{
"name": "fetch_url",
"description": "Fetch content from a URL",
"inputSchema": {"type": "object", "properties": {"url": {"type": "string"}}}
}
]
# Sample with tools - LLM can call these tools during reasoning
result = await context.sample(
messages=[{"role": "user", "content": f"Research: {topic}"}],
tools=research_tools,
max_tokens=4096
)
return {"research": result.content, "tools_used": result.tool_calls}
Single-Step Sampling:
@mcp.tool()
async def get_single_response(prompt: str, context: Context) -> dict:
"""Get a single LLM response without tool loop."""
# sample_step() returns SampleStep for inspection
step = await context.sample_step(
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
return {
"content": step.content,
"model": step.model,
"stop_reason": step.stop_reason
}
Sampling Handlers:
AnthropicSamplingHandler - For Claude models (v2.14.1+)OpenAISamplingHandler - For GPT modelsBuilt on py-key-value-aio for OAuth tokens, response caching, persistent state.
Available Backends:
FernetEncryptionWrapper, platform-aware (Mac/Windows default)Basic Usage:
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.
⚠️ 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.
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):
mcp = FastMCP("Server", lifespan=mcp_lifespan)
app = FastAPI(lifespan=mcp.lifespan) # ✅ MUST pass lifespan!
State Management:
context.fastmcp_context.set_state(key, value) # Store
context.fastmcp_context.get_state(key, default=None) # Retrieve
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:
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:
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)
Two Strategies:
import_server() - Static snapshot: One-time copy at import, changes don't propagate, fast (no runtime delegation). Use for: Finalized component bundles.
mount() - Dynamic link: Live runtime link, changes immediately visible, runtime delegation (slower). Use for: Modular runtime composition.
Basic Usage:
# 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:
@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:
resource://prefix/pathprefix+resource://pathmain_server.mount(subserver, prefix="api", resource_prefix_format="path")
4 Authentication Patterns:
JWTVerifier): Validate external tokensRemoteAuthProvider): OAuth 2.0/OIDC with DCROAuthProxy): Bridge to providers without DCR (GitHub, Google, Azure, AWS, Discord, Facebook)OAuthProvider): Complete authorization serverPattern 1: Token Validation
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)
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: 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):
httpx.AsyncClient with base_url/headers/timeoutFastMCP.from_openapi(spec, client, route_maps) - GET→Resources/Templates, POST/PUT/DELETE→ToolsFastMCP.from_fastapi(app, httpx_client_kwargs)Cloud Deployment Critical Requirements:
mcp, server, or app# ✅ 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):
{"mcpServers": {"my-server": {"url": "https://project.fastmcp.app/mcp", "transport": "http"}}}
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: 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: TypeError: missing 1 required positional argument: 'context'
Cause: Missing Context type annotation
Solution: async def tool(context: Context) - type hint required!
Error: ValueError: Invalid resource URI: missing scheme
Cause: Resource URI missing scheme prefix
Solution: Use @mcp.resource("data://config") not @mcp.resource("config")
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: 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: 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: 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: 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: 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: 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: 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: 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: DeprecationWarning: datetime.utcnow() is deprecated
Cause: Using deprecated Python 3.12+ methods
Solution: Use datetime.now(timezone.utc) instead of datetime.utcnow()
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: 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: 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: RuntimeError: Rate limit not checked before caching
Cause: Incorrect middleware ordering (order matters!)
Solution: ErrorHandling → Timing → Logging → RateLimiting → ResponseCaching (this order)
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: 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: 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: 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: 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: 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: 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
Error: ImportError: cannot import name 'BearerAuthProvider' from 'fastmcp.auth'
Cause: BearerAuthProvider module removed in v2.14.0
Solution: Use JWTVerifier for token validation or OAuthProxy for full OAuth flows:
# Before (v2.13.x)
from fastmcp.auth import BearerAuthProvider
# After (v2.14.0+)
from fastmcp.auth import JWTVerifier
auth = JWTVerifier(issuer="...", audience="...", public_key="...")
Error: AttributeError: 'Context' object has no attribute 'get_http_request'
Cause: Context.get_http_request() method removed in v2.14.0
Solution: Access request info through middleware or use InitializeResult exposed to middleware
Error: ImportError: cannot import name 'Image' from 'fastmcp'
Cause: fastmcp.Image top-level import removed in v2.14.0
Solution: Use new import path:
# Before (v2.13.x)
from fastmcp import Image
# After (v2.14.0+)
from fastmcp.utilities import Image
4 Production Patterns:
utils.py with Config class, format_success/error helpershttpx.AsyncClient with get_client() class methodretry_with_backoff(func, max_retries=3, initial_delay=1.0, exponential_base=2.0)TimeBasedCache(ttl=300) with .get() and .set() methodsTesting:
pytest + create_test_client(test_server) + await client.call_tool()Client("server.py") + list_tools() + call_tool() + list_resources()CLI Commands:
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:
server.py, requirements.txt, .env, README.mdsrc/ (server.py, utils.py, tools/, resources/, prompts/), tests/, pyproject.tomlOfficial: https://github.com/jlowin/fastmcp, https://fastmcp.cloud, https://modelcontextprotocol.io, Context7: /jlowin/fastmcp
Related Skills: openai-api, claude-api, cloudflare-worker-base, typescript-mcp
Package Versions: fastmcp>=2.14.2 (PyPI), Python>=3.10, httpx, pydantic, py-key-value-aio, cryptography
Last Updated: 2026-01-09
17 Key Takeaways:
import_server() (static) vs mount() (dynamic)fastmcp dev)task=True)ctx.sample(tools=[...]))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, background tasks, agentic sampling
Prevents 25+ errors. 90-95% token savings.
This skill should be used when the user asks to "create a slash command", "add a command", "write a custom command", "define command arguments", "use command frontmatter", "organize commands", "create command with file references", "interactive command", "use AskUserQuestion in command", or needs guidance on slash command structure, YAML frontmatter fields, dynamic arguments, bash execution in commands, user interaction patterns, or command development best practices for Claude Code.
This skill should be used when the user asks to "create an agent", "add an agent", "write a subagent", "agent frontmatter", "when to use description", "agent examples", "agent tools", "agent colors", "autonomous agent", or needs guidance on agent structure, system prompts, triggering conditions, or agent development best practices for Claude Code plugins.
This skill should be used when the user asks to "create a hook", "add a PreToolUse/PostToolUse/Stop hook", "validate tool use", "implement prompt-based hooks", "use ${CLAUDE_PLUGIN_ROOT}", "set up event-driven automation", "block dangerous commands", or mentions hook events (PreToolUse, PostToolUse, Stop, SubagentStop, SessionStart, SessionEnd, UserPromptSubmit, PreCompact, Notification). Provides comprehensive guidance for creating and implementing Claude Code plugin hooks with focus on advanced prompt-based hooks API.