Building your first MCP server transforms from theoretical understanding to practical implementation. This comprehensive guide walks you through creating production-ready MCP servers using Python’s FastMCP framework, the most popular and well-supported implementation for Python developers. By the end of this tutorial, you will have built, tested, and deployed a functional MCP server that exposes custom tools and resources to AI applications.
FastMCP 2.0 represents the evolution of MCP server development in Python. Originally incorporated into the official MCP SDK as FastMCP 1.0, version 2.0 extends far beyond basic protocol implementation to provide enterprise-grade features including authentication, deployment tools, testing frameworks, and comprehensive client libraries. This makes FastMCP the framework of choice for developers building production systems.
This guide covers environment setup, implementing tools and resources, handling errors gracefully, testing your server, and deploying to production. The code examples demonstrate real-world patterns used by teams at companies like Microsoft, AWS, and GitHub to build their official MCP servers.
Environment Setup and Installation
Setting up a proper Python environment for MCP development requires specific tools and dependencies. While you can use traditional virtual environments, the recommended approach uses uv, a modern Python package manager that provides faster installations and better dependency resolution.
First, install uv if you do not already have it. On macOS and Linux, use the installation script. On Windows, download the installer from the Astral website or use pip to install uv globally.
# macOS/Linux installation
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"
# Alternative: Install via pip
pip install uvCreate a new project directory and initialize a Python environment. The uv tool creates isolated environments automatically, similar to conda or virtualenv but with superior performance.
# Create project directory
mkdir my-mcp-server
cd my-mcp-server
# Initialize Python environment with uv
uv venv
# Activate the environment
# On macOS/Linux:
source .venv/bin/activate
# On Windows:
.venv\Scripts\activate
# Install FastMCP
uv pip install fastmcp
# Install additional dependencies for this tutorial
uv pip install httpx # For HTTP requests
uv pip install python-dotenv # For environment variablesVerify your installation by checking the FastMCP version and importing the module. This confirms that all dependencies installed correctly and your environment is configured properly.
# Check FastMCP version
python -c "import fastmcp; print(fastmcp.__version__)"
# Expected output: 2.x.xBuilding Your First MCP Server
The simplest MCP server exposes basic functionality through tools. This example creates a server that provides mathematical operations, demonstrating FastMCP’s core concepts without unnecessary complexity.
Create a file named server.py with the following code. This server exposes two tools: one for addition and one for calculating powers. FastMCP automatically generates JSON schemas from Python type hints and docstrings, eliminating boilerplate code.
from fastmcp import FastMCP
# Initialize the MCP server
mcp = FastMCP("MathServer")
@mcp.tool()
def add(a: float, b: float) -> float:
"""
Add two numbers together.
Args:
a: First number
b: Second number
Returns:
The sum of a and b
"""
return a + b
@mcp.tool()
def power(base: float, exponent: float) -> float:
"""
Calculate base raised to exponent.
Args:
base: The base number
exponent: The exponent
Returns:
base ** exponent
"""
return base ** exponent
if __name__ == "__main__":
# Run the server
mcp.run()This code demonstrates FastMCP’s decorator-based API. The @mcp.tool() decorator transforms a regular Python function into an MCP tool. FastMCP handles capability negotiation, request parsing, response formatting, and error handling automatically.
Run your server using the Python interpreter or uv. The server starts on localhost and waits for connections from MCP clients.
# Run the server
python server.py
# Or with uv
uv run server.py
# Server output:
# FastMCP server 'MathServer' running on stdio transport
# Available tools: add, powerTo test your server, use the MCP Inspector, a debugging tool provided by Anthropic. Open a new terminal window and start the inspector while your server runs.
# Install MCP Inspector globally
npx @modelcontextprotocol/inspector python server.py
# The inspector opens a web interface at http://localhost:5173The inspector displays your server’s available tools and allows you to invoke them with test parameters. This provides immediate feedback during development without connecting to a full AI application.
Implementing Resources for Data Access
Resources represent read-only data that AI models can access to gather context. Unlike tools which perform actions, resources expose information for retrieval. This section demonstrates both static and dynamic resources.
Create a new file named data_server.py that exposes configuration data and user profiles. This example shows how to structure resources with proper URI schemes and parameter handling.
from fastmcp import FastMCP
import json
from typing import Dict, Any
mcp = FastMCP("DataServer")
# In-memory data store
USER_DATA = {
"1": {"id": "1", "name": "Alice Johnson", "role": "Engineer", "department": "Product"},
"2": {"id": "2", "name": "Bob Smith", "role": "Manager", "department": "Sales"},
"3": {"id": "3", "name": "Carol White", "role": "Designer", "department": "Product"}
}
@mcp.resource("config://app/settings")
def app_settings() -> str:
"""
Application configuration settings.
Returns:
JSON string containing application configuration
"""
config = {
"version": "1.0.0",
"environment": "production",
"features": {
"analytics": True,
"beta_features": False
},
"limits": {
"max_requests_per_minute": 1000,
"max_concurrent_connections": 100
}
}
return json.dumps(config, indent=2)
@mcp.resource("user://{user_id}/profile")
def user_profile(user_id: str) -> Dict[str, Any]:
"""
Get user profile information by ID.
Args:
user_id: The unique identifier for the user
Returns:
Dictionary containing user profile data
Raises:
KeyError: If user_id does not exist
"""
if user_id not in USER_DATA:
raise KeyError(f"User {user_id} not found")
return USER_DATA[user_id]
@mcp.resource("user://list")
def list_users() -> Dict[str, Any]:
"""
List all available users.
Returns:
Dictionary containing all user profiles
"""
return {
"total": len(USER_DATA),
"users": list(USER_DATA.values())
}
if __name__ == "__main__":
mcp.run()This server demonstrates three types of resources. The config resource provides static application settings that rarely change. The user profile resource accepts parameters through URI templates, enabling dynamic data retrieval. The user list resource aggregates data from multiple sources.
Resource URIs follow a consistent pattern. The scheme (config:// or user://) identifies the resource type. The path specifies the resource location. Parameters appear in curly braces and map to function arguments. This design makes resources discoverable and predictable for AI clients.
graph TB
subgraph "Resource Types"
S[Static Resource]
D[Dynamic Resource]
A[Aggregate Resource]
end
subgraph "Static Example"
S --> S1["config://app/settings"]
S1 --> S2[Fixed Configuration]
end
subgraph "Dynamic Example"
D --> D1["user://{user_id}/profile"]
D1 --> D2[Parameter Extraction]
D2 --> D3[Data Lookup]
end
subgraph "Aggregate Example"
A --> A1["user://list"]
A1 --> A2[Collect All Users]
A2 --> A3[Format Response]
endBuilding a Real-World Database Integration Server
Production MCP servers typically integrate with external systems like databases, APIs, or file systems. This example builds a server that connects to a SQLite database, demonstrating patterns applicable to any data source.
First, create a sample database for testing. This script initializes a SQLite database with sample customer and order data.
import sqlite3
# Create sample database
conn = sqlite3.connect('customers.db')
cursor = conn.cursor()
# Create tables
cursor.execute('''
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER,
product TEXT NOT NULL,
amount DECIMAL(10, 2),
status TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (customer_id) REFERENCES customers(id)
)
''')
# Insert sample data
customers = [
(1, 'Alice Johnson', 'alice@example.com'),
(2, 'Bob Smith', 'bob@example.com'),
(3, 'Carol White', 'carol@example.com')
]
orders = [
(1, 1, 'Laptop', 1299.99, 'delivered'),
(2, 1, 'Mouse', 29.99, 'delivered'),
(3, 2, 'Keyboard', 89.99, 'pending'),
(4, 3, 'Monitor', 449.99, 'shipped')
]
cursor.executemany('INSERT OR REPLACE INTO customers VALUES (?, ?, ?, CURRENT_TIMESTAMP)', customers)
cursor.executemany('INSERT OR REPLACE INTO orders VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)', orders)
conn.commit()
conn.close()
print("Database created successfully")Now create the MCP server that exposes database operations through tools and resources. This server demonstrates proper connection management, parameter validation, and error handling.
from fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any, Optional
from contextlib import contextmanager
mcp = FastMCP("DatabaseServer")
# Database configuration
DB_PATH = "customers.db"
@contextmanager
def get_db_connection():
"""
Context manager for database connections.
Ensures proper cleanup even if errors occur.
"""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row # Return rows as dictionaries
try:
yield conn
finally:
conn.close()
@mcp.resource("db://schema")
def database_schema() -> Dict[str, Any]:
"""
Get the database schema information.
Returns:
Dictionary containing table definitions and column information
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# Get all tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
schema = {}
for table in tables:
cursor.execute(f"PRAGMA table_info({table})")
columns = [
{
"name": row[1],
"type": row[2],
"not_null": bool(row[3]),
"primary_key": bool(row[5])
}
for row in cursor.fetchall()
]
schema[table] = columns
return schema
@mcp.tool()
def get_customer(customer_id: int) -> Dict[str, Any]:
"""
Retrieve customer information by ID.
Args:
customer_id: The customer's unique identifier
Returns:
Dictionary containing customer data
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM customers WHERE id = ?",
(customer_id,)
)
row = cursor.fetchone()
if row is None:
raise ValueError(f"Customer {customer_id} not found")
return dict(row)
@mcp.tool()
def search_customers(name_query: str) -> List[Dict[str, Any]]:
"""
Search for customers by name (case-insensitive partial match).
Args:
name_query: Search string to match against customer names
Returns:
List of matching customer records
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM customers WHERE name LIKE ? ORDER BY name",
(f"%{name_query}%",)
)
return [dict(row) for row in cursor.fetchall()]
@mcp.tool()
def get_customer_orders(customer_id: int, status: Optional[str] = None) -> List[Dict[str, Any]]:
"""
Get all orders for a specific customer with optional status filter.
Args:
customer_id: The customer's unique identifier
status: Optional status filter (delivered, pending, shipped)
Returns:
List of order records for the customer
"""
with get_db_connection() as conn:
cursor = conn.cursor()
if status:
cursor.execute(
"SELECT * FROM orders WHERE customer_id = ? AND status = ? ORDER BY created_at DESC",
(customer_id, status)
)
else:
cursor.execute(
"SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC",
(customer_id,)
)
return [dict(row) for row in cursor.fetchall()]
@mcp.tool()
def create_order(
customer_id: int,
product: str,
amount: float,
status: str = "pending"
) -> Dict[str, Any]:
"""
Create a new order for a customer.
Args:
customer_id: The customer's unique identifier
product: Product name or description
amount: Order amount in dollars
status: Order status (default: pending)
Returns:
Dictionary containing the created order data
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# Verify customer exists
cursor.execute("SELECT id FROM customers WHERE id = ?", (customer_id,))
if cursor.fetchone() is None:
raise ValueError(f"Customer {customer_id} does not exist")
# Insert order
cursor.execute(
"INSERT INTO orders (customer_id, product, amount, status) VALUES (?, ?, ?, ?)",
(customer_id, product, amount, status)
)
conn.commit()
# Return the created order
order_id = cursor.lastrowid
cursor.execute("SELECT * FROM orders WHERE id = ?", (order_id,))
return dict(cursor.fetchone())
if __name__ == "__main__":
mcp.run()This server implements several production patterns. The context manager ensures database connections close properly even when errors occur. Parameter validation prevents invalid operations. The row factory configuration returns results as dictionaries for easier JSON serialization. Error handling provides clear messages when operations fail.
Error Handling and Production Patterns
Robust error handling separates development prototypes from production systems. MCP servers must distinguish between expected errors that clients can handle and unexpected errors that indicate server problems.
FastMCP provides the ToolError exception for client-visible errors. These represent expected failure cases that the AI should understand and potentially recover from. For example, attempting to access a non-existent user is an expected error that should be communicated clearly.
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from typing import Dict, Any
import logging
mcp = FastMCP("ErrorHandlingServer")
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@mcp.tool()
def divide(a: float, b: float) -> float:
"""
Divide two numbers with proper error handling.
Args:
a: Numerator
b: Denominator
Returns:
Result of a / b
Raises:
ToolError: If division by zero is attempted
"""
if b == 0:
# This is an expected error - communicate it clearly to the client
raise ToolError(
"Cannot divide by zero. Please provide a non-zero denominator.",
code="DIVISION_BY_ZERO"
)
return a / b
@mcp.tool()
def process_data(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Process data with comprehensive error handling.
Args:
data: Dictionary containing data to process
Returns:
Processed data dictionary
"""
try:
# Validate required fields
required_fields = ['id', 'value']
missing_fields = [f for f in required_fields if f not in data]
if missing_fields:
raise ToolError(
f"Missing required fields: {', '.join(missing_fields)}",
code="MISSING_FIELDS"
)
# Validate data types
if not isinstance(data['id'], (int, str)):
raise ToolError(
"Field 'id' must be a string or integer",
code="INVALID_TYPE"
)
if not isinstance(data['value'], (int, float)):
raise ToolError(
"Field 'value' must be a number",
code="INVALID_TYPE"
)
# Process the data
result = {
"id": data['id'],
"value": data['value'] * 2,
"processed": True
}
logger.info(f"Successfully processed data for ID {data['id']}")
return result
except ToolError:
# Re-raise ToolErrors as-is
raise
except Exception as e:
# Catch unexpected errors and log them
logger.error(f"Unexpected error processing data: {str(e)}", exc_info=True)
# Return a generic error to the client without exposing internal details
raise ToolError(
"An unexpected error occurred while processing your request",
code="INTERNAL_ERROR"
)
if __name__ == "__main__":
mcp.run()This pattern provides several benefits. Expected errors use ToolError with descriptive messages and error codes. Unexpected errors get logged with full stack traces for debugging while presenting generic messages to clients. This prevents information leakage while maintaining debuggability.
Testing Your MCP Server
FastMCP includes a built-in client for testing servers without connecting to AI applications. This enables unit testing, integration testing, and debugging during development.
Create a test file named test_server.py that exercises your server’s functionality. This example demonstrates testing patterns for tools, resources, and error conditions.
import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
async def test_math_server():
"""
Test the MathServer tools.
"""
print("Testing MathServer...")
# Create client (assumes server running on http://localhost:8000/mcp)
async with Client("http://localhost:8000/mcp") as client:
# Test add tool
result = await client.call_tool("add", arguments={"a": 5, "b": 3})
print(f"add(5, 3) = {result}")
assert result == 8, f"Expected 8, got {result}"
# Test power tool
result = await client.call_tool("power", arguments={"base": 2, "exponent": 10})
print(f"power(2, 10) = {result}")
assert result == 1024, f"Expected 1024, got {result}"
print("MathServer tests passed!")
async def test_database_server():
"""
Test the DatabaseServer tools and resources.
"""
print("\nTesting DatabaseServer...")
async with Client("http://localhost:8001/mcp") as client:
# Test schema resource
schema = await client.read_resource("db://schema")
print(f"Database schema: {schema}")
assert "customers" in schema
assert "orders" in schema
# Test get_customer tool
customer = await client.call_tool("get_customer", arguments={"customer_id": 1})
print(f"Customer 1: {customer}")
assert customer['name'] == "Alice Johnson"
# Test search_customers tool
results = await client.call_tool("search_customers", arguments={"name_query": "John"})
print(f"Search results for 'John': {results}")
assert len(results) > 0
# Test get_customer_orders tool
orders = await client.call_tool("get_customer_orders", arguments={"customer_id": 1})
print(f"Orders for customer 1: {orders}")
assert len(orders) > 0
# Test create_order tool
new_order = await client.call_tool("create_order", arguments={
"customer_id": 1,
"product": "Test Product",
"amount": 99.99,
"status": "pending"
})
print(f"Created order: {new_order}")
assert new_order['product'] == "Test Product"
print("DatabaseServer tests passed!")
async def test_error_handling():
"""
Test error handling in servers.
"""
print("\nTesting error handling...")
async with Client("http://localhost:8002/mcp") as client:
# Test division by zero
try:
result = await client.call_tool("divide", arguments={"a": 10, "b": 0})
print("ERROR: Should have raised exception for division by zero")
except Exception as e:
print(f"Correctly caught error: {str(e)}")
# Test missing required fields
try:
result = await client.call_tool("process_data", arguments={"data": {}})
print("ERROR: Should have raised exception for missing fields")
except Exception as e:
print(f"Correctly caught error: {str(e)}")
print("Error handling tests passed!")
async def run_all_tests():
"""
Run all test suites.
"""
await test_math_server()
await test_database_server()
await test_error_handling()
print("\n✅ All tests passed successfully!")
if __name__ == "__main__":
asyncio.run(run_all_tests())To run these tests, start your servers in separate terminal windows with HTTP transport enabled, then execute the test script. The tests verify that tools execute correctly, resources return expected data, and error conditions are handled appropriately.
Integrating with Claude Desktop
The ultimate goal is connecting your server to AI applications like Claude Desktop. This requires adding configuration that tells Claude how to launch and communicate with your server.
On macOS, the configuration file is located at ~/Library/Application Support/Claude/claude_desktop_config.json. On Windows, use %APPDATA%\Claude\claude_desktop_config.json. Create or edit this file to add your server.
{
"mcpServers": {
"math-server": {
"command": "python",
"args": [
"-m",
"fastmcp",
"run",
"/absolute/path/to/your/server.py"
]
},
"database-server": {
"command": "python",
"args": [
"-m",
"fastmcp",
"run",
"/absolute/path/to/your/database_server.py"
],
"env": {
"DB_PATH": "/absolute/path/to/customers.db"
}
}
}
}After adding this configuration, restart Claude Desktop. Your servers will be available for the AI to use. You can verify the connection by asking Claude to use your tools. For example, “Can you add 15 and 27 using the math server?” or “Show me all customers in the database.”
For production deployments, use HTTP transport instead of stdio. This allows running servers as independent services that multiple clients can connect to.
# Run server with HTTP transport
python server.py --transport sse --port 8000
# Configuration for Claude Desktop with HTTP
{
"mcpServers": {
"production-server": {
"type": "http",
"url": "http://localhost:8000/mcp",
"headers": {
"Authorization": "Bearer your-api-key-here"
}
}
}
}graph LR
subgraph "Development Flow"
C[Create Server]
T[Write Tests]
D[Debug with Inspector]
I[Integrate with Claude]
end
C --> T
T --> D
D --> T
D --> I
subgraph "Testing Tools"
D --> D1[MCP Inspector]
D --> D2[FastMCP Client]
D --> D3[Unit Tests]
end
subgraph "Deployment"
I --> I1[Local Stdio]
I --> I2[HTTP Server]
I --> I3[Cloud Platform]
endAdvanced Features and Best Practices
Production MCP servers benefit from several advanced patterns that improve reliability, security, and maintainability.
Implement connection pooling for database servers to avoid creating new connections for each request. Use context managers to ensure resources clean up properly. Add caching for expensive operations that return the same data repeatedly.
from fastmcp import FastMCP
import sqlite3
from functools import lru_cache
import time
mcp = FastMCP("OptimizedDatabaseServer")
# Connection pool
class ConnectionPool:
def __init__(self, db_path, pool_size=5):
self.db_path = db_path
self.pool = []
for _ in range(pool_size):
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.row_factory = sqlite3.Row
self.pool.append(conn)
def get_connection(self):
if self.pool:
return self.pool.pop()
return sqlite3.connect(self.db_path)
def return_connection(self, conn):
if len(self.pool) < 5:
self.pool.append(conn)
else:
conn.close()
pool = ConnectionPool("customers.db")
@lru_cache(maxsize=100)
def cached_schema():
"""Cache database schema for 1 hour."""
conn = pool.get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
schema = {}
for table in tables:
cursor.execute(f"PRAGMA table_info({table})")
schema[table] = [dict(row) for row in cursor.fetchall()]
return schema
finally:
pool.return_connection(conn)
@mcp.resource("db://schema")
def get_schema():
"""Get database schema with caching."""
return cached_schema()
# Clear cache periodically
import threading
def clear_cache_periodically():
while True:
time.sleep(3600) # Clear every hour
cached_schema.cache_clear()
threading.Thread(target=clear_cache_periodically, daemon=True).start()
if __name__ == "__main__":
mcp.run()Add request logging and monitoring to track server usage and identify problems. Use structured logging with JSON output for easy parsing by monitoring systems.
import logging
import json
from datetime import datetime
# Configure structured logging
class StructuredLogger:
def __init__(self, name):
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
def log_request(self, tool_name, arguments, duration_ms, success=True, error=None):
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"tool": tool_name,
"arguments": arguments,
"duration_ms": duration_ms,
"success": success
}
if error:
log_entry["error"] = str(error)
self.logger.info(json.dumps(log_entry))
logger = StructuredLogger("mcp-server")
# Use in tools
@mcp.tool()
def monitored_operation(param: str) -> str:
start_time = time.time()
try:
# Perform operation
result = f"Processed: {param}"
duration_ms = (time.time() - start_time) * 1000
logger.log_request("monitored_operation", {"param": param}, duration_ms)
return result
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
logger.log_request("monitored_operation", {"param": param}, duration_ms, success=False, error=e)
raiseConclusion and Next Steps
You have now built functional MCP servers using Python and FastMCP. The skills demonstrated in this guide apply to any MCP server implementation, whether you are exposing databases, APIs, file systems, or custom business logic.
Key takeaways include understanding FastMCP's decorator-based API for defining tools and resources, implementing proper error handling with ToolError for expected failures, using context managers for resource cleanup, testing servers with the FastMCP client and MCP Inspector, and integrating servers with Claude Desktop through configuration files.
The next article in this series explores implementing MCP servers in Node.js and C#, demonstrating how the same concepts apply across programming languages. You will see how FastAPI-style patterns in Python translate to Express.js patterns in Node.js and ASP.NET Core patterns in C#. This cross-platform perspective helps you choose the right implementation language for your requirements.
For immediate next steps, explore the FastMCP documentation for advanced features like authentication providers, middleware support, and server composition. Review the official MCP servers repository on GitHub to see production implementations from companies like Microsoft, AWS, and GitHub. Join the MCP community on Discord to connect with other developers building MCP servers.
References
- FastMCP Official Documentation
- FastMCP GitHub Repository
- MCP Python SDK (Official)
- DataCamp - Building an MCP Server and Client with FastMCP 2.0
- Firecrawl - How to Build MCP Servers in Python: Complete FastMCP Tutorial
- MCPcat - Build MCP Servers in Python with FastMCP
- Awesome MCP Servers - Community Collection
- Official MCP Servers Repository
- MCP Example Servers Documentation
- UV Python Package Manager
