Ready-to-use patterns for common MCP server implementations. Each recipe includes a description, use case, and implementation sketch.
Expose SQL query capability to Claude with parameterized, read-only access.
@app.call_tool()
async def handle_tool(name, arguments):
if name == "query_database":
query = arguments["query"]
params = arguments.get("params", [])
# Only allow SELECT statements
if not query.strip().upper().startswith("SELECT"):
return [{"type": "text", "text": "Error: Only SELECT queries are permitted"}]
async with pool.acquire() as conn:
rows = await conn.fetch(query, *params)
return [{"type": "text", "text": json.dumps([dict(r) for r in rows])}]
Always restrict database MCP servers to read-only access unless you have explicit need for writes and have implemented approval gates.
Expose a scoped directory to Claude for reading and optionally writing files.
import os
from pathlib import Path
BASE_DIR = Path("/workspace/project").resolve()
def safe_path(relative_path):
"""Ensure path stays within BASE_DIR"""
full_path = (BASE_DIR / relative_path).resolve()
if not str(full_path).startswith(str(BASE_DIR)):
raise ValueError(f"Path traversal detected: {relative_path}")
return full_path
@app.call_tool()
async def handle_tool(name, arguments):
if name == "read_file":
path = safe_path(arguments["path"])
content = path.read_text()
return [{"type": "text", "text": content}]
Wrap an existing REST API with typed, documented MCP tools so Claude can call it without knowing the raw API details.
import httpx
async def call_api(endpoint, method="GET", data=None, headers=None):
async with httpx.AsyncClient(base_url=BASE_URL) as client:
headers = {"Authorization": f"Bearer {API_TOKEN}", **(headers or {})}
response = await getattr(client, method.lower())(endpoint, json=data, headers=headers)
response.raise_for_status()
return response.json()
@app.call_tool()
async def handle_tool(name, arguments):
if name == "get_user":
user = await call_api(f"/users/{arguments['user_id']}")
# Return only safe fields — never return full internal objects
return [{"type": "text", "text": json.dumps({
"id": user["id"],
"name": user["name"],
"email": user["email"]
})}]
Expose a semantic search tool backed by a vector store.
from anthropic import Anthropic
client = Anthropic()
async def get_embedding(text):
# Use your embedding model here
# This example uses a hypothetical embedding function
return embed(text)
@app.call_tool()
async def handle_tool(name, arguments):
if name == "search_knowledge_base":
query = arguments["query"]
limit = arguments.get("limit", 5)
embedding = await get_embedding(query)
results = vector_store.search(embedding, limit=limit)
formatted = []
for doc in results:
formatted.append(f"## {doc.title}\n{doc.content[:500]}...")
return [{"type": "text", "text": "\n\n".join(formatted)}]
Handle large datasets with cursor-based pagination so Claude can navigate without context overflow.
@app.call_tool()
async def handle_tool(name, arguments):
if name == "list_records":
page_size = min(arguments.get("page_size", 10), 50) # cap at 50
cursor = arguments.get("cursor") # None for first page
records, next_cursor = await db.list_records(
cursor=cursor, limit=page_size
)
return [{
"type": "text",
"text": json.dumps({
"records": records,
"next_cursor": next_cursor,
"has_more": next_cursor is not None
})
}]
Expose a series of related operations that Claude calls in sequence to complete a workflow.
def error_response(code, message, retryable=False):
return [{
"type": "text",
"text": json.dumps({
"error": True,
"code": code,
"message": message,
"retryable": retryable
})
}]
# Usage
if not record:
return error_response("NOT_FOUND", f"Record {record_id} does not exist", retryable=False)
if rate_limited:
return error_response("RATE_LIMITED", "Try again in 30 seconds", retryable=True)