Building MCP Servers in Python: A Complete Step-by-Step Guide
Introduction
The Model Context Protocol (MCP) is an open standard
introduced by Anthropic with the goal to standardize how AI applications
(chatbots, IDE assistants, or custom agents) connect with external tools, data
sources, and systems. Think of it like USB for AI integrations. In this
comprehensive guide, we'll build a real-world MCP server from scratch using
Python.
What is MCP?
The Model Context Protocol is an open standard that enables
developers to build secure, two-way connections between their data sources and
AI-powered tools. MCP is an open protocol that standardizes how applications
provide context to LLMs. Think of MCP like a USB-C port for AI applications.
MCP Architecture Overview
┌─────────────────┐ MCP Protocol ┌─────────────────┐
│ MCP Client │◄──────────────────►│ MCP Server │
│ (AI Assistant) │ │ (Your Python App)│
└─────────────────┘ └─────────────────┘
│ │
│ │
┌────┴────┐ ┌──────┴──────┐
│ Claude │ │ Database │
│ GPT-4 │ │ APIs │
│ Others │ │ File System │
└─────────┘ └─────────────┘
Use Case: Customer Support Analytics Server
For this tutorial, we'll build an MCP server that provides
customer support analytics capabilities. This server will:
- Resources:
Expose customer ticket data and analytics reports
- Tools:
Provide functions to query tickets, generate reports, and update ticket
statuses
- Prompts:
Offer pre-defined prompts for common support scenarios
This is a practical example that demonstrates how MCP can
bridge AI assistants with business systems.
Step 1: Environment Setup
Prerequisites
- Python
3.8 or higher
- Virtual
environment (recommended)
Installation
# Create and activate virtual environment
python -m venv mcp_env
source mcp_env/bin/activate
# On Windows: mcp_env\Scripts\activate
# Install required packages
pip install fastmcp
pip install sqlite3 #
For our database example
pip install aiofiles
# For async file operations
pip install pandas
# For data analysis
Project Structure
customer_support_mcp/
├── server.py # Main MCP server
├── database.py # Database operations
├── models.py # Data models
├── analytics.py # Analytics functions
├── data/
│ └──
tickets.db # SQLite database
└── requirements.txt
Step 2: Database Setup
Let's start by creating our data models and database setup:
models.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List
from enum import Enum
class TicketStatus(Enum):
OPEN =
"open"
IN_PROGRESS =
"in_progress"
RESOLVED =
"resolved"
CLOSED =
"closed"
class Priority(Enum):
LOW =
"low"
MEDIUM =
"medium"
HIGH =
"high"
CRITICAL =
"critical"
@dataclass
class Ticket:
id: int
customer_id: str
subject: str
description: str
status:
TicketStatus
priority: Priority
created_at:
datetime
updated_at:
datetime
assigned_to:
Optional[str] = None
resolution:
Optional[str] = None
@dataclass
class Customer:
id: str
name: str
email: str
tier: str # bronze, silver, gold, platinum
total_tickets: int
database.py
import sqlite3
import aiosqlite
from typing import List, Optional
from datetime import datetime, timedelta
import json
from models import Ticket, Customer, TicketStatus, Priority
class DatabaseManager:
def __init__(self,
db_path: str = "data/tickets.db"):
self.db_path =
db_path
self.init_database()
def
init_database(self):
"""Initialize the database with tables and sample
data"""
conn =
sqlite3.connect(self.db_path)
cursor =
conn.cursor()
# Create
tables
cursor.execute('''
CREATE
TABLE IF NOT EXISTS tickets (
id
INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id
TEXT NOT NULL,
subject TEXT NOT NULL,
description TEXT NOT NULL,
status
TEXT NOT NULL,
priority TEXT NOT NULL,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL,
assigned_to TEXT,
resolution TEXT
)
''')
cursor.execute('''
CREATE
TABLE IF NOT EXISTS customers (
id
TEXT PRIMARY KEY,
name
TEXT NOT NULL,
email
TEXT NOT NULL,
tier
TEXT NOT NULL,
total_tickets INTEGER DEFAULT 0
)
''')
# Insert
sample data if tables are empty
cursor.execute("SELECT COUNT(*) FROM customers")
if
cursor.fetchone()[0] == 0:
self._insert_sample_data(cursor)
conn.commit()
conn.close()
def
_insert_sample_data(self, cursor):
"""Insert
sample customer and ticket data"""
# Sample
customers
customers = [
("CUST001", "Acme Corp",
"contact@acme.com", "gold", 5),
("CUST002", "TechStart Inc",
"support@techstart.com", "silver", 3),
("CUST003", "Global Solutions",
"help@globalsol.com", "platinum", 8),
("CUST004", "Small Business",
"owner@smallbiz.com", "bronze", 2)
]
cursor.executemany(
"INSERT INTO customers (id, name, email, tier, total_tickets)
VALUES (?, ?, ?, ?, ?)",
customers
)
# Sample
tickets
now =
datetime.now()
tickets = [
("CUST001", "Login issues", "Users cannot log
into the system", "open", "high",
now -
timedelta(hours=2), now - timedelta(hours=2), "agent1", None),
("CUST001", "Performance problems",
"Application is running slowly", "in_progress",
"medium",
now -
timedelta(days=1), now - timedelta(hours=3), "agent2", None),
("CUST002", "Feature request", "Need bulk
export functionality", "open", "low",
now -
timedelta(hours=5), now - timedelta(hours=5), None, None),
("CUST003", "Critical bug", "Data loss in
reports module", "resolved", "critical",
now -
timedelta(days=2), now - timedelta(hours=1), "agent1", "Fixed in
version 2.1.3"),
("CUST004", "Account setup", "Need help setting
up new account", "closed", "medium",
now -
timedelta(days=3), now - timedelta(days=2), "agent3", "Setup
completed successfully")
]
cursor.executemany(
"""INSERT
INTO tickets (customer_id, subject, description, status, priority,
created_at, updated_at, assigned_to, resolution) VALUES (?, ?, ?, ?, ?,
?, ?, ?, ?)""",
tickets
)
async def
get_tickets(self, status: Optional[str] = None, priority: Optional[str] = None,
customer_id: Optional[str] = None) -> List[Ticket]:
"""Get tickets with optional filtering"""
async with
aiosqlite.connect(self.db_path) as db:
query =
"SELECT * FROM tickets WHERE 1=1"
params =
[]
if status:
query
+= " AND status = ?"
params.append(status)
if
priority:
query
+= " AND priority = ?"
params.append(priority)
if
customer_id:
query
+= " AND customer_id = ?"
params.append(customer_id)
query +=
" ORDER BY created_at DESC"
async with
db.execute(query, params) as cursor:
rows =
await cursor.fetchall()
return
[self._row_to_ticket(row) for row in rows]
async def
get_ticket(self, ticket_id: int) -> Optional[Ticket]:
"""Get
a specific ticket by ID"""
async with
aiosqlite.connect(self.db_path) as db:
async with
db.execute("SELECT * FROM tickets WHERE id = ?", (ticket_id,)) as
cursor:
row =
await cursor.fetchone()
return
self._row_to_ticket(row) if row else None
async def
update_ticket_status(self, ticket_id: int, status: str, resolution:
Optional[str] = None) -> bool:
"""Update ticket status and resolution"""
async with
aiosqlite.connect(self.db_path) as db:
await
db.execute(
"UPDATE tickets SET status = ?, resolution = ?, updated_at = ?
WHERE id = ?",
(status, resolution, datetime.now(), ticket_id)
)
await
db.commit()
return
True
async def
get_customers(self) -> List[Customer]:
"""Get all customers"""
async with
aiosqlite.connect(self.db_path) as db:
async with
db.execute("SELECT * FROM customers") as cursor:
rows =
await cursor.fetchall()
return
[self._row_to_customer(row) for row in rows]
def
_row_to_ticket(self, row) -> Ticket:
"""Convert database row to Ticket
object"""
return Ticket(
id=row[0],
customer_id=row[1],
subject=row[2],
description=row[3],
status=TicketStatus(row[4]),
priority=Priority(row[5]),
created_at=datetime.fromisoformat(row[6]),
updated_at=datetime.fromisoformat(row[7]),
assigned_to=row[8],
resolution=row[9]
)
def
_row_to_customer(self, row) -> Customer:
"""Convert database row to Customer
object"""
return
Customer(
id=row[0],
name=row[1],
email=row[2],
tier=row[3],
total_tickets=row[4]
)
Step 3: Analytics Functions
analytics.py
import pandas as pd
from typing import Dict, List, Any
from datetime import datetime, timedelta
from database import DatabaseManager
from models import TicketStatus, Priority
class AnalyticsEngine:
def __init__(self,
db_manager: DatabaseManager):
self.db =
db_manager
async def
generate_summary_report(self, days: int = 30) -> Dict[str, Any]:
"""Generate a comprehensive summary
report"""
# Get all
tickets from the specified period
all_tickets =
await self.db.get_tickets()
cutoff_date =
datetime.now() - timedelta(days=days)
recent_tickets
= [t for t in all_tickets if t.created_at >= cutoff_date]
# Status
distribution
status_counts
= {}
for status in
TicketStatus:
status_counts[status.value] = len([t for t in recent_tickets if t.status
== status])
# Priority
distribution
priority_counts = {}
for priority
in Priority:
priority_counts[priority.value] = len([t for t in recent_tickets if
t.priority == priority])
# Average
resolution time for closed tickets
closed_tickets
= [t for t in recent_tickets if t.status in [TicketStatus.RESOLVED,
TicketStatus.CLOSED]]
avg_resolution_time = 0
if
closed_tickets:
total_time
= sum([(t.updated_at - t.created_at).total_seconds() for t in closed_tickets])
avg_resolution_time = total_time / len(closed_tickets) / 3600 # Convert to hours
return {
"period_days":
days,
"total_tickets": len(recent_tickets),
"status_distribution": status_counts,
"priority_distribution": priority_counts,
"average_resolution_hours": round(avg_resolution_time, 2),
"open_tickets": status_counts.get("open", 0),
"overdue_tickets": len([t for t in recent_tickets
if t.status
== TicketStatus.OPEN and
(datetime.now() - t.created_at).days > 3])
}
async def
get_customer_analytics(self, customer_id: str) -> Dict[str, Any]:
"""Get analytics for a specific
customer"""
customer_tickets = await self.db.get_tickets(customer_id=customer_id)
customers =
await self.db.get_customers()
customer =
next((c for c in customers if c.id == customer_id), None)
if not
customer:
return
{"error": "Customer not found"}
# Calculate
metrics
open_tickets =
len([t for t in customer_tickets if t.status == TicketStatus.OPEN])
resolved_tickets = len([t for t in customer_tickets
if t.status in [TicketStatus.RESOLVED, TicketStatus.CLOSED]])
# Recent
activity (last 30 days)
recent_cutoff
= datetime.now() - timedelta(days=30)
recent_tickets
= [t for t in customer_tickets if t.created_at >= recent_cutoff]
return {
"customer_id": customer_id,
"customer_name":
customer.name,
"customer_tier": customer.tier,
"total_tickets": len(customer_tickets),
"open_tickets": open_tickets,
"resolved_tickets": resolved_tickets,
"recent_tickets_30d": len(recent_tickets),
"satisfaction_trend": "stable" # This would be calculated from actual
satisfaction data
}
async def
get_agent_performance(self) -> List[Dict[str, Any]]:
"""Get performance metrics for support
agents"""
all_tickets =
await self.db.get_tickets()
agent_stats =
{}
for ticket in
all_tickets:
if
ticket.assigned_to:
agent
= ticket.assigned_to
if
agent not in agent_stats:
agent_stats[agent] = {
"name": agent,
"total_assigned": 0,
"resolved": 0,
"average_resolution_time": 0,
"open_tickets": 0
}
agent_stats[agent]["total_assigned"] += 1
if
ticket.status in [TicketStatus.RESOLVED, TicketStatus.CLOSED]:
agent_stats[agent]["resolved"]
+= 1
elif
ticket.status == TicketStatus.OPEN:
agent_stats[agent]["open_tickets"] += 1
# Calculate
resolution rates
for agent_data
in agent_stats.values():
if
agent_data["total_assigned"] > 0:
agent_data["resolution_rate"] = round(
agent_data["resolved"] /
agent_data["total_assigned"] * 100, 1
)
return
list(agent_stats.values())
Step 4: Creating the MCP Server
Now let's create the main MCP server using FastMCP:
server.py
import asyncio
import json
from typing import List, Optional, Dict, Any
from fastmcp import FastMCP
from database import DatabaseManager
from analytics import AnalyticsEngine
from models import TicketStatus, Priority
# Initialize the MCP server
mcp = FastMCP("Customer Support Analytics Server")
# Initialize database and analytics
db_manager = DatabaseManager()
analytics = AnalyticsEngine(db_manager)
#
=============================================================================
# TOOLS (Functions that the AI can call)
#
=============================================================================
@mcp.tool
async def get_tickets(
status:
Optional[str] = None,
priority:
Optional[str] = None,
customer_id:
Optional[str] = None
) -> List[Dict[str, Any]]:
"""
Retrieve customer
support tickets with optional filtering.
Args:
status: Filter
by ticket status (open, in_progress, resolved, closed)
priority:
Filter by priority (low, medium, high, critical)
customer_id:
Filter by specific customer ID
Returns:
List of ticket
dictionaries with all ticket details
"""
tickets = await
db_manager.get_tickets(status, priority, customer_id)
return [
{
"id": ticket.id,
"customer_id": ticket.customer_id,
"subject": ticket.subject,
"description":
ticket.description,
"status": ticket.status.value,
"priority": ticket.priority.value,
"created_at": ticket.created_at.isoformat(),
"updated_at": ticket.updated_at.isoformat(),
"assigned_to": ticket.assigned_to,
"resolution": ticket.resolution
}
for ticket in
tickets
]
@mcp.tool
async def update_ticket_status(
ticket_id: int,
status: str,
resolution:
Optional[str] = None
) -> Dict[str, Any]:
"""
Update the status
of a customer support ticket.
Args:
ticket_id: The
ID of the ticket to update
status: New
status (open, in_progress, resolved, closed)
resolution:
Resolution description (required for resolved/closed status)
Returns:
Success status
and updated ticket information
"""
# Validate status
valid_statuses =
[s.value for s in TicketStatus]
if status not in
valid_statuses:
return {
"success":
False,
"error": f"Invalid status. Must be one of: {',
'.join(valid_statuses)}"
}
# Check if
resolution is required
if status in
["resolved", "closed"] and not resolution:
return {
"success": False,
"error": "Resolution description is required for
resolved/closed tickets"
}
success = await
db_manager.update_ticket_status(ticket_id, status, resolution)
if success:
updated_ticket
= await db_manager.get_ticket(ticket_id)
return {
"success": True,
"message": f"Ticket {ticket_id} status updated to
{status}",
"ticket": {
"id": updated_ticket.id,
"status": updated_ticket.status.value,
"resolution": updated_ticket.resolution,
"updated_at": updated_ticket.updated_at.isoformat()
} if
updated_ticket else None
}
else:
return {
"success": False,
"error":
f"Failed to update ticket {ticket_id}"
}
@mcp.tool
async def generate_analytics_report(days: int = 30) ->
Dict[str, Any]:
"""
Generate a
comprehensive analytics report for customer support.
Args:
days: Number
of days to include in the report (default: 30)
Returns:
Detailed
analytics report with metrics and insights
"""
report = await
analytics.generate_summary_report(days)
return report
@mcp.tool
async def get_customer_analytics(customer_id: str) ->
Dict[str, Any]:
"""
Get detailed
analytics for a specific customer.
Args:
customer_id:
The ID of the customer to analyze
Returns:
Customer-specific
analytics and metrics
"""
customer_analytics
= await analytics.get_customer_analytics(customer_id)
return
customer_analytics
@mcp.tool
async def get_agent_performance() -> List[Dict[str,
Any]]:
"""
Get performance
metrics for all support agents.
Returns:
List of agent
performance metrics including resolution rates
"""
performance_data =
await analytics.get_agent_performance()
return
performance_data
#
=============================================================================
# RESOURCES (Data that can be loaded into AI context)
#
=============================================================================
@mcp.resource("support://tickets/recent")
async def recent_tickets() -> str:
"""Recent customer support tickets (last 7
days)"""
from datetime
import datetime, timedelta
cutoff_date =
datetime.now() - timedelta(days=7)
all_tickets =
await db_manager.get_tickets()
recent_tickets =
[t for t in all_tickets if t.created_at >= cutoff_date]
report = "#
Recent Customer Support Tickets (Last 7 Days)\n\n"
if not
recent_tickets:
report +=
"No tickets created in the last 7 days.\n"
else:
for ticket in
recent_tickets:
report +=
f"## Ticket #{ticket.id}\n"
report +=
f"**Customer:** {ticket.customer_id}\n"
report +=
f"**Subject:** {ticket.subject}\n"
report +=
f"**Status:** {ticket.status.value.title()}\n"
report +=
f"**Priority:** {ticket.priority.value.title()}\n"
report +=
f"**Created:** {ticket.created_at.strftime('%Y-%m-%d %H:%M')}\n"
if
ticket.assigned_to:
report
+= f"**Assigned to:** {ticket.assigned_to}\n"
report +=
f"**Description:** {ticket.description}\n\n"
return report
@mcp.resource("support://analytics/dashboard")
async def analytics_dashboard() -> str:
"""Customer support analytics dashboard"""
report_data =
await analytics.generate_summary_report(30)
agent_performance
= await analytics.get_agent_performance()
dashboard =
"# Customer Support Analytics Dashboard\n\n"
# Summary metrics
dashboard +=
"## Summary (Last 30 Days)\n"
dashboard +=
f"- **Total Tickets:** {report_data['total_tickets']}\n"
dashboard +=
f"- **Open Tickets:** {report_data['open_tickets']}\n"
dashboard +=
f"- **Overdue Tickets:** {report_data['overdue_tickets']}\n"
dashboard +=
f"- **Average Resolution Time:** {report_data['average_resolution_hours']}
hours\n\n"
# Status
distribution
dashboard +=
"## Ticket Status Distribution\n"
for status, count
in report_data['status_distribution'].items():
dashboard +=
f"- **{status.title()}:** {count}\n"
dashboard +=
"\n"
# Priority
distribution
dashboard +=
"## Priority Distribution\n"
for priority,
count in report_data['priority_distribution'].items():
dashboard +=
f"- **{priority.title()}:** {count}\n"
dashboard +=
"\n"
# Agent
performance
dashboard +=
"## Agent Performance\n"
for agent in
agent_performance:
dashboard +=
f"### {agent['name']}\n"
dashboard +=
f"- **Total Assigned:** {agent['total_assigned']}\n"
dashboard +=
f"- **Resolved:** {agent['resolved']}\n"
dashboard +=
f"- **Resolution Rate:** {agent.get('resolution_rate', 0)}%\n"
dashboard +=
f"- **Open Tickets:** {agent['open_tickets']}\n\n"
return dashboard
@mcp.resource("support://customers/list")
async def customer_list() -> str:
"""List of all customers with basic
information"""
customers = await
db_manager.get_customers()
customer_list =
"# Customer Directory\n\n"
for customer in
customers:
customer_list
+= f"## {customer.name}\n"
customer_list
+= f"- **Customer ID:** {customer.id}\n"
customer_list
+= f"- **Email:** {customer.email}\n"
customer_list
+= f"- **Tier:** {customer.tier.title()}\n"
customer_list
+= f"- **Total Tickets:** {customer.total_tickets}\n\n"
return
customer_list
#
=============================================================================
# PROMPTS (Pre-defined prompts for common scenarios)
#
=============================================================================
@mcp.prompt("analyze_ticket_trends")
async def analyze_ticket_trends() -> str:
"""Analyze customer support ticket trends and provide
insights"""
return
"""
Please analyze the
recent customer support tickets and provide insights on:
1. **Trend
Analysis**:
- Are ticket
volumes increasing or decreasing?
- What are the
most common issues?
- Which
customers are submitting the most tickets?
2. **Performance
Metrics**:
- How is the
team performing on resolution times?
- Are there any
concerning patterns in ticket statuses?
- Which agents
need additional support?
3.
**Recommendations**:
- What process
improvements would you suggest?
- Are there any
recurring issues that need attention?
- How can we
improve customer satisfaction?
Use the available
tools to gather the necessary data and provide a comprehensive analysis.
"""
@mcp.prompt("customer_health_check")
async def customer_health_check() -> str:
"""Perform a customer health check
analysis"""
return
"""
Please perform a
comprehensive customer health check by:
1. **Customer
Analysis**:
- Review each
customer's ticket history
- Identify
customers with multiple open tickets
- Look for
patterns in customer complaints
2. **Risk
Assessment**:
- Identify
at-risk customers (high ticket volume, unresolved issues)
- Highlight
customers who might be considering churn
- Note any
escalation patterns
3. **Action
Items**:
- Recommend
immediate actions for high-risk customers
- Suggest
proactive outreach opportunities
- Identify
upselling or cross-selling opportunities
Use the customer
analytics tools to gather detailed information for each customer.
"""
@mcp.prompt("weekly_support_report")
async def weekly_support_report() -> str:
"""Generate a weekly support team
report"""
return
"""
Please generate a
comprehensive weekly support report including:
1. **Week
Overview**:
- Total tickets
handled
- New tickets
vs. resolved tickets
- Overall team
performance metrics
2. **Agent
Performance**:
- Individual
agent statistics
- Resolution
rates and times
- Workload
distribution
3. **Customer
Insights**:
- Top customers
by ticket volume
- Most common
issue categories
- Customer
satisfaction indicators
4. **Action
Items**:
- Priority
items for next week
- Process
improvements needed
- Training
recommendations
Format the report
in a professional manner suitable for management review.
"""
#
=============================================================================
# SERVER STARTUP
#
=============================================================================
if __name__ == "__main__":
print("🎧
Starting Customer Support Analytics MCP Server...")
print("📊
Database initialized with sample data")
print("🔧
Available tools: get_tickets, update_ticket_status,
generate_analytics_report")
print("📋
Available resources: recent tickets, analytics dashboard, customer list")
print("💡
Available prompts: analyze trends, health check, weekly report")
print("\n🚀
Server running on stdio transport...")
mcp.run()
Step 5: Testing the MCP Server
requirements.txt
fastmcp>=0.9.0
aiosqlite>=0.19.0
pandas>=2.0.0
aiofiles>=23.0.0
Test Script (test_server.py)
import asyncio
import json
from database import DatabaseManager
from analytics import AnalyticsEngine
async def test_server_components():
"""Test the server components before running the MCP
server"""
print("🧪
Testing Customer Support MCP Server Components...\n")
# Test database
print("1.
Testing Database Connection...")
db =
DatabaseManager()
tickets = await
db.get_tickets()
print(f" ✅ Found {len(tickets)} tickets in
database")
customers = await
db.get_customers()
print(f" ✅ Found {len(customers)}
customers in database")
# Test analytics
print("\n2.
Testing Analytics Engine...")
analytics =
AnalyticsEngine(db)
report = await
analytics.generate_summary_report(30)
print(f" ✅ Generated summary report:
{report['total_tickets']} tickets in last 30 days")
customer_analytics
= await analytics.get_customer_analytics("CUST001")
print(f" ✅ Customer analytics for CUST001:
{customer_analytics['total_tickets']} total tickets")
agent_performance
= await analytics.get_agent_performance()
print(f" ✅ Agent performance data:
{len(agent_performance)} agents")
# Test ticket
operations
print("\n3.
Testing Ticket Operations...")
open_tickets =
await db.get_tickets(status="open")
print(f" ✅ Found {len(open_tickets)} open
tickets")
if open_tickets:
ticket_id =
open_tickets[0].id
success =
await db.update_ticket_status(ticket_id, "in_progress")
print(f" ✅
Updated ticket {ticket_id} status: {success}")
print("\n🎉
All tests passed! Server is ready to run.")
if __name__ == "__main__":
asyncio.run(test_server_components())
Step 6: Integration with AI Clients
Now that we have a functional MCP server, let's integrate it
with AI clients like Claude Desktop or other MCP-compatible applications.
6.1 Claude Desktop Integration
First, we need to configure Claude Desktop to recognize our
MCP server. Create or modify the Claude Desktop configuration file:
On macOS: ~/Library/Application
Support/Claude/claude_desktop_config.json On Windows: %APPDATA%/Claude/claude_desktop_config.json
{
"mcpServers": {
"file-operations": {
"command": "python",
"args":
["/path/to/your/file_operations_server.py"],
"env":
{
"WORKING_DIRECTORY": "/path/to/safe/directory"
}
}
}
}
6.2 Testing the Integration
Start Claude Desktop and verify that your MCP server is
loaded. You should see tools available for:
- Reading
files
- Writing
files
- Listing
directories
- Managing
file operations
Test with prompts like:
- "List
the files in my current directory"
- "Read
the contents of config.txt"
- "Create
a new file called notes.md with some sample content"
MCP Communication Flow
AI Client Request:
┌─────────────────────────────────────────────────────┐
│ "Show me all high priority open tickets" │
└─────────────────┬───────────────────────────────────┘
│
▼
MCP Protocol Translation:
┌─────────────────────────────────────────────────────┐
│ tool_call: get_tickets(status="open", │
│ priority="high") │
└─────────────────┬───────────────────────────────────┘
│
▼
Python Server Execution:
┌─────────────────────────────────────────────────────┐
│ 1. Query SQLite database │
│ 2. Filter by status and priority │
│ 3. Return formatted results │
└─────────────────┬───────────────────────────────────┘
│
▼
Response to AI:
┌─────────────────────────────────────────────────────┐
│ [{"id": 1, "subject": "Login issues", │
│ "priority": "high", "status": "open"...}] │
└─────────────────────────────────────────────────────
Step 7: Advanced MCP Features
7.1 Adding Resource Support
Resources in MCP allow servers to expose data that can be
read by clients. Let's add file resource support:
import mimetypes
from pathlib import Path
class FileOperationsServer:
def
__init__(self):
self.working_dir = Path(os.getenv('WORKING_DIRECTORY', '.'))
async def
list_resources(self) -> list[types.Resource]:
"""List available file resources"""
resources = []
# Recursively
find all files
for file_path
in self.working_dir.rglob('*'):
if
file_path.is_file():
relative_path = file_path.relative_to(self.working_dir)
mime_type, _ = mimetypes.guess_type(str(file_path))
resources.append(types.Resource(
uri=f"file://{relative_path}",
name=str(relative_path),
description=f"File: {relative_path}",
mimeType=mime_type or "application/octet-stream"
))
return
resources
async def
read_resource(self, uri: str) -> str:
"""Read a specific resource"""
if not
uri.startswith("file://"):
raise
ValueError(f"Unsupported URI scheme: {uri}")
file_path =
self.working_dir / uri[7:] # Remove
"file://" prefix
if not
file_path.exists():
raise
FileNotFoundError(f"File not found: {file_path}")
if not
self._is_safe_path(file_path):
raise
PermissionError("Access denied")
try:
return
file_path.read_text(encoding='utf-8')
except
UnicodeDecodeError:
# Handle
binary files
return
f"Binary file: {file_path.name} ({file_path.stat().st_size} bytes)"
# Update the server setup
@server.list_resources()
async def handle_list_resources() ->
list[types.Resource]:
return await
file_ops.list_resources()
@server.read_resource()
async def handle_read_resource(uri: str) -> str:
return await
file_ops.read_resource(uri)
7.2 Adding Prompt Support
Prompts allow your server to provide reusable prompt
templates:
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
return [
types.Prompt(
name="analyze_file",
description="Analyze a file's content and structure",
arguments=[
types.PromptArgument(
name="file_path",
description="Path to the file to analyze",
required=True
)
]
),
types.Prompt(
name="create_project_structure",
description="Create a new project structure",
arguments=[
types.PromptArgument(
name="project_name",
description="Name of the project",
required=True
),
types.PromptArgument(
name="project_type",
description="Type of project (web, cli, api, etc.)",
required=False
)
]
)
]
@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict[str,
str] | None) -> types.GetPromptResult:
if name ==
"analyze_file":
file_path =
arguments.get("file_path") if arguments else None
if not
file_path:
raise
ValueError("file_path is required")
return
types.GetPromptResult(
description=f"Analyze the file: {file_path}",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"""Please analyze the file at {file_path}.
Consider:
1. File structure and organization
2. Code quality and patterns
3. Potential improvements
4. Dependencies and imports
5. Security considerations
Provide a comprehensive analysis with specific
recommendations."""
)
)
]
)
elif name ==
"create_project_structure":
project_name =
arguments.get("project_name") if arguments else None
project_type =
arguments.get("project_type", "web") if arguments else
"web"
return
types.GetPromptResult(
description=f"Create project structure for {project_name}",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"""Create a complete project structure for a
{project_type} project named '{project_name}'.
Include:
- Appropriate directory structure
- Configuration files
- Documentation templates
- Testing setup
- Build/deployment configurations
- Best practices for {project_type} projects
Create all necessary files with appropriate starter
content."""
)
)
]
)
raise
ValueError(f"Unknown prompt: {name}")
Step 8: Error Handling and Logging
Robust error handling is crucial for production MCP servers:
import logging
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s
- %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_server.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class FileOperationsServer:
def
__init__(self):
self.working_dir = Path(os.getenv('WORKING_DIRECTORY', '.'))
logger.info(f"Initialized FileOperationsServer with working
directory: {self.working_dir}")
async def
read_file_safe(self, file_path: str) -> str:
"""Safely read a file with comprehensive error
handling"""
try:
logger.info(f"Reading file: {file_path}")
full_path
= self.working_dir / file_path
# Security
checks
if not
self._is_safe_path(full_path):
logger.warning(f"Access denied for path: {full_path}")
raise
PermissionError(f"Access denied: {file_path}")
if not
full_path.exists():
logger.error(f"File
not found: {full_path}")
raise
FileNotFoundError(f"File not found: {file_path}")
if
full_path.is_dir():
logger.error(f"Path is a directory, not a file: {full_path}")
raise
IsADirectoryError(f"Path is a directory: {file_path}")
# Size
check (prevent reading huge files)
file_size
= full_path.stat().st_size
if
file_size > 10 * 1024 * 1024: # 10MB
limit
logger.warning(f"File too large: {full_path} ({file_size}
bytes)")
raise
ValueError(f"File too large: {file_path} ({file_size} bytes)")
content =
full_path.read_text(encoding='utf-8')
logger.info(f"Successfully read file: {file_path} ({len(content)}
characters)")
return
content
except
UnicodeDecodeError as e:
logger.error(f"Unicode decode error reading {file_path}: {e}")
raise
ValueError(f"Unable to read file as text: {file_path}")
except
PermissionError as e:
logger.error(f"Permission error: {e}")
raise
except
Exception as e:
logger.error(f"Unexpected error reading {file_path}: {e}")
raise
RuntimeError(f"Failed to read file: {str(e)}")
# Update tool handlers with better error handling
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict |
None) -> list[types.TextContent]:
try:
if name ==
"read_file":
file_path
= arguments.get("path") if arguments else None
if not
file_path:
raise
ValueError("Missing required argument: path")
content =
await file_ops.read_file_safe(file_path)
return
[types.TextContent(type="text", text=content)]
# ... other
tools ...
except
FileNotFoundError as e:
logger.error(f"File not found: {e}")
return
[types.TextContent(type="text", text=f"Error: {str(e)}")]
except
PermissionError as e:
logger.error(f"Permission denied: {e}")
return
[types.TextContent(type="text", text=f"Error: {str(e)}")]
except ValueError
as e:
logger.error(f"Invalid input: {e}")
return
[types.TextContent(type="text", text=f"Error: {str(e)}")]
except Exception
as e:
logger.error(f"Unexpected error in tool {name}: {e}")
return
[types.TextContent(type="text", text=f"Internal error:
{str(e)}")]
Step 9: Testing and Validation
9.1 Unit Tests
Create comprehensive tests for your MCP server:
# test_file_operations_server.py
import pytest
import tempfile
import os
from pathlib import Path
from unittest.mock import patch, AsyncMock
from your_server import FileOperationsServer
class TestFileOperationsServer:
@pytest.fixture
def
temp_dir(self):
with
tempfile.TemporaryDirectory() as temp_dir:
yield
Path(temp_dir)
@pytest.fixture
def server(self,
temp_dir):
with
patch.dict(os.environ, {'WORKING_DIRECTORY': str(temp_dir)}):
return
FileOperationsServer()
@pytest.mark.asyncio
async def
test_read_file_success(self, server, temp_dir):
# Create test
file
test_file =
temp_dir / "test.txt"
test_content =
"Hello, MCP!"
test_file.write_text(test_content)
# Test reading
result = await
server.read_file_safe("test.txt")
assert result
== test_content
@pytest.mark.asyncio
async def
test_read_file_not_found(self, server):
with
pytest.raises(FileNotFoundError):
await
server.read_file_safe("nonexistent.txt")
@pytest.mark.asyncio
async def
test_security_path_traversal(self, server):
with
pytest.raises(PermissionError):
await
server.read_file_safe("../../../etc/passwd")
@pytest.mark.asyncio
async def
test_list_files(self, server, temp_dir):
# Create test
files
(temp_dir /
"file1.txt").write_text("content1")
(temp_dir /
"file2.py").write_text("print('hello')")
(temp_dir /
"subdir").mkdir()
(temp_dir /
"subdir" / "file3.md").write_text("# Header")
files = await
server.list_files(".")
assert
len(files) >= 3
assert
any("file1.txt" in f for f in files)
assert
any("file2.py" in f for f in files)
# Run tests
# pytest test_file_operations_server.py -v
9.2 Integration Tests
Test the complete MCP protocol integration:
# test_mcp_integration.py
import pytest
import json
from mcp.server.stdio import stdio_server
from your_server import server
class TestMCPIntegration:
@pytest.mark.asyncio
async def
test_list_tools(self):
"""Test
that tools are properly listed"""
tools = await
server.list_tools()
tool_names =
[tool.name for tool in tools]
expected_tools
= ["read_file", "write_file", "list_files",
"create_directory"]
assert
all(tool in tool_names for tool in expected_tools)
@pytest.mark.asyncio
async def
test_tool_schemas(self):
"""Test that tool schemas are valid"""
tools = await
server.list_tools()
for tool in
tools:
assert
tool.name
assert
tool.description
assert
tool.inputSchema
# Validate
JSON schema
assert
"type" in tool.inputSchema
assert
"properties" in tool.inputSchema
Step 10: Deployment and Production Considerations
10.1 Configuration Management
Create a robust configuration system:
# config.py
import os
from dataclasses import dataclass
from pathlib import Path
import json
@dataclass
class ServerConfig:
working_directory:
Path
max_file_size: int
= 10 * 1024 * 1024 # 10MB
allowed_extensions: list[str] = None
log_level: str =
"INFO"
enable_write_operations: bool = True
enable_delete_operations: bool = False
@classmethod
def from_env(cls)
-> 'ServerConfig':
return cls(
working_directory=Path(os.getenv('WORKING_DIRECTORY', '.')),
max_file_size=int(os.getenv('MAX_FILE_SIZE', str(10 * 1024 * 1024))),
allowed_extensions=os.getenv('ALLOWED_EXTENSIONS', '').split(',') if
os.getenv('ALLOWED_EXTENSIONS') else None,
log_level=os.getenv('LOG_LEVEL', 'INFO'),
enable_write_operations=os.getenv('ENABLE_WRITE_OPS', 'true').lower() ==
'true',
enable_delete_operations=os.getenv('ENABLE_DELETE_OPS', 'false').lower()
== 'true'
)
@classmethod
def from_file(cls,
config_path: str) -> 'ServerConfig':
with
open(config_path, 'r') as f:
data =
json.load(f)
return cls(
working_directory=Path(data['working_directory']),
max_file_size=data.get('max_file_size', 10 * 1024 * 1024),
allowed_extensions=data.get('allowed_extensions'),
log_level=data.get('log_level', 'INFO'),
enable_write_operations=data.get('enable_write_operations', True),
enable_delete_operations=data.get('enable_delete_operations', False)
)
# Usage in server
config = ServerConfig.from_env()
file_ops = FileOperationsServer(config)
10.2 Security Hardening
Implement additional security measures:
import hashlib
import hmac
from datetime import datetime, timedelta
class SecureFileOperationsServer:
def __init__(self,
config: ServerConfig):
self.config =
config
self.working_dir = config.working_directory.resolve()
self.access_log = []
def
_validate_file_extension(self, file_path: Path) -> bool:
"""Check if file extension is allowed"""
if not
self.config.allowed_extensions:
return
True
return
file_path.suffix.lower() in self.config.allowed_extensions
def
_rate_limit_check(self, operation: str, limit: int = 10) -> bool:
"""Simple rate limiting"""
now =
datetime.now()
recent_ops = [
log for
log in self.access_log
if
log['timestamp'] > now - timedelta(minutes=1) and log['operation'] ==
operation
]
if
len(recent_ops) >= limit:
return
False
self.access_log.append({
'operation': operation,
'timestamp': now
})
# Clean old
logs
self.access_log = [
log for
log in self.access_log
if
log['timestamp'] > now - timedelta(hours=1)
]
return True
async def
read_file_secure(self, file_path: str) -> str:
"""Secure file reading with all checks"""
if not
self._rate_limit_check('read_file'):
raise
PermissionError("Rate limit exceeded")
full_path =
self.working_dir / file_path
if not
self._is_safe_path(full_path):
raise
PermissionError("Access denied")
if not
self._validate_file_extension(full_path):
raise
PermissionError(f"File extension not allowed: {full_path.suffix}")
return await
self.read_file_safe(file_path)
Conclusion
You've now built a comprehensive MCP server with advanced
features including:
- Core
Functionality: File operations with proper error handling
- Security:
Path validation, rate limiting, and access controls
- Advanced
Features: Resources, prompts, and comprehensive logging
- Testing:
Unit and integration tests
- Production
Ready: Configuration management and security hardening
Next Steps
- Extend
Functionality: Add more domain-specific tools
- Performance:
Implement caching and optimization
- Monitoring:
Add metrics and health checks
- Documentation:
Create comprehensive API documentation
- Distribution:
Package and distribute your MCP server
Best Practices Summary
- Always
validate and sanitize inputs
- Implement
comprehensive error handling
- Use
structured logging for debugging
- Follow
security best practices
- Write
thorough tests
- Document
your server's capabilities
- Consider
rate limiting and resource constraints
- Make
configuration flexible and environment-specific
Your MCP server is now ready to integrate with AI clients
and provide powerful file operations capabilities in a secure, reliable manner!
Comments
Post a Comment