The Model Context Protocol (MCP) is the standard interface for connecting AI models to external tools, data sources, and services. An MCP server exposes three primitives: tools (functions the model can call), resources (data the model can read), and prompts (reusable prompt templates). This skill covers building production-quality MCP servers with FastMCP.
Core Mental Model
Think of an MCP server as a typed API contract between your service and any MCP-compatible client (Claude Desktop, Claude API, custom agents). Tools are for actions and computation; resources are for data access; prompts are for workflow templates. The key design principle: tools should be composable and narrowly scoped. A tool that does one thing well can be composed by the model into complex workflows. A monolithic "do everything" tool is hard to use reliably.
MCP Specification Fundamentals
MCP Architecture:
┌─────────────────┐ ┌──────────────────────┐ ┌──────────────────┐
│ MCP Client │────▶│ MCP Server │────▶│ External APIs │
│ (Claude, Agent) │◀────│ (Your FastMCP app) │◀────│ Databases, etc. │
└─────────────────┘ └──────────────────────┘ └──────────────────┘
Primitives:
Tools → Functions with JSON Schema input, called on demand
Resources → URIs that expose readable content (text, JSON, binary)
Prompts → Named prompt templates with typed argumentsComplete FastMCP Server
# server.py — Production MCP server with tools, resources, and prompts
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from pydantic import BaseModel, Field
from typing import Optional
import httpx
import json
mcp = FastMCP(
name="company-tools",
version="1.0.0",
description="Company database and API integration tools",
)
# ============================================================
# TOOLS — Functions the model can invoke
# ============================================================
class CustomerLookupInput(BaseModel):
customer_id: str = Field(description="Customer ID (e.g., 'cust_abc123')")
include_history: bool = Field(default=False, description="Include purchase history")
@mcp.tool()
async def lookup_customer(params: CustomerLookupInput) -> dict:
"""
Look up a customer by ID. Returns profile, contact info, and optionally purchase history.
Use this when the user asks about a specific customer or needs to verify customer details.
"""
# Input validation — fail fast with clear error
if not params.customer_id.startswith("cust_"):
raise ToolError(
f"Invalid customer ID format: '{params.customer_id}'. Must start with 'cust_'"
)
async with httpx.AsyncClient() as client:
try:
response = await client.get(
f"https://api.company.com/customers/{params.customer_id}",
headers={"Authorization": f"Bearer {get_api_key()}"},
timeout=10.0,
)
response.raise_for_status()
except httpx.TimeoutException:
raise ToolError("Customer API timed out. Try again in a moment.")
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
raise ToolError(f"Customer '{params.customer_id}' not found.")
raise ToolError(f"API error: {e.response.status_code}")
data = response.json()
if params.include_history:
# Parallel fetch for efficiency
hist_response = await client.get(
f"https://api.company.com/customers/{params.customer_id}/orders",
headers={"Authorization": f"Bearer {get_api_key()}"},
)
data["order_history"] = hist_response.json()
return data
@mcp.tool()
async def search_knowledge_base(
query: str = Field(description="Search query"),
max_results: int = Field(default=5, ge=1, le=20, description="Max results to return"),
category: Optional[str] = Field(default=None, description="Filter by category"),
) -> list[dict]:
"""
Search the internal knowledge base. Returns ranked articles with title, snippet, and URL.
Use when answering questions about products, policies, or procedures.
"""
results = await kb_client.search(query, limit=max_results, category=category)
return [
{
"title": r.title,
"snippet": r.snippet[:500], # Limit snippet length
"url": r.url,
"relevance_score": r.score,
}
for r in results
]
@mcp.tool()
async def create_support_ticket(
customer_id: str = Field(description="Customer ID"),
subject: str = Field(description="Ticket subject (max 100 chars)"),
description: str = Field(description="Detailed problem description"),
priority: str = Field(description="Priority: low, medium, high, urgent"),
) -> dict:
"""
Create a support ticket. Returns ticket ID and estimated response time.
Use when a customer issue requires follow-up action beyond immediate resolution.
"""
if priority not in ("low", "medium", "high", "urgent"):
raise ToolError(f"Invalid priority '{priority}'. Must be: low, medium, high, urgent")
if len(subject) > 100:
raise ToolError(f"Subject too long ({len(subject)} chars). Max 100 characters.")
# Sanitize description — strip potentially dangerous content
clean_description = description[:5000] # Hard limit
ticket = await ticketing_system.create({
"customer_id": customer_id,
"subject": subject,
"description": clean_description,
"priority": priority,
})
return {
"ticket_id": ticket.id,
"status": "created",
"estimated_response": ticket.sla_deadline.isoformat(),
}
# ============================================================
# RESOURCES — Data sources the model can read
# ============================================================
@mcp.resource("company://docs/{doc_id}")
async def get_document(doc_id: str) -> str:
"""Retrieve a company document by ID. Returns markdown content."""
doc = await doc_store.get(doc_id)
if not doc:
raise ValueError(f"Document '{doc_id}' not found")
return doc.content_markdown
@mcp.resource("company://customers/{customer_id}/profile")
async def get_customer_profile_resource(customer_id: str) -> str:
"""Customer profile as formatted text for context injection."""
customer = await db.get_customer(customer_id)
return f"""
Customer: {customer.name}
ID: {customer.id}
Email: {customer.email}
Tier: {customer.subscription_tier}
Account Age: {customer.account_age_days} days
Total Orders: {customer.total_orders}
Lifetime Value: ${customer.lifetime_value:.2f}
"""
@mcp.resource("company://metrics/dashboard")
async def get_metrics_dashboard() -> str:
"""Current business metrics dashboard. Updates every 5 minutes."""
metrics = await metrics_client.get_current()
return json.dumps(metrics, indent=2)
# ============================================================
# PROMPTS — Reusable prompt templates
# ============================================================
@mcp.prompt()
def customer_escalation_prompt(
customer_id: str,
issue_summary: str,
previous_attempts: str,
) -> list[dict]:
"""Template for escalating a complex customer issue to a specialist."""
return [
{
"role": "user",
"content": f"""
Please handle this escalated customer issue requiring specialist attention.
Customer ID: {customer_id}
Issue Summary: {issue_summary}
Previous Resolution Attempts: {previous_attempts}
Please:
1. Look up the customer profile
2. Review relevant knowledge base articles
3. Propose a resolution
4. Create a support ticket if needed
""",
}
]
# ============================================================
# SERVER TRANSPORT
# ============================================================
if __name__ == "__main__":
# stdio for local tools (Claude Desktop, CLI agents)
mcp.run(transport="stdio")
# For remote/HTTP deployment:
# mcp.run(transport="sse", host="0.0.0.0", port=8080)Server Transports
# stdio: local process communication — simplest, for Claude Desktop integration
mcp.run(transport="stdio")
# SSE (Server-Sent Events): HTTP-based, supports remote connections
# Client connects via HTTP, server streams events
mcp.run(
transport="sse",
host="0.0.0.0",
port=8080,
# Configure CORS for browser clients
)
# Streamable HTTP: newer standard (MCP 2025-03-26+)
mcp.run(transport="streamable-http", port=8080)Claude Desktop Integration (stdio)
// ~/Library/Application Support/Claude/claude_desktop_config.json
{
"mcpServers": {
"company-tools": {
"command": "python",
"args": ["/path/to/server.py"],
"env": {
"COMPANY_API_KEY": "your-key-here",
"DATABASE_URL": "postgresql://..."
}
}
}
}Authentication for Remote Servers
from fastmcp import FastMCP
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
security = HTTPBearer()
def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
"""Validate JWT bearer token on all MCP requests."""
try:
payload = jwt.decode(
credentials.credentials,
key=os.environ["JWT_SECRET"],
algorithms=["HS256"],
)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
# FastMCP with auth middleware
mcp = FastMCP(name="authenticated-server")
# Add middleware to the underlying FastAPI app
mcp.app.add_middleware(AuthMiddleware, verify_fn=verify_token)Error Handling Patterns
from fastmcp.exceptions import ToolError
from enum import Enum
class ErrorCode(str, Enum):
NOT_FOUND = "NOT_FOUND"
INVALID_INPUT = "INVALID_INPUT"
RATE_LIMITED = "RATE_LIMITED"
UPSTREAM_ERROR = "UPSTREAM_ERROR"
PERMISSION_DENIED = "PERMISSION_DENIED"
def raise_tool_error(code: ErrorCode, message: str) -> None:
"""Standardized error format — models can parse and respond appropriately."""
raise ToolError(json.dumps({
"error_code": code.value,
"message": message,
"recoverable": code in (ErrorCode.RATE_LIMITED, ErrorCode.UPSTREAM_ERROR),
}))
@mcp.tool()
async def sensitive_operation(resource_id: str, user_context: dict) -> dict:
"""Example with comprehensive error handling."""
# Authorization check first
if not await check_permission(user_context["user_id"], "write", resource_id):
raise_tool_error(ErrorCode.PERMISSION_DENIED,
f"User lacks write permission for resource {resource_id}")
# Resource exists?
resource = await db.get(resource_id)
if not resource:
raise_tool_error(ErrorCode.NOT_FOUND, f"Resource '{resource_id}' not found")
try:
result = await upstream_api.operate(resource_id)
return result
except RateLimitError:
raise_tool_error(ErrorCode.RATE_LIMITED,
"Rate limit hit. Wait 60 seconds before retrying.")
except UpstreamServiceError as e:
raise_tool_error(ErrorCode.UPSTREAM_ERROR,
f"Upstream service unavailable: {str(e)}")Testing MCP Servers
# Unit tests using FastMCP test client
import pytest
from fastmcp.testing import MCPTestClient
@pytest.fixture
def client():
return MCPTestClient(mcp)
@pytest.mark.asyncio
async def test_lookup_customer_success(client):
result = await client.call_tool("lookup_customer", {
"customer_id": "cust_test123",
"include_history": False,
})
assert result["id"] == "cust_test123"
assert "email" in result
@pytest.mark.asyncio
async def test_lookup_customer_invalid_id(client):
with pytest.raises(ToolError) as exc_info:
await client.call_tool("lookup_customer", {"customer_id": "invalid"})
assert "Invalid customer ID format" in str(exc_info.value)
@pytest.mark.asyncio
async def test_list_tools(client):
tools = await client.list_tools()
tool_names = [t.name for t in tools]
assert "lookup_customer" in tool_names
assert "search_knowledge_base" in tool_names
# Integration test with MCP Inspector CLI:
# npx @modelcontextprotocol/inspector python server.pySecurity Considerations
# 1. Input validation — always sanitize before passing to external systems
import re
def validate_customer_id(customer_id: str) -> str:
"""Strict allowlist validation — reject anything that doesn't match."""
if not re.match(r'^cust_[a-zA-Z0-9]{8,32}$', customer_id):
raise ToolError("Invalid customer ID format")
return customer_id
# 2. Scope limiting — tools should only access what they need
# BAD: Generic database tool with raw SQL access
@mcp.tool()
async def run_sql(query: str) -> list: # DANGEROUS — full DB access
return await db.execute(query)
# GOOD: Specific, scoped read-only queries
@mcp.tool()
async def get_customer_orders(customer_id: str, limit: int = 10) -> list:
"""Returns up to 20 orders for a customer. Read-only."""
return await db.query(
"SELECT id, status, total FROM orders WHERE customer_id = $1 LIMIT $2",
[validate_customer_id(customer_id), min(limit, 20)], # Hard cap on limit
)
# 3. Rate limiting per tool
from slowapi import Limiter
limiter = Limiter(key_func=get_client_id)
@mcp.tool()
@limiter.limit("10/minute")
async def expensive_operation(params: dict) -> dict:
...
# 4. Secrets never in tool output
@mcp.tool()
async def get_config() -> dict:
config = await config_store.get_all()
# Redact secrets before returning
return {k: "***REDACTED***" if "secret" in k.lower() or "key" in k.lower()
else v for k, v in config.items()}Anti-Patterns
❌ Monolithic tools that do everything
A tool named manage_customer that handles lookup, update, and deletion is hard for models to use correctly. Split into get_customer, update_customer, delete_customer.
❌ Poor tool descriptions
The description IS the model's documentation. Vague descriptions lead to incorrect tool use. Include: what it does, when to use it, what parameters mean, what it returns.
❌ Letting exceptions propagate raw
Python exceptions become opaque error messages. Always catch and convert to ToolError with human-readable context.
❌ No input validation
Models make mistakes. Always validate inputs server-side — don't trust that the model will pass correct types or formats.
❌ Mutable side effects without confirmation
For destructive operations (delete, send email, charge card), consider requiring an explicit confirm=True parameter so models can't accidentally execute them.
Quick Reference
Tool description template:
"[Action verb] [what it operates on]. Returns [what].
Use when [trigger condition].
[Important constraints or caveats]."
Transport selection:
Local (Claude Desktop, CLI) → stdio
Remote (HTTP clients) → SSE or streamable-http
High-throughput production → streamable-http + load balancer
Error types:
ToolError → Expected failure (not found, validation failed, permission denied)
Exception → Unexpected failure (bug) — FastMCP converts to internal error
Validation checklist:
☐ Allowlist validation on IDs (regex or enum)
☐ Length limits on string inputs
☐ Range limits on numeric inputs
☐ Secrets redacted from output
☐ Rate limiting on expensive/dangerous tools
☐ Authorization check before any data access