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 datafile://- File resourcesresource://- General resourcesinfo://- Information/metadataapi://- 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
en