Skip to main content

Building MCP Servers

FastMCP is a high-level Python framework for building MCP servers. It handles protocol details, transport setup, and message formatting so you can focus on your tool logic.

Installation

bash
uv add fastmcp
# or
pip install fastmcp

Your First MCP Server

python
from fastmcp import FastMCP

# Create the server
mcp = FastMCP("my-tools-server")

@mcp.tool()
def add_numbers(a: int, b: int) -> int:
"""Add two numbers together.

Args:
a: First number
b: Second number

Returns:
The sum of a and b
"""
return a + b

@mcp.tool()
def search_docs(query: str, max_results: int = 5) -> list[dict]:
"""Search project documentation for relevant pages.

Args:
query: Search query string
max_results: Maximum number of results to return (default 5)

Returns:
List of matching documents with title and excerpt
"""
# Simulated search — replace with real search logic
docs = [
{"title": "Getting Started", "content": "Welcome to the project..."},
{"title": "API Reference", "content": "The API supports REST endpoints..."},
{"title": "Configuration", "content": "Set up your environment variables..."},
]
results = [d for d in docs if query.lower() in d["content"].lower() or query.lower() in d["title"].lower()]
return results[:max_results]

if __name__ == "__main__":
mcp.run()

Run it:

bash
fastmcp run server.py

Adding Resources

Resources are data sources that clients can read. Unlike tools (which the LLM invokes), resources are application-controlled — the user or host decides when to read them.

python
import json
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("project-assistant")

PROJECT_ROOT = Path("/home/user/my-project")

@mcp.resource(f"file://{PROJECT_ROOT}/README.md")
def read_readme() -> str:
"""Read the project README."""
return (PROJECT_ROOT / "README.md").read_text()

@mcp.resource("config://project/settings")
def get_project_settings() -> str:
"""Get current project settings as JSON."""
settings = {
"name": "my-project",
"version": "1.0.0",
"python": "3.12",
"framework": "FastAPI",
}
return json.dumps(settings, indent=2)

@mcp.resource("db://project/stats")
def get_project_stats() -> str:
"""Get project statistics."""
stats = {
"total_files": len(list(PROJECT_ROOT.rglob("*"))),
"python_files": len(list(PROJECT_ROOT.rglob("*.py"))),
"test_files": len(list(PROJECT_ROOT.rglob("test_*.py"))),
}
return json.dumps(stats, indent=2)

Adding Prompts

Prompts are reusable templates that users can select from the client.

python
@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
"""Generate a code review prompt."""
return (
"Please review the following " + language + " code:\n"
+ code
+ "\nFocus on: correctness, performance, security, and best practices."
)

@mcp.prompt()
def explain_error(error_message: str, language: str = "python") -> str:
"""Generate a prompt to explain an error message."""
return (
"Explain the following " + language + " error message:\n"
+ error_message
+ "\nProvide: meaning, common cause, fix steps, and prevention tips."
)

Real-World Example: Database MCP Server

python
import sqlite3
import json
from fastmcp import FastMCP
from contextlib import contextmanager

mcp = FastMCP("sqlite-assistant")

DB_PATH = "/data/app.db"

@contextmanager
def get_db():
"""Get a database connection."""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()

@mcp.tool()
def list_tables() -> list[str]:
"""List all tables in the database.

Returns:
List of table names
"""
with get_db() as conn:
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
return [row["name"] for row in cursor.fetchall()]

@mcp.tool()
def describe_table(table_name: str) -> str:
"""Get the schema of a database table.

Args:
table_name: Name of the table to describe

Returns:
Column information for the table
"""
with get_db() as conn:
cursor = conn.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
if not columns:
return f"Table '{table_name}' not found."
result = []
for col in columns:
result.append(
f" {col['name']} ({col['type']})"
f"{' PRIMARY KEY' if col['pk'] else ''}"
f"{' NOT NULL' if col['notnull'] else ''}"
)
return f"Table: {table_name}\n" + "\n".join(result)

@mcp.tool()
def execute_query(sql: str) -> str:
"""Execute a SELECT query on the database.

Args:
sql: SQL query to execute. Only SELECT statements are allowed.

Returns:
Query results as formatted text
"""
if not sql.strip().upper().startswith("SELECT"):
return "Error: Only SELECT queries are allowed for safety."

with get_db() as conn:
try:
cursor = conn.execute(sql)
rows = cursor.fetchall()
if not rows:
return "No results found."
columns = [desc[0] for desc in cursor.description]
result = [columns]
for row in rows:
result.append(list(row))
return json.dumps(result, indent=2, default=str)
except sqlite3.Error as e:
return f"SQL Error: {e}"

@mcp.resource("db://schema")
def get_full_schema() -> str:
"""Get the complete database schema."""
with get_db() as conn:
cursor = conn.execute(
"SELECT sql FROM sqlite_master WHERE type='table' AND sql IS NOT NULL"
)
return "\n\n".join(row["sql"] for row in cursor.fetchall())

if __name__ == "__main__":
mcp.run()

Deploying Your MCP Server

Local (stdio) — For Development

bash
# Run directly
fastmcp run server.py

# Or via Claude Desktop config
# Add to claude_desktop_config.json:
json
{
"mcpServers": {
"my-tools": {
"command": "uv",
"args": ["run", "fastmcp", "run", "server.py"]
}
}
}

Remote (SSE) — For Production

bash
# Run as SSE server on port 8000
fastmcp run server.py --transport sse --port 8000
bash
# Deploy to Cloud Run
gcloud run deploy mcp-server \
--source . \
--region us-central1 \
--allow-unauthenticated \
--set-env-vars "PORT=8080"
FastMCP Inspector

Use fastmcp dev server.py to open the MCP Inspector — a web UI for testing your tools, resources, and prompts interactively before connecting a real client.

Testing MCP Servers

python
import pytest
from fastmcp import Client

@pytest.mark.asyncio
async def test_add_numbers():
async with Client(mcp) as client:
result = await client.call_tool("add_numbers", {"a": 3, "b": 5})
assert result[0].text == "8"

@pytest.mark.asyncio
async def test_list_tables():
async with Client(mcp) as client:
result = await client.call_tool("list_tables", {})
tables = result[0].text
assert "users" in tables

@pytest.mark.asyncio
async def test_read_resource():
async with Client(mcp) as client:
content = await client.read_resource("config://project/settings")
assert "my-project" in content