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