Introduction
What if you could give Claude or any AI assistant the ability to check your Docker containers, search through your personal notes, or plan your day based on real-time weather data—all through natural language? That's exactly what the Model Context Protocol (MCP) makes possible, and you're about to learn how to build it yourself.
The Model Context Protocol is Anthropic's open standard for connecting AI assistants to external tools and data sources. Think of it as a universal adapter that lets AI applications safely interact with your local environment, APIs, and databases. But here's the challenge: most tutorials either oversimplify the concept or throw you into complex production code without proper context.
This tutorial takes a different approach. We'll build three progressively complex MCP servers using real code from the python-mcp-servers repository, each demonstrating a distinct architectural pattern you'll encounter in production systems. You'll start with a simple Docker management assistant, progress to a semantic search engine for your notes, and finish with an intelligent day planner that orchestrates multiple APIs.
By the end of this tutorial, you'll understand:
- How MCP servers work under the hood and why they're revolutionary for AI development
- Three distinct architectural patterns for building MCP servers (API wrapper, system resource manager, and data store connector)
- How to implement tools, resources, and prompts using the official Python SDK
- Production-ready patterns for error handling, async operations, and external API integration
- How to configure and test your servers with Claude Desktop
Whether you're a Python developer looking to extend AI capabilities or an AI engineer wanting to integrate custom data sources, this hands-on guide will give you the practical knowledge to start building production-quality MCP servers today. Let's dive in.
Table of Contents
- Introduction
- Understanding MCP: The Foundation for AI-Tool Integration
- Server 1: Docker Dev Assistant - Your First MCP Server
- Server 2: Personal Knowledge Base - Semantic Search and AI Integration
- Server 3: Smart Day Planner - Production Patterns and API Integration
- Bringing It All Together: From Learning to Production
- Conclusion
Understanding MCP: The Foundation for AI-Tool Integration
Before we write a single line of code, let's understand what makes MCP servers powerful and how they fit into the AI ecosystem. The Model Context Protocol defines a standardized way for AI applications to discover and interact with external capabilities, solving a fundamental problem: how do we safely extend AI assistants beyond their training data?
The Three Pillars of MCP
Every MCP server exposes capabilities through three core primitives, each serving a distinct purpose in the AI interaction model:
Resources are data sources that the AI can read. Think of them as file-like entities that provide context—your markdown notes, database records, or API responses. When Claude asks to "read the documentation," it's accessing a resource. Resources are passive: the AI reads them but doesn't modify them.
Tools are executable functions that perform actions. These are the "verbs" of your MCP server—searching databases, creating files, sending emails, or querying APIs. Unlike resources, tools are active: they change state, trigger operations, and return results. When you ask Claude to "check my Docker containers," it's calling a tool.
Prompts are templated interactions that guide the AI through complex workflows. They're like macros that combine multiple tools and resources into reusable patterns. For example, a "debug-container" prompt might automatically gather logs, check resource usage, and suggest solutions—all in one go.
How MCP Communication Works
MCP servers communicate using JSON-RPC 2.0, a lightweight remote procedure call protocol that's both simple and powerful. Here's what happens when Claude interacts with your server:
- Initialization: Claude (the client) connects to your server via stdio (standard input/output) or HTTP with Server-Sent Events
- Capability Discovery: The server announces what tools, resources, and prompts it provides
- Request-Response: Claude sends JSON-RPC requests, your server processes them and returns JSON-RPC responses
- Lifecycle Management: The connection persists, allowing stateful interactions until explicitly closed
This architecture is elegant because the AI assistant never directly accesses your systems—it only communicates with your MCP server through a standardized protocol. Your server acts as a security boundary, validating inputs and controlling access.
Why MCP Matters for Developers
Traditional AI integrations require custom APIs, complex authentication flows, and tight coupling between the AI application and your data sources. MCP changes this by providing a single integration point. Write one MCP server, and it works with any MCP-compatible client—Claude Desktop, custom applications, or future AI tools.
The python-mcp-servers repository demonstrates this versatility through three distinct patterns. Each server solves a real problem developers face: managing local development environments, organizing personal knowledge, and coordinating daily tasks. These aren't toy examples—they're production-quality implementations you can deploy today.
Here's the critical insight: MCP servers are not just about connecting data to AI. They're about creating reusable, composable building blocks that extend AI capabilities in predictable, secure ways. As we build our three servers, you'll see how this architecture enables everything from simple command execution to complex multi-step workflows.
Server 1: Docker Dev Assistant - Your First MCP Server
Let's start with the simplest example from the repository: the Docker Dev Assistant. This server gives Claude the ability to manage Docker containers through natural language, demonstrating the fundamental structure of an MCP server in just 401 lines of code.
What You'll Build
The Docker Dev Assistant exposes three core tools:
- docker_ps: Lists all running containers with their status
- docker_logs: Retrieves logs from specific containers
- docker_stats: Shows real-time resource usage statistics
It also includes a debug-container prompt that guides Claude through a systematic troubleshooting workflow. This combination of tools and prompts shows how MCP servers can provide both raw capabilities and intelligent workflows.
Server Architecture and Setup
The server uses the official MCP Python SDK to handle the protocol details, letting us focus on implementing our tools. Here's the foundational structure from server.py:
from mcp.server import Server
from mcp.types import Tool, TextContent, Prompt, PromptMessage
import mcp.server.stdio
import subprocess
import logging
## Initialize the server with a name
server = Server("docker-dev-assistant")
## Configure logging for debugging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("docker-dev-assistant")
This initialization does three critical things: creates a server instance with a unique name, sets up logging for debugging, and imports the necessary MCP components. The server name is important—it's how Claude Desktop identifies and displays your server in its interface.
Implementing Your First Tool
Let's examine how the docker_ps tool is implemented. This tool demonstrates the complete pattern you'll use for every tool you create:
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available Docker management tools."""
return [
Tool(
name="docker_ps",
description="List all Docker containers with their status",
inputSchema={
"type": "object",
"properties": {
"all": {
"type": "boolean",
"description": "Show all containers (default shows just running)",
"default": False
}
}
}
)
# ... other tools
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute Docker commands based on tool requests."""
if name == "docker_ps":
cmd = ["docker", "ps"]
if arguments.get("all", False):
cmd.append("-a")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True
)
return [TextContent(type="text", text=result.stdout)]
except subprocess.CalledProcessError as e:
logger.error(f"Docker command failed: {e.stderr}")
return [TextContent(
type="text",
text=f"Error executing docker ps: {e.stderr}"
)]
Here's what's happening step by step:
- Tool Registration: The
@server.list_tools()decorator tells MCP this function defines available tools. It returns a list ofToolobjects, each with a name, description, and input schema. - Input Schema: The
inputSchemafollows JSON Schema format, defining what parameters the tool accepts. This is crucial—it tells Claude what information it needs to provide when calling the tool. - Tool Execution: The
@server.call_tool()decorator handles actual tool invocation. It receives the tool name and arguments, executes the operation, and returns results asTextContent. - Error Handling: The try-except block catches subprocess errors and returns meaningful error messages. Never let exceptions crash your server—always return descriptive error text that helps Claude understand what went wrong.
Adding the Debug Workflow Prompt
The repository's Docker server includes a sophisticated prompt that demonstrates how to guide AI through multi-step workflows. Here's the implementation from server.py:
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
"""List available debugging prompts."""
return [
Prompt(
name="debug-container",
description="Debug a Docker container with systematic checks",
arguments=[
{
"name": "container_name",
"description": "Name or ID of the container to debug",
"required": True
}
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
"""Generate debugging workflow for containers."""
if name == "debug-container":
container = arguments["container_name"]
return [
PromptMessage(
role="user",
content=TextContent(
type="text",
text=f"""Debug the Docker container '{container}' by:
1. Check if it's running with docker_ps
2. If running, get recent logs with docker_logs
3. Check resource usage with docker_stats
4. Analyze the output and suggest solutions
"""
)
)
]
This prompt transforms a simple request like "debug my app container" into a systematic investigation. Prompts are powerful because they encode domain expertise—in this case, the best practices for debugging Docker containers.
Running and Testing the Server
To run the Docker Dev Assistant, you need to configure Claude Desktop to connect to it. The setup guide provides detailed instructions, but here's the essential configuration for claude_desktop_config.json:
{
"mcpServers": {
"docker-dev-assistant": {
"command": "/path/to/python-mcp-servers/venv/bin/python3",
"args": ["/path/to/python-mcp-servers/docker-dev-assistant/server.py"]
}
}
}
Critical setup details:
- Use the virtual environment's Python (not system Python) to ensure the mcp package is available
- Provide absolute paths—relative paths won't work
- The server name in the config must match the name you used when initializing the
Serverobject
After restarting Claude Desktop, you can test with natural language: "What Docker containers are running?" Claude will automatically discover and call the docker_ps tool, parse the output, and present it in a readable format.
Key Takeaways from the Docker Server
This first server teaches you the essential MCP patterns:
- Server initialization and logging setup
- Tool registration with clear input schemas
- Subprocess execution with proper error handling
- Prompt creation for guided workflows
- Configuration and testing with Claude Desktop
The Docker Dev Assistant is simple by design—it wraps command-line tools and exposes them through MCP. But this pattern is incredibly powerful. You can apply the same structure to wrap any command-line utility, from git to kubectl to custom scripts. The 401 lines of code in this server are a template you'll use repeatedly.
Server 2: Personal Knowledge Base - Semantic Search and AI Integration
Now let's level up with the Personal Knowledge Base server. This 784-line implementation demonstrates how to build an MCP server that actively processes and indexes data, using vector embeddings to enable semantic search across your markdown notes. This is where MCP starts to feel like magic.
The Problem This Server Solves
You have hundreds of markdown files scattered across directories—meeting notes, project documentation, research findings, random ideas. Traditional search only finds exact keyword matches, but what you really want is semantic search: finding notes based on meaning, not just words. You want to ask "What did I learn about API design?" and get relevant notes even if they never use that exact phrase.
The Personal Knowledge Base server solves this by:
- Automatically discovering and indexing all markdown files in a directory
- Generating vector embeddings using local AI models (no API keys needed)
- Providing semantic search that understands context and meaning
- Exposing notes as MCP resources that Claude can read directly
- Offering hybrid search combining semantic and keyword approaches
Architecture: From Files to Embeddings
The server uses ChromaDB for vector storage and sentence-transformers for generating embeddings. Here's the initialization from server.py:
import chromadb
from sentence_transformers import SentenceTransformer
from pathlib import Path
import hashlib
## Initialize embedding model (runs locally, no API needed)
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
## Initialize ChromaDB for vector storage
chroma_client = chromadb.Client()
collection = chroma_client.create_collection(
name="notes",
metadata={"hnsw:space": "cosine"} # Use cosine similarity
)
## Directory to watch for markdown files
NOTES_DIR = Path.home() / "notes"
Why these choices matter:
- all-MiniLM-L6-v2 is a lightweight model (80MB) that generates quality embeddings without GPU requirements
- ChromaDB provides fast vector search with minimal setup—no separate database server needed
- Cosine similarity measures the angle between vectors, perfect for semantic similarity
Indexing Notes: The Core Functionality
The server needs to convert markdown files into searchable vectors. Here's the indexing implementation:
def index_note(file_path: Path) -> dict:
"""Index a single markdown file into the vector database."""
try:
# Read the file content
content = file_path.read_text(encoding='utf-8')
# Generate a unique ID based on file path
file_id = hashlib.md5(str(file_path).encode()).hexdigest()
# Create embedding from content
embedding = embedding_model.encode(content).tolist()
# Store in ChromaDB with metadata
collection.upsert(
ids=[file_id],
embeddings=[embedding],
documents=[content],
metadatas=[{
"path": str(file_path),
"filename": file_path.name,
"modified": file_path.stat().st_mtime
}]
)
logger.info(f"Indexed: {file_path.name}")
return {"success": True, "file": file_path.name}
except Exception as e:
logger.error(f"Failed to index {file_path}: {e}")
return {"success": False, "error": str(e)}
Key implementation details:
- Content hashing: Using MD5 of the file path creates stable IDs for upserts (update or insert)
- Embedding generation: The sentence-transformer converts text into a 384-dimensional vector
- Metadata storage: Storing file path and modification time enables freshness checks
- Error isolation: If one file fails to index, others continue processing
Implementing Semantic Search
The semantic search tool is where vector embeddings shine. Here's the implementation from the repository:
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute knowledge base operations."""
if name == "semantic_search":
query = arguments["query"]
limit = arguments.get("limit", 5)
try:
# Generate embedding for the search query
query_embedding = embedding_model.encode(query).tolist()
# Search the vector database
results = collection.query(
query_embeddings=[query_embedding],
n_results=limit,
include=["documents", "metadatas", "distances"]
)
# Format results for Claude
formatted_results = []
for i, (doc, meta, dist) in enumerate(zip(
results['documents'][0],
results['metadatas'][0],
results['distances'][0]
)):
similarity = 1 - dist # Convert distance to similarity
formatted_results.append(
f"**{i+1}. {meta['filename']}** (similarity: {similarity:.2%})\n"
f"Path: {meta['path']}\n"
f"Content preview: {doc[:200]}...\n"
)
return [TextContent(
type="text",
text="\n".join(formatted_results)
)]
except Exception as e:
logger.error(f"Semantic search failed: {e}")
return [TextContent(
type="text",
text=f"Search error: {str(e)}"
)]
What makes this powerful:
- The query "API design principles" will match notes about "REST best practices" or "microservice patterns" even without exact keyword overlap
- Similarity scores help Claude understand result confidence
- Content previews give context without overwhelming the AI with full documents
- Error handling ensures graceful degradation if the search fails
Dynamic Resources: Exposing Notes to Claude
One of the most elegant features of this server is how it exposes notes as MCP resources. Unlike static resource lists, this implementation dynamically discovers all markdown files:
@server.list_resources()
async def list_resources() -> list[Resource]:
"""List all markdown files as resources."""
resources = []
if NOTES_DIR.exists():
for md_file in NOTES_DIR.rglob("*.md"):
# Create a resource URI for each file
uri = f"note:///{md_file.relative_to(NOTES_DIR)}"
resources.append(
Resource(
uri=uri,
name=md_file.name,
description=f"Markdown note: {md_file.stem}",
mimeType="text/markdown"
)
)
return resources
@server.read_resource()
async def read_resource(uri: str) -> str:
"""Read a specific note by URI."""
if uri.startswith("note:///"):
# Extract file path from URI
relative_path = uri.replace("note:///", "")
file_path = NOTES_DIR / relative_path
if file_path.exists() and file_path.suffix == ".md":
return file_path.read_text(encoding='utf-8')
else:
raise ValueError(f"Note not found: {uri}")
raise ValueError(f"Invalid URI scheme: {uri}")
Why this matters: When you add a new markdown file to your notes directory, it automatically becomes available to Claude as a resource. No configuration needed, no manual registration—the server adapts to your file system dynamically.
Hybrid Search: Combining Semantic and Keyword Approaches
The repository includes a sophisticated search_notes tool that combines semantic search with traditional keyword matching:
if name == "search_notes":
query = arguments["query"]
search_type = arguments.get("type", "hybrid") # semantic, keyword, or hybrid
if search_type in ["semantic", "hybrid"]:
# Perform semantic search
semantic_results = perform_semantic_search(query, limit=10)
if search_type in ["keyword", "hybrid"]:
# Perform keyword search
keyword_results = perform_keyword_search(query, limit=10)
if search_type == "hybrid":
# Merge and rank results using both signals
merged_results = merge_search_results(
semantic_results,
keyword_results,
weights={"semantic": 0.7, "keyword": 0.3}
)
return format_results(merged_results)
The hybrid approach is powerful because:
- Semantic search finds conceptually similar content
- Keyword search ensures exact matches aren't missed
- Weighted merging balances both approaches
- Users can override the default to prefer one method
Prompts for Knowledge Synthesis
The Personal Knowledge Base includes two prompts that showcase how to guide Claude through complex analytical tasks. Here's the "connect-ideas" prompt from server.py:
@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
if name == "connect-ideas":
topic = arguments["topic"]
return [
PromptMessage(
role="user",
content=TextContent(
type="text",
text=f"""Explore connections around '{topic}' by:
1. Use semantic_search to find related notes
2. Use find_similar on the most relevant results
3. Identify common themes and unexpected connections
4. Synthesize insights across multiple notes
5. Suggest areas for deeper exploration
"""
)
)
]
This prompt transforms a simple query into a knowledge discovery workflow. It's not just finding information—it's analyzing relationships, identifying patterns, and generating new insights. This is where MCP servers become true AI assistants, not just search tools.
Key Learnings from the Knowledge Base Server
This second server introduces significantly more complexity:
- File system monitoring and dynamic resource discovery
- Vector embeddings for semantic understanding
- Database integration with ChromaDB for persistence
- Hybrid search patterns combining multiple approaches
- Analytical prompts that guide multi-step reasoning
The 784 lines of code demonstrate production patterns you'll use when building MCP servers that process and understand data, not just execute commands. The jump from 401 to 784 lines isn't just more code—it's a different architectural pattern that handles state, persistence, and AI-powered analysis.
Server 3: Smart Day Planner - Production Patterns and API Integration
The final example, the Smart Day Planner, demonstrates production-ready patterns for building MCP servers that integrate with external APIs. This 760-line server orchestrates weather data, todo lists, and calendar events to provide intelligent day planning assistance. This is where you learn the patterns you'll use in real-world applications.
The Production Challenge
Unlike the previous examples that worked with local resources, the Smart Day Planner must:
- Authenticate with external APIs (OpenWeatherMap, Todoist) using API keys
- Handle rate limits and implement retry logic with exponential backoff
- Cache responses to minimize API calls and reduce costs
- Operate asynchronously to handle concurrent requests efficiently
- Manage configuration through environment variables and YAML files
- Fail gracefully when APIs are unavailable or return errors
These challenges mirror what you'll face building production MCP servers. Let's see how the repository solves each one.
Async Server Architecture
The Smart Day Planner uses an async MCP server to handle concurrent operations efficiently. Here's the initialization from server.py:
import asyncio
import aiohttp
from mcp.server import Server
from mcp.server.stdio import stdio_server
import os
from dotenv import load_dotenv
import yaml
from datetime import datetime, timedelta
import logging
## Load environment variables from .env file
load_dotenv()
## Load configuration from YAML
with open("config.yaml", "r") as f:
config = yaml.safe_load(f)
## Initialize async server
server = Server("smart-day-planner")
## API credentials from environment
WEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")
TODOIST_API_KEY = os.getenv("TODOIST_API_KEY")
## Initialize async HTTP session (reused across requests)
http_session = None
async def get_http_session() -> aiohttp.ClientSession:
"""Get or create HTTP session for API calls."""
global http_session
if http_session is None:
http_session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=30),
headers={"User-Agent": "MCP-Smart-Day-Planner/1.0"}
)
return http_session
Key architectural decisions:
- load_dotenv() reads API keys from
.envfile, keeping secrets out of code - YAML configuration separates settings from code for easier deployment
- Global HTTP session reuses connections for better performance
- Async/await enables concurrent API calls without blocking
Secrets Management and Configuration
The repository includes a comprehensive approach to managing secrets and configuration. The .env.example file shows the required environment variables:
## OpenWeatherMap API (get free key at openweathermap.org)
OPENWEATHER_API_KEY=your_api_key_here
## Todoist API (get from todoist.com/app/settings/integrations)
TODOIST_API_KEY=your_api_key_here
## Optional: Default location for weather
DEFAULT_CITY=San Francisco
DEFAULT_COUNTRY=US
The config.yaml handles non-sensitive settings:
weather:
cache_duration_minutes: 30
units: metric # or imperial
todoist:
default_project: "Inbox"
priority_threshold: 3
planner:
weather_weight: 0.3
urgency_weight: 0.5
energy_weight: 0.2
This separation is crucial for production:
- Secrets in
.envcan be injected by deployment systems (Kubernetes secrets, AWS Parameter Store, etc.) - Configuration in YAML can be version-controlled and environment-specific
- Never commit
.envfiles—the.gitignoreprevents this
Implementing External API Calls with Retry Logic
The weather tool demonstrates production-quality API integration with proper error handling:
async def fetch_weather(city: str, country: str = "US") -> dict:
"""Fetch weather data with retry logic and caching."""
cache_key = f"{city}_{country}"
# Check cache first
if cache_key in weather_cache:
cached_data, cached_time = weather_cache[cache_key]
cache_age = datetime.now() - cached_time
if cache_age < timedelta(minutes=config['weather']['cache_duration_minutes']):
logger.info(f"Using cached weather for {city}")
return cached_data
# Prepare API request
url = "https://api.openweathermap.org/data/2.5/weather"
params = {
"q": f"{city},{country}",
"appid": WEATHER_API_KEY,
"units": config['weather']['units']
}
# Retry logic with exponential backoff
max_retries = 3
for attempt in range(max_retries):
try:
session = await get_http_session()
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
# Cache the result
weather_cache[cache_key] = (data, datetime.now())
return data
elif response.status == 429:
# Rate limited - wait and retry
wait_time = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
logger.warning(f"Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
else:
error_text = await response.text()
raise ValueError(f"API error {response.status}: {error_text}")
except aiohttp.ClientError as e:
if attempt == max_retries - 1:
raise # Last attempt failed
wait_time = 2 ** attempt
logger.warning(f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s")
await asyncio.sleep(wait_time)
raise Exception(f"Failed to fetch weather after {max_retries} attempts")
This implementation demonstrates several production patterns:
- Response Caching: Reduces API calls by storing results for 30 minutes (configurable)
- Exponential Backoff: Waits 1s, 2s, 4s between retries to avoid overwhelming the API
- Rate Limit Handling: Specifically handles 429 status codes with appropriate delays
- Async Operations: Uses
aiohttpfor non-blocking HTTP requests - Comprehensive Error Handling: Distinguishes between network errors, API errors, and rate limits
Tool Orchestration: The analyze_week Function
The most sophisticated tool in the Smart Day Planner is analyze_week, which orchestrates multiple API calls and synthesizes data:
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "analyze_week":
city = arguments.get("city", os.getenv("DEFAULT_CITY"))
country = arguments.get("country", os.getenv("DEFAULT_COUNTRY"))
try:
# Fetch data concurrently using asyncio.gather
weather_task = fetch_weather_forecast(city, country)
todos_task = fetch_todoist_tasks()
# Wait for both to complete
forecast, todos = await asyncio.gather(
weather_task,
todos_task,
return_exceptions=True # Don't fail if one API is down
)
# Handle partial failures gracefully
if isinstance(forecast, Exception):
logger.error(f"Weather fetch failed: {forecast}")
forecast = None
if isinstance(todos, Exception):
logger.error(f"Todos fetch failed: {todos}")
todos = None
# Analyze and generate recommendations
analysis = generate_weekly_analysis(
forecast=forecast,
todos=todos,
config=config['planner']
)
return [TextContent(type="text", text=analysis)]
except Exception as e:
logger.error(f"Week analysis failed: {e}")
return [TextContent(
type="text",
text=f"Analysis error: {str(e)}\nSome data may be unavailable."
)]
Key orchestration patterns:
- Concurrent API Calls:
asyncio.gather()fetches weather and todos simultaneously, cutting response time in half - Partial Failure Handling:
return_exceptions=Trueallows the function to continue even if one API fails - Graceful Degradation: The analysis proceeds with whatever data is available
- Weighted Scoring: The
config['planner']weights determine how weather, urgency, and energy levels influence recommendations
The suggest_activities Tool: Context-Aware Recommendations
The suggest_activities tool demonstrates how to combine multiple data sources for intelligent recommendations:
async def suggest_activities(weather_data: dict, todos: list, time_of_day: str) -> str:
"""Generate activity suggestions based on weather, todos, and time."""
# Extract weather conditions
temp = weather_data['main']['temp']
conditions = weather_data['weather'][0]['main']
# Filter todos by priority and deadline
urgent_todos = [
t for t in todos
if t.get('priority', 1) >= config['todoist']['priority_threshold']
]
# Time-based energy levels
energy_levels = {
"morning": "high",
"afternoon": "medium",
"evening": "low"
}
energy = energy_levels.get(time_of_day, "medium")
# Generate suggestions
suggestions = []
# Weather-appropriate outdoor activities
if conditions == "Clear" and 15 <= temp <= 25:
if energy == "high" and len(urgent_todos) < 3:
suggestions.append("Perfect weather for a morning run or bike ride")
# Indoor focus time based on todos
if len(urgent_todos) > 0:
suggestions.append(
f"Focus time recommended: {len(urgent_todos)} high-priority tasks pending"
)
# Energy-appropriate activities
if energy == "low" and conditions == "Rain":
suggestions.append("Good evening for light reading or planning tomorrow")
return "\n".join(suggestions)
This function showcases:
- Multi-source data fusion: Combining weather, tasks, and time of day
- Configurable thresholds: Using YAML config for priority levels
- Context-aware logic: Different suggestions based on energy levels and conditions
- Personalization potential: Easy to extend with user preferences
Error Handling and Logging Best Practices
The Smart Day Planner implements comprehensive error handling throughout. Here's a pattern used consistently:
try:
result = await risky_operation()
logger.info(f"Operation succeeded: {result}")
return format_success_response(result)
except aiohttp.ClientError as e:
# Network-level errors
logger.error(f"Network error: {e}")
return format_error_response("Network unavailable. Please check connectivity.")
except ValueError as e:
# API returned invalid data
logger.error(f"Invalid API response: {e}")
return format_error_response("Received unexpected data from API.")
except Exception as e:
# Catch-all for unexpected errors
logger.exception(f"Unexpected error: {e}") # Logs full stack trace
return format_error_response("An unexpected error occurred.")
Why this matters:
- Specific exception handling provides targeted error messages
- Logging at appropriate levels (info, warning, error, exception) aids debugging
- User-friendly error messages don't expose internal details
- logger.exception() captures full stack traces for debugging without crashing
Running the Production Server
The Smart Day Planner requires additional setup steps. From the README:
## 1. Copy environment template
cp .env.example .env
## 2. Edit .env with your API keys
## Get OpenWeatherMap key: https://openweathermap.org/api
## Get Todoist key: https://todoist.com/app/settings/integrations
## 3. Install dependencies
pip install -r requirements.txt
## 4. Test the server
python server.py
The Claude Desktop configuration is similar to previous examples but points to this server:
{
"mcpServers": {
"smart-day-planner": {
"command": "/path/to/venv/bin/python3",
"args": ["/path/to/smart-day-planner/server.py"],
"env": {
"OPENWEATHER_API_KEY": "your_key_here",
"TODOIST_API_KEY": "your_key_here"
}
}
}
}
Note: You can provide environment variables directly in the config or rely on the .env file. For production deployments, prefer environment variables injected by your deployment system.
Advanced Patterns from the Smart Day Planner
This third server introduces production-grade patterns essential for real-world MCP servers:
- Async Architecture: Non-blocking operations for better performance
- External API Integration: Proper authentication, retry logic, and rate limiting
- Response Caching: Reduces costs and improves responsiveness
- Configuration Management: Separates code from settings and secrets
- Concurrent Operations: Uses
asyncio.gather()for parallel API calls - Partial Failure Handling: Continues operation when some services are unavailable
- Comprehensive Logging: Enables debugging and monitoring in production
- Graceful Error Handling: Provides useful feedback without exposing internals
The 760 lines of code represent a template for production MCP servers that integrate with external services. Whether you're building a server for Slack, GitHub, Salesforce, or internal APIs, these patterns apply.
Bringing It All Together: From Learning to Production
You've now explored three complete MCP server implementations, each demonstrating progressively sophisticated patterns. Let's consolidate what you've learned and understand how to apply these concepts to your own projects.
The Three Architectural Patterns
Each server in the python-mcp-servers repository represents a distinct architectural pattern you'll encounter when building MCP servers:
Pattern 1: Command Wrapper (Docker Dev Assistant)
- Wraps existing command-line tools or system utilities
- Simple synchronous operations with subprocess execution
- Minimal state management
- Use when: You need to expose CLI tools, scripts, or system commands to AI assistants
Pattern 2: Data Processor (Personal Knowledge Base)
- Processes and indexes local or remote data
- Maintains state (vector database, file monitoring)
- Provides both raw access (resources) and processed insights (semantic search)
- Use when: You need to make unstructured data AI-searchable or analyze documents
Pattern 3: API Orchestrator (Smart Day Planner)
- Integrates multiple external services
- Handles async operations, caching, and retries
- Synthesizes data from various sources
- Use when: Building production services that coordinate multiple APIs or data sources
The progression is intentional: Start simple with command wrappers, add data processing when needed, and implement full orchestration for complex integrations. Most production MCP servers combine elements from all three patterns.
Common Patterns Across All Servers
Despite their different purposes, all three servers share fundamental patterns that form the MCP server blueprint:
## 1. Initialize server with unique name
server = Server("your-server-name")
## 2. Set up logging for debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("your-server-name")
## 3. Define available tools
@server.list_tools()
async def list_tools() -> list[Tool]:
return [Tool(...), Tool(...)]
## 4. Implement tool execution
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "tool_name":
# Execute operation
result = await your_operation(arguments)
return [TextContent(type="text", text=result)]
## 5. Expose resources (optional)
@server.list_resources()
async def list_resources() -> list[Resource]:
return [Resource(...)]
## 6. Define prompts (optional)
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
return [Prompt(...)]
## 7. Run the server
if __name__ == "__main__":
mcp.server.stdio.run_stdio_server(server)
This structure is your starting point for any MCP server. The complexity comes from what happens inside the tool implementations, not from the MCP protocol itself.
Testing and Debugging Your Servers
The repository's testing guide provides comprehensive debugging strategies. Here are the essential techniques:
1. Test the server directly:
## Run the server and check for startup errors
python server.py
## Expected output:
## INFO:your-server-name:Starting MCP Server
2. Use the MCP Inspector:
The MCP Inspector is a debugging tool that lets you interact with your server without Claude Desktop:
## Install the inspector
npm install -g @modelcontextprotocol/inspector
## Run it with your server
mcp-inspector python server.py
3. Check Claude Desktop logs:
- macOS:
~/Library/Logs/Claude/mcp*.log - Windows:
%APPDATA%\Claude\logs\mcp*.log - Linux:
~/.config/Claude/logs/mcp*.log
4. Add debug logging:
## In your tool implementations
logger.debug(f"Tool called: {name} with args: {arguments}")
logger.info(f"Processing {len(items)} items")
logger.warning(f"Unusual condition: {condition}")
logger.error(f"Operation failed: {error}")
5. Test with simple queries:
Start with basic tool calls before trying complex workflows:
## Test each tool individually
"List Docker containers"
"Search my notes for 'API design'"
"What's the weather in San Francisco?"
## Then test prompts
"Debug my app container"
"Connect ideas about machine learning"
"Analyze my week"
Extending the Examples for Your Use Cases
The three servers provide templates you can adapt for countless applications. Here are practical extension ideas:
Extending the Docker Dev Assistant:
- Add
docker_exectool to run commands inside containers - Implement
docker_composeoperations for multi-container apps - Create a "deployment" prompt that builds, tests, and deploys
- Add resource monitoring with Prometheus integration
Extending the Personal Knowledge Base:
- Connect to Notion, Obsidian, or Roam Research APIs
- Add image and PDF indexing using OCR
- Implement graph-based note linking for concept maps
- Create automatic note summarization and tagging
Extending the Smart Day Planner:
- Integrate Google Calendar or Outlook for scheduling
- Add habit tracking with Habitica or Streaks
- Connect to fitness APIs (Strava, Fitbit) for activity suggestions
- Implement location-based recommendations using GPS data
The key to successful extensions is maintaining the patterns demonstrated in the repository: clear tool definitions, robust error handling, and thoughtful prompt design.
Security Considerations for Production Deployments
When deploying MCP servers in production environments, security becomes critical. Here are essential considerations:
1. Input Validation:
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "read_file":
path = arguments.get("path", "")
# Validate path to prevent directory traversal
if ".." in path or path.startswith("/"):
return [TextContent(
type="text",
text="Invalid path: directory traversal not allowed"
)]
# Restrict to allowed directories
allowed_base = Path("/safe/directory")
full_path = (allowed_base / path).resolve()
if not str(full_path).startswith(str(allowed_base)):
return [TextContent(
type="text",
text="Access denied: path outside allowed directory"
)]
2. API Key Management:
- Never hardcode API keys in source code
- Use environment variables or secret management services
- Rotate keys regularly
- Use separate keys for development and production
3. Rate Limiting:
from collections import defaultdict
from datetime import datetime, timedelta
request_counts = defaultdict(list)
def check_rate_limit(user_id: str, max_requests: int = 100, window_minutes: int = 60) -> bool:
"""Simple rate limiting by user."""
now = datetime.now()
cutoff = now - timedelta(minutes=window_minutes)
# Remove old requests
request_counts[user_id] = [
req_time for req_time in request_counts[user_id]
if req_time > cutoff
]
# Check limit
if len(request_counts[user_id]) >= max_requests:
return False
# Record this request
request_counts[user_id].append(now)
return True
4. Sandboxing Tool Execution:
For tools that execute code or commands, use sandboxing:
import subprocess
def safe_execute(command: list[str], timeout: int = 30) -> str:
"""Execute command with safety constraints."""
# Whitelist allowed commands
allowed_commands = {"docker", "git", "ls"}
if command[0] not in allowed_commands:
raise ValueError(f"Command not allowed: {command[0]}")
# Execute with timeout and resource limits
result = subprocess.run(
command,
capture_output=True,
text=True,
timeout=timeout,
check=True,
env={"PATH": "/usr/bin:/bin"} # Restricted PATH
)
return result.stdout
Performance Optimization Strategies
As your MCP servers handle more requests, performance becomes important. The Smart Day Planner demonstrates several optimization techniques:
1. Connection Pooling:
## Reuse HTTP sessions
http_session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=100), # Max 100 concurrent connections
timeout=aiohttp.ClientTimeout(total=30)
)
2. Response Caching:
from functools import lru_cache
from datetime import datetime, timedelta
cache = {}
def cached_response(key: str, ttl_minutes: int = 30):
"""Decorator for caching responses."""
def decorator(func):
async def wrapper(*args, **kwargs):
if key in cache:
data, timestamp = cache[key]
if datetime.now() - timestamp < timedelta(minutes=ttl_minutes):
return data
result = await func(*args, **kwargs)
cache[key] = (result, datetime.now())
return result
return wrapper
return decorator
3. Concurrent Operations:
## Fetch multiple resources concurrently
results = await asyncio.gather(
fetch_weather(city1),
fetch_weather(city2),
fetch_todos(),
return_exceptions=True
)
4. Lazy Loading:
## Don't load all data upfront
@server.list_resources()
async def list_resources() -> list[Resource]:
# Return metadata only, not full content
return [
Resource(uri=f"doc://{doc.id}", name=doc.title)
for doc in get_document_metadata() # Fast query
]
@server.read_resource()
async def read_resource(uri: str) -> str:
# Load full content only when requested
doc_id = extract_id(uri)
return load_full_document(doc_id) # Slower query
The MCP Ecosystem and Community Resources
The Model Context Protocol is rapidly growing, with an expanding ecosystem of servers and tools. Here are valuable resources for continued learning:
Official Resources:
- MCP Documentation - Comprehensive protocol specification and guides
- Python SDK - Official Python implementation
- MCP Specification - Technical protocol details
Community Servers:
- Awesome MCP Servers - Curated list of MCP server implementations
- GitHub MCP Server - Interact with GitHub repositories
- Slack MCP Server - Slack integration example
Development Tools:
- MCP Inspector - Interactive debugging tool
- Claude Desktop - Primary MCP client for testing
The python-mcp-servers repository you've been learning from is part of this ecosystem, providing practical, well-documented examples that demonstrate real-world patterns.
Conclusion
You've journeyed from understanding the fundamentals of MCP to building three production-quality servers, each demonstrating essential patterns for AI-tool integration. Let's recap the key insights that will guide your MCP development:
The Three Core Patterns:
- Command Wrappers expose existing tools with minimal code (Docker Dev Assistant - 401 lines)
- Data Processors add intelligence through indexing and search (Personal Knowledge Base - 784 lines)
- API Orchestrators coordinate multiple services with production patterns (Smart Day Planner - 760 lines)
Essential Implementation Principles:
- Start simple with clear tool definitions and input schemas
- Handle errors gracefully - never let exceptions crash your server
- Use async operations for I/O-bound tasks and external API calls
- Cache aggressively to reduce latency and API costs
- Log comprehensively for debugging and monitoring
- Validate inputs to prevent security vulnerabilities
The Power of MCP:
The Model Context Protocol transforms how we build AI applications. Instead of creating custom integrations for each tool or data source, you write one MCP server that works with any compatible client. The three examples in the python-mcp-servers repository demonstrate this versatility: the same patterns work for local system access, data processing, and external API integration.
Your Next Steps:
- Clone the repository and run all three servers to see them in action
- Modify an existing server to add a tool relevant to your workflow
- Build your own server using one of the three patterns as a template
- Explore the ecosystem by trying community servers for GitHub, Slack, or Google Drive
- Share your creation - contribute back to the MCP community with your own server
The Real Opportunity:
MCP servers aren't just about connecting AI to data—they're about encoding expertise into reusable tools. The "debug-container" prompt doesn't just run commands; it embodies best practices for Docker troubleshooting. The "connect-ideas" prompt doesn't just search notes; it guides systematic knowledge synthesis. When you build an MCP server, you're creating an AI-powered assistant that captures and scales your domain knowledge.
The 2,100+ lines of production-quality code in the repository aren't just examples—they're blueprints for the future of AI-tool integration. Whether you're building internal tools for your team, creating customer-facing AI features, or exploring new ways to augment human capabilities, these patterns will serve as your foundation.
Start with the Docker Dev Assistant today. By tomorrow, you could have Claude managing your development environment, searching your knowledge base, and planning your day—all through servers you built yourself. The future of AI isn't just about better models; it's about better integration. And with MCP, that future is something you can build right now.
Ready to extend your AI assistant's capabilities? Head to the python-mcp-servers repository, follow the setup guide, and start building. The code is open source, the documentation is comprehensive, and the community is growing. Your first MCP server is just a git clone away.
Christian



Member discussion