Intermediate
Building Your First MCP Server
Building Your First MCP Server
This lesson guides you through building a complete MCP server using the Python SDK. You’ll create a server that exposes resources and tools, integrate it with Claude Desktop, and test the full integration.
Python MCP SDK Overview
The official MCP SDK handles protocol details, allowing you to focus on implementing your server’s logic:
pip install mcpServer Initialization
from mcp.server import Server
from mcp.types import (
Resource, Tool, TextContent, ToolResult
)
import json
# Create server instance
server = Server("my-data-server")
# Add metadata
server.info.name = "My Data Server"
server.info.version = "1.0.0"Building a Complete Database Server
Let’s build an MCP server that exposes a database with resources and tools:
from mcp.server import Server
from mcp.types import (
Resource, Tool, TextContent, ToolResult,
ListResourcesResult, ListToolsResult,
CallToolResult
)
from dataclasses import dataclass
import json
import sqlite3
from typing import List, Dict, Any
@dataclass
class DatabaseServer:
"""MCP server exposing a SQLite database."""
def __init__(self, db_path: str):
self.server = Server("database-server")
self.db_path = db_path
self.setup_handlers()
def setup_handlers(self):
"""Register request handlers."""
@self.server.list_resources()
async def list_resources() -> ListResourcesResult:
"""List available database resources."""
return ListResourcesResult(
resources=[
Resource(
uri="sqlite:///schema",
name="Database Schema",
description="Database tables and columns",
mimeType="text/plain"
),
Resource(
uri="sqlite:///tables",
name="Tables",
description="Available tables in database",
mimeType="application/json"
),
Resource(
uri="sqlite:///stats",
name="Database Statistics",
description="Row counts and sizes per table",
mimeType="application/json"
)
]
)
@self.server.read_resource()
async def read_resource(uri: str) -> str:
"""Read resource content."""
if uri == "sqlite:///schema":
return self._get_schema()
elif uri == "sqlite:///tables":
return json.dumps(self._get_tables())
elif uri == "sqlite:///stats":
return json.dumps(self._get_stats())
else:
raise ValueError(f"Unknown resource: {uri}")
@self.server.list_tools()
async def list_tools() -> ListToolsResult:
"""List available tools."""
return ListToolsResult(
tools=[
Tool(
name="query",
description="Execute a SQL SELECT query",
inputSchema={
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "SQL SELECT query"
},
"limit": {
"type": "integer",
"description": "Max rows to return",
"default": 100
}
},
"required": ["sql"]
}
),
Tool(
name="insert",
description="Insert rows into a table",
inputSchema={
"type": "object",
"properties": {
"table": {
"type": "string",
"description": "Table name"
},
"values": {
"type": "object",
"description": "Column values"
}
},
"required": ["table", "values"]
}
),
Tool(
name="count_rows",
description="Count rows in a table",
inputSchema={
"type": "object",
"properties": {
"table": {
"type": "string",
"description": "Table name"
}
},
"required": ["table"]
}
)
]
)
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict) -> CallToolResult:
"""Execute a tool."""
try:
if name == "query":
result = self._execute_query(
arguments["sql"],
arguments.get("limit", 100)
)
return CallToolResult(
content=[TextContent(type="text", text=result)],
isError=False
)
elif name == "insert":
result = self._insert_rows(
arguments["table"],
arguments["values"]
)
return CallToolResult(
content=[TextContent(type="text", text=result)],
isError=False
)
elif name == "count_rows":
result = self._count_rows(arguments["table"])
return CallToolResult(
content=[TextContent(type="text", text=result)],
isError=False
)
else:
return CallToolResult(
content=[TextContent(type="text", text="Unknown tool")],
isError=True
)
except Exception as e:
return CallToolResult(
content=[TextContent(type="text", text=str(e))],
isError=True
)
def _get_schema(self) -> str:
"""Get database schema."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
schema = {}
for (table_name,) in tables:
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
schema[table_name] = [
{"name": col[1], "type": col[2]} for col in columns
]
conn.close()
return json.dumps(schema, indent=2)
def _get_tables(self) -> List[str]:
"""Get list of tables."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
conn.close()
return tables
def _get_stats(self) -> Dict[str, int]:
"""Get database statistics."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
stats = {}
for (table_name,) in tables:
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
count = cursor.fetchone()[0]
stats[table_name] = count
conn.close()
return stats
def _execute_query(self, sql: str, limit: int) -> str:
"""Execute a SELECT query."""
# Security: In production, validate/sanitize SQL
if not sql.strip().upper().startswith("SELECT"):
raise ValueError("Only SELECT queries allowed")
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
cursor.execute(f"{sql} LIMIT {limit}")
rows = cursor.fetchall()
results = [dict(row) for row in rows]
return json.dumps(results, indent=2)
finally:
conn.close()
def _insert_rows(self, table: str, values: Dict[str, Any]) -> str:
"""Insert a row into a table."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
try:
columns = ", ".join(values.keys())
placeholders = ", ".join(["?" for _ in values])
sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
cursor.execute(sql, list(values.values()))
conn.commit()
return f"Inserted row with {len(values)} columns"
finally:
conn.close()
def _count_rows(self, table: str) -> str:
"""Count rows in a table."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
try:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
count = cursor.fetchone()[0]
return f"Table '{table}' has {count} rows"
finally:
conn.close()
async def run(self):
"""Run the server."""
async with self.server:
print("Database MCP server running...")Integrating with Claude Desktop
To use your MCP server with Claude Desktop, create a configuration file:
{
"mcpServers": {
"database": {
"command": "python",
"args": ["/path/to/database_server.py"],
"disabled": false
}
}
}Location depends on your OS:
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json - Linux:
~/.config/Claude/claude_desktop_config.json
Testing Your MCP Server
import asyncio
from mcp.client import Client
from mcp.client.stdio import StdioClientTransport
async def test_server():
"""Test the MCP server."""
# Create client with stdio transport
transport = StdioClientTransport(
command="python",
args=["/path/to/database_server.py"]
)
client = Client()
async with client:
# Initialize connection
await client.initialize()
# List resources
resources = await client.list_resources()
print("Available resources:")
for resource in resources.resources:
print(f" - {resource.name}: {resource.uri}")
# List tools
tools = await client.list_tools()
print("\nAvailable tools:")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description}")
# Read a resource
schema = await client.read_resource("sqlite:///schema")
print(f"\nSchema: {schema}")
# Call a tool
result = await client.call_tool("count_rows", {"table": "users"})
print(f"\nQuery result: {result.content[0].text}")
# Run test
asyncio.run(test_server())Error Handling
Proper error handling ensures robustness:
class RobustMCPServer:
"""MCP server with comprehensive error handling."""
def __init__(self):
self.server = Server("robust-server")
self.setup_handlers()
def setup_handlers(self):
"""Setup handlers with error handling."""
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict) -> CallToolResult:
"""Execute tool with error handling."""
try:
# Input validation
if not self._validate_tool_input(name, arguments):
return CallToolResult(
content=[TextContent(
type="text",
text="Invalid arguments"
)],
isError=True
)
# Resource limits
if not self._check_resources():
return CallToolResult(
content=[TextContent(
type="text",
text="Server resource limit exceeded"
)],
isError=True
)
# Execute tool
result = await self._execute_tool(name, arguments)
return CallToolResult(
content=[TextContent(type="text", text=result)],
isError=False
)
except TimeoutError:
return CallToolResult(
content=[TextContent(type="text", text="Tool execution timeout")],
isError=True
)
except Exception as e:
return CallToolResult(
content=[TextContent(type="text", text=f"Error: {str(e)}")],
isError=True
)
def _validate_tool_input(self, name: str, arguments: Dict) -> bool:
"""Validate tool input."""
# Implement validation logic
return True
def _check_resources(self) -> bool:
"""Check system resources."""
# Implement resource checking
return True
async def _execute_tool(self, name: str, arguments: Dict) -> str:
"""Execute the actual tool."""
# Implement tool execution
return ""Best Practices for Production Servers
1. Input Validation and Sanitization
def validate_sql_query(sql: str) -> bool:
"""Validate SQL queries for safety."""
dangerous_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
sql_upper = sql.upper()
for keyword in dangerous_keywords:
if keyword in sql_upper:
return False
return True2. Logging and Monitoring
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def logged_tool_call(name: str, arguments: Dict):
"""Log all tool calls for debugging."""
logger.info(f"Tool called: {name} with {arguments}")3. Rate Limiting
import time
from collections import defaultdict
class RateLimiter:
"""Simple rate limiter for tool calls."""
def __init__(self, calls_per_minute: int = 60):
self.calls_per_minute = calls_per_minute
self.call_history = defaultdict(list)
def is_allowed(self, client_id: str) -> bool:
"""Check if tool call is allowed."""
now = time.time()
cutoff = now - 60 # Last minute
# Clean old calls
self.call_history[client_id] = [
t for t in self.call_history[client_id] if t > cutoff
]
if len(self.call_history[client_id]) >= self.calls_per_minute:
return False
self.call_history[client_id].append(now)
return TrueKey Takeaway
Building MCP servers with the Python SDK abstracts protocol complexity, letting you focus on your server’s core logic. Proper error handling, validation, and testing ensure your servers are robust and production-ready.
Exercises
-
Database Server: Create a complete MCP server that exposes a SQLite database with schema resources and query tools.
-
File Server: Build an MCP server that lists files from a directory and provides read/write tools with proper access control.
-
API Server: Create a wrapper server that exposes a third-party REST API as MCP resources and tools.
-
Testing Suite: Write comprehensive tests for your server’s tools and resource handlers.
-
Integration: Configure Claude Desktop to use your server and test interactions through the Claude UI.
-
Error Scenarios: Test how your server handles invalid inputs, timeouts, and resource exhaustion.