MCP Server Skill
Build MCP (Model Context Protocol) servers using the official Python SDK with FastMCP high-level API.
Architecture
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Server (FastMCP) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ @tool() │ │ @resource() │ │ @prompt() │ │
│ │ add_task │ │ tasks:// │ │ task_prompt │ │
│ │ list_tasks │ │ user:// │ │ help_prompt │ │
│ │complete_task│ │ │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └─────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Database Layer (SQLModel) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
│ Transports: stdio | SSE | streamable-http
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Client (Agent) │
│ OpenAI Agents SDK / Claude / Other Clients │
└─────────────────────────────────────────────────────────────────────────┘
Quick Start
Installation
# pip
pip install mcp
# poetry
poetry add mcp
# uv
uv add mcp
Environment Variables
DATABASE_URL=postgresql://user:password@localhost:5432/mydb
Core Concepts
| Concept | Decorator | Purpose |
|---------|-----------|---------|
| Tools | @mcp.tool() | Perform actions, computations, side effects |
| Resources | @mcp.resource() | Expose data for reading (URI-based) |
| Prompts | @mcp.prompt() | Reusable prompt templates |
Basic MCP Server
from mcp.server.fastmcp import FastMCP
# Create server instance
mcp = FastMCP("Todo Server", json_response=True)
# Define a tool
@mcp.tool()
def add_task(title: str, description: str = None) -> dict:
"""Add a new task to the todo list."""
# Implementation here
return {"task_id": 1, "status": "created"}
# Define a resource
@mcp.resource("tasks://{user_id}")
def get_user_tasks(user_id: str) -> str:
"""Get all tasks for a user."""
return "task data as string"
# Define a prompt
@mcp.prompt()
def task_assistant(task_type: str = "general") -> str:
"""Generate a task management prompt."""
return f"You are a helpful {task_type} task assistant."
# Run the server
if __name__ == "__main__":
mcp.run(transport="streamable-http")
Reference
| Pattern | Guide | |---------|-------| | Tools | reference/tools.md | | Resources | reference/resources.md | | Prompts | reference/prompts.md | | Transports | reference/transports.md | | FastAPI Integration | reference/fastapi-integration.md |
Examples
| Example | Description | |---------|-------------| | examples/todo-server.md | Complete todo MCP server with CRUD tools | | examples/database-integration.md | MCP server with SQLModel database |
Templates
| Template | Purpose | |----------|---------| | templates/mcp_server.py | Basic MCP server template | | templates/mcp_tools.py | Tool definitions template | | templates/mcp_fastapi.py | FastAPI + MCP integration template |
FastAPI/Starlette Integration
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCP
# Create MCP server
mcp = FastMCP("Todo Server", stateless_http=True, json_response=True)
@mcp.tool()
def add_task(title: str) -> dict:
"""Add a task."""
return {"status": "created"}
# Lifespan manager for session handling
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
async with mcp.session_manager.run():
yield
# Create Starlette app with MCP mounted
app = Starlette(
routes=[
Mount("/mcp", app=mcp.streamable_http_app()),
],
lifespan=lifespan,
)
# Add CORS for browser clients
app = CORSMiddleware(
app,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE"],
expose_headers=["Mcp-Session-Id"],
)
# Run with: uvicorn server:app --reload
# MCP endpoint: http://localhost:8000/mcp/mcp
Tool with Context (Progress, Logging)
from mcp.server.fastmcp import Context, FastMCP
mcp = FastMCP("Progress Server")
@mcp.tool()
async def long_task(steps: int, ctx: Context) -> str:
"""Execute a task with progress updates."""
await ctx.info("Starting task...")
for i in range(steps):
progress = (i + 1) / steps
await ctx.report_progress(
progress=progress,
total=1.0,
message=f"Step {i + 1}/{steps}",
)
await ctx.debug(f"Completed step {i + 1}")
return f"Task completed in {steps} steps"
Database Integration with Lifespan
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppContext:
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP):
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
await db.disconnect()
mcp = FastMCP("DB Server", lifespan=app_lifespan)
@mcp.tool()
def query_tasks(user_id: str, ctx: Context) -> list:
"""Query tasks from database."""
app_ctx = ctx.request_context.lifespan_context
return app_ctx.db.query(f"SELECT * FROM tasks WHERE user_id = '{user_id}'")
Transport Options
| Transport | Use Case | Command |
|-----------|----------|---------|
| stdio | CLI tools, local agents | mcp.run() or mcp.run(transport="stdio") |
| SSE | Web clients, real-time | mcp.run(transport="sse", port=8000) |
| streamable-http | Production APIs | mcp.run(transport="streamable-http") |
Stateless vs Stateful
# Stateless (recommended for production)
mcp = FastMCP("Server", stateless_http=True, json_response=True)
# Stateful (maintains session state)
mcp = FastMCP("Server")
Security Considerations
- Validate all inputs - Never trust user-provided data
- Use parameterized queries - Prevent SQL injection
- Authenticate requests - Verify user identity before operations
- Limit resource access - Only expose necessary data
- Log tool invocations - Audit trail for debugging
Troubleshooting
Server won't start
- Check port availability
- Verify dependencies installed
- Check for syntax errors in tool definitions
Client can't connect
- Verify transport matches (stdio/SSE/HTTP)
- Check CORS configuration for web clients
- Ensure MCP endpoint URL is correct
Tools not appearing
- Verify
@mcp.tool()decorator is applied - Check function has docstring (used as description)
- Restart server after adding new tools