concept_number: 20 title: Model Context Protocol (MCP) stage: Advanced difficulty: ⭐⭐⭐☆☆ time_to_learn: 3-4 hours status: not-started date_created: 2026-03-07 date_started: date_completed: progress: 0% prerequisites:
- "Agents"
- "Retrieval Augmented Generation"
- "Large Language Models" next_concepts:
- "Agents"
- "RAG" related_concepts:
- "Context Engineering"
- "Vector Database" tags:
- ai
- mcp
- protocol
- integration
- anthropic
- claude
🗺️ Concept Mind Map
MODEL CONTEXT PROTOCOL (MCP)
|
+-----------+-----------+
| | |
COMPONENTS BENEFITS USE CASES
| | |
| | |
MCP Client Standardization Enterprise
- Claude - One protocol - Company data
- ChatGPT - Any tool - Databases
- Custom AI - No custom - CRM systems
- Your app code - File systems
| | |
| | |
MCP Server Interoperability Development
- Salesforce- Works with - IDE integration
- Google all AIs - Code repos
Drive - Future-proof - Documentation
- Slack - Ecosystem - APIs
- Database | |
| | |
Protocol Security Personal
- JSON-RPC - Auth tokens - Local files
- WebSocket - Permissions - Databases
- HTTP - Encryption - Notes
- Standard - Audit logs - Calendar
| | |
| | |
Resources Scalability Automation
- Files - Handle 1000s - Workflows
- Data of tools - Agents
- APIs - Efficient - Pipelines
- Tools - Caching - Integration
Architecture:
┌─────────────────────────────────────┐
│ AI APPLICATION (Client) │
│ - Claude, ChatGPT, Custom AI │
└─────────────┬───────────────────────┘
│ MCP Protocol
│ (Standardized)
┌─────────────┴───────────────────────┐
│ MCP SERVERS (Data Sources) │
├─────────────────────────────────────┤
│ • Salesforce MCP Server │
│ • Google Drive MCP Server │
│ • Slack MCP Server │
│ • Database MCP Server │
│ • File System MCP Server │
│ • Custom API MCP Server │
└─────────────────────────────────────┘
Before vs After MCP:
BEFORE MCP:
AI ─┬─ Custom Connector 1 ─→ Salesforce
├─ Custom Connector 2 ─→ Google Drive
├─ Custom Connector 3 ─→ Slack
├─ Custom Connector 4 ─→ Database
└─ Custom Connector N ─→ Tool N
Problem: N tools = N custom integrations!
AFTER MCP:
AI ─→ MCP Protocol ─┬─→ Salesforce (MCP)
├─→ Google Drive (MCP)
├─→ Slack (MCP)
├─→ Database (MCP)
└─→ Any Tool (MCP)
Solution: 1 protocol = ∞ tools!
MCP Message Flow:
1. USER: "What's our Q4 revenue?"
↓
2. AI: Needs Salesforce data
↓
3. MCP REQUEST:
{
"method": "query",
"resource": "salesforce://revenue",
"params": {"quarter": "Q4"}
}
↓
4. MCP SERVER: Queries Salesforce
↓
5. MCP RESPONSE:
{
"data": {"revenue": "$2.5M"},
"metadata": {"source": "Salesforce"}
}
↓
6. AI: "Q4 revenue was $2.5M, up 15% from Q3"
Key Benefits:
- Standardization: One protocol for all tools
- Interoperability: Works with any AI
- Security: Built-in auth and permissions
- Scalability: Add tools without code changes
- Future-proof: New tools work automatically
🌟 Start With a Real-World Scenario
You want ChatGPT to access your company data:
The Problem:
You: "What's our Q4 revenue?"
ChatGPT: "I don't have access to your company's financial data."
You: "Check our Salesforce database"
ChatGPT: "I can't access external databases."
You: "Read the latest sales report"
ChatGPT: "I can't access your files."
❌ ChatGPT is isolated! Can't access YOUR data!
The Old Solution (Messy):
Build custom integrations for EACH tool:
- Custom Salesforce connector
- Custom Google Drive connector
- Custom Slack connector
- Custom Database connector
- Custom API for everything!
Result: 100+ custom integrations, maintenance nightmare!
The New Solution: Model Context Protocol (MCP)
One standard protocol for ALL tools!
You: "What's our Q4 revenue?"
MCP:
1. Connects to Salesforce (via MCP)
2. Queries revenue data
3. Returns to ChatGPT
4. ChatGPT answers: "Q4 revenue was $2.5M, up 15% from Q3"
✅ Works with ANY tool that supports MCP!
The Power of MCP:
Before MCP:
ChatGPT → Custom Connector 1 → Salesforce
ChatGPT → Custom Connector 2 → Google Drive
ChatGPT → Custom Connector 3 → Slack
ChatGPT → Custom Connector 4 → Database
... (100+ custom connectors!)
With MCP:
ChatGPT → MCP Protocol → Any Tool!
- Salesforce (MCP server)
- Google Drive (MCP server)
- Slack (MCP server)
- Database (MCP server)
- ANY tool with MCP support!
One protocol, infinite tools! 🔌
Real Example:
MCP Server for Your Database:
# Your database becomes accessible to ANY AI!
mcp_server = MCPServer(
name="company_db",
tools=[
"query_revenue",
"get_customers",
"sales_report"
]
)
# Now Claude, ChatGPT, or ANY AI can use it!
Using it:
You: "Show me top 10 customers by revenue"
Claude (via MCP):
1. Calls company_db.query_revenue()
2. Gets data
3. Formats response
4. Returns: "Top 10 customers:
1. Acme Corp - $500K
2. TechStart - $450K
..."
Real-World Applications:
1. Enterprise AI Assistant
MCP connects to:
- Salesforce (CRM data)
- Google Workspace (docs, emails)
- Slack (team communication)
- Jira (project management)
- Database (company data)
Result: AI that knows EVERYTHING about your company!
2. Personal AI Assistant
MCP connects to:
- Calendar (schedule)
- Email (communications)
- Notes (personal knowledge)
- Files (documents)
- Browser (web access)
Result: AI that manages your life!
3. Development Tools
MCP connects to:
- GitHub (code)
- Terminal (commands)
- Database (data)
- APIs (services)
- Documentation (knowledge)
Result: AI that codes with full context!
The MCP Ecosystem:
MCP Servers (Data Sources):
- Filesystem (read/write files)
- Database (SQL queries)
- Web (HTTP requests)
- Git (version control)
- Custom (anything you build!)
MCP Clients (AI Models):
- Claude (Anthropic)
- ChatGPT (OpenAI)
- Custom AI apps
- Any LLM with MCP support
MCP Tools (Actions):
- Read data
- Write data
- Execute commands
- Call APIs
- Anything programmable!
The Benefits:
For Developers:
- Build once, works with all AIs
- Standard protocol (no custom code)
- Easy to maintain
- Reusable components
For Users:
- AI that accesses YOUR data
- Seamless integrations
- Privacy (data stays local)
- Powerful capabilities
For Companies:
- Reduce integration costs
- Faster deployment
- Better AI applications
- Competitive advantage
By the end of this guide, you'll understand:
- How MCP works (protocol details)
- How to build MCP servers
- How to connect AI to your tools
- Security and privacy considerations
- Real-world deployment
- Future of AI integrations
The revolution: MCP is the "USB for AI" - one standard protocol to connect AI to everything! 🔌
Let's dive in! 🚀
🔌 Model Context Protocol (MCP)
Standardized protocol for connecting AI models to external data sources and tools
zations](#interactive-tools--visualizations)
- Code Examples
- Hands-On Exercises
- Assignments & Practice
- Common Pitfalls & Debugging
- Fun Facts & Hacks
- Related Concepts
- Resources
What is MCP?
Model Context Protocol (MCP) is an open standard developed by Anthropic that enables AI models to securely connect to external data sources, tools, and services. Think of it as a universal adapter that lets AI assistants access your files, databases, APIs, and applications in a standardized way.
Key Idea
AI Model (Claude, GPT, etc.)
↕️ MCP Protocol
MCP Server
↕️
External Resources (Files, DBs, APIs, Tools)
Real-World Analogy
Think of MCP like USB-C:
- Before: Every device had different connectors (chaos!)
- After: One standard connector works everywhere
- MCP: One protocol connects AI to any data source
- Result: Build once, use with any AI model
Why MCP Matters
1. Standardization 🔧
- One protocol for all AI models
- No vendor lock-in
- Reusable integrations
2. Security 🔒
- Controlled access to resources
- Permission management
- Audit trails
3. Flexibility 🎯
- Connect to any data source
- Add custom tools
- Extend functionality
4. Composability 🧩
- Mix and match servers
- Build complex workflows
- Modular architecture
5. Developer Experience 💻
- Simple API
- Clear documentation
- Active ecosystem
Core Concepts
1. MCP Components
┌─────────────────┐
│ MCP Client │ (AI Application)
│ (Claude, etc.) │
└────────┬────────┘
│ MCP Protocol
┌────────┴────────┐
│ MCP Server │ (Integration Layer)
└────────┬────────┘
│
┌────────┴────────┐
│ Resources │ (Files, DBs, APIs)
└─────────────────┘
2. Resources
Resources are data sources that AI can read:
# File resource
{
"uri": "file:///path/to/document.txt",
"name": "Project Documentation",
"mimeType": "text/plain"
}
# Database resource
{
"uri": "postgres://localhost/mydb/users",
"name": "User Database",
"mimeType": "application/sql"
}
# API resource
{
"uri": "https://api.example.com/data",
"name": "External API",
"mimeType": "application/json"
}
3. Tools
Tools are functions that AI can execute:
# Calculator tool
{
"name": "calculate",
"description": "Perform mathematical calculations",
"inputSchema": {
"type": "object",
"properties": {
"expression": {"type": "string"}
}
}
}
# Search tool
{
"name": "search",
"description": "Search the web",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
}
}
}
4. Prompts
Reusable prompt templates:
{
"name": "code_review",
"description": "Review code for best practices",
"arguments": [
{
"name": "code",
"description": "Code to review",
"required": True
}
]
}
How It Works
MCP Request-Response Flow
sequenceDiagram
participant Client as AI Client
participant Server as MCP Server
participant Resource as Data Source
Client->>Server: List Resources
Server-->>Client: Available Resources
Client->>Server: Read Resource
Server->>Resource: Fetch Data
Resource-->>Server: Data
Server-->>Client: Resource Content
Client->>Server: Call Tool
Server->>Resource: Execute Action
Resource-->>Server: Result
Server-->>Client: Tool Response
Message Types
1. Initialize
{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "1.0",
"capabilities": {
"resources": {},
"tools": {}
}
}
}
2. List Resources
{
"jsonrpc": "2.0",
"method": "resources/list"
}
3. Read Resource
{
"jsonrpc": "2.0",
"method": "resources/read",
"params": {
"uri": "file:///path/to/file.txt"
}
}
4. Call Tool
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "search",
"arguments": {
"query": "MCP protocol"
}
}
}
MCP Architecture
Server Types
1. Filesystem Server
# Provides access to local files
mcp_server = FilesystemServer(
root_path="/path/to/documents",
allowed_extensions=[".txt", ".md", ".pdf"]
)
2. Database Server
# Connects to databases
mcp_server = DatabaseServer(
connection_string="postgresql://localhost/mydb",
allowed_tables=["users", "products"]
)
3. API Server
# Wraps external APIs
mcp_server = APIServer(
base_url="https://api.example.com",
auth_token="your-token"
)
4. Custom Server
# Your own integration
class CustomMCPServer(MCPServer):
def list_resources(self):
# Return available resources
pass
def read_resource(self, uri):
# Fetch resource content
pass
def list_tools(self):
# Return available tools
pass
def call_tool(self, name, arguments):
# Execute tool
pass
Local Setup
Installation (15-20 minutes)
# Install MCP SDK
pip install mcp
# Install MCP servers (examples)
pip install mcp-server-filesystem
pip install mcp-server-postgres
pip install mcp-server-github
# Install Claude Desktop (for testing)
# Download from: https://claude.ai/download
Quick Start - Filesystem Server
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool
import os
# Create MCP server
app = Server("filesystem-server")
@app.list_resources()
async def list_resources():
"""List available files"""
files = []
for root, dirs, filenames in os.walk("/path/to/docs"):
for filename in filenames:
if filename.endswith(('.txt', '.md')):
path = os.path.join(root, filename)
files.append(Resource(
uri=f"file://{path}",
name=filename,
mimeType="text/plain"
))
return files
@app.read_resource()
async def read_resource(uri: str):
"""Read file content"""
path = uri.replace("file://", "")
with open(path, 'r') as f:
content = f.read()
return content
# Run server
if __name__ == "__main__":
stdio_server(app)
Configuration
// claude_desktop_config.json
{
"mcpServers": {
"filesystem": {
"command": "python",
"args": ["/path/to/filesystem_server.py"],
"env": {
"ROOT_PATH": "/path/to/documents"
}
},
"postgres": {
"command": "mcp-server-postgres",
"args": ["postgresql://localhost/mydb"]
}
}
}
Debug MCP connections
from mcp.inspector import Inspector
inspector = Inspector() inspector.connect("filesystem-server") inspector.list_resources() inspector.test_tool("search", {"query": "test"})
### 2. **MCP Playground**
```python
# Interactive testing environment
from mcp.playground import Playground
playground = Playground()
playground.add_server("filesystem", filesystem_server)
playground.add_server("database", database_server)
playground.start() # Opens web interface
3. Resource Visualizer
import networkx as nx
import matplotlib.pyplot as plt
def visualize_mcp_architecture(servers):
"""Visualize MCP server connections"""
G = nx.DiGraph()
# Add nodes
G.add_node("AI Client", node_type="client")
for server in servers:
G.add_node(server.name, node_type="server")
G.add_edge("AI Client", server.name)
for resource in server.list_resources():
G.add_node(resource.uri, node_type="resource")
G.add_edge(server.name, resource.uri)
# Draw
pos = nx.spring_layout(G)
colors = ['red' if G.nodes[n].get('node_type') == 'client' else
'blue' if G.nodes[n].get('node_type') == 'server' else
'green' for n in G.nodes()]
nx.draw(G, pos, node_color=colors, with_labels=True)
plt.show()
Code Examples
Example 1: Basic MCP Server (Beginner)
from mcp.server import Server
from mcp.types import Resource, TextContent
import json
# Create server
app = Server("simple-server")
# In-memory data store
data_store = {
"users": [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
],
"products": [
{"id": 1, "name": "Widget", "price": 9.99}
]
}
@app.list_resources()
async def list_resources():
"""List available data collections"""
return [
Resource(
uri="data://users",
name="Users",
mimeType="application/json",
description="User database"
),
Resource(
uri="data://products",
name="Products",
mimeType="application/json",
description="Product catalog"
)
]
@app.read_resource()
async def read_resource(uri: str):
"""Read data from collection"""
collection = uri.split("://")[1]
data = data_store.get(collection, [])
return TextContent(
type="text",
text=json.dumps(data, indent=2)
)
@app.list_tools()
async def list_tools():
"""List available tools"""
return [
Tool(
name="add_user",
description="Add a new user",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string"}
},
"required": ["name"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
"""Execute tool"""
if name == "add_user":
new_id = max([u["id"] for u in data_store["users"]]) + 1
new_user = {"id": new_id, "name": arguments["name"]}
data_store["users"].append(new_user)
return f"Added user: {new_user}"
return "Unknown tool"
# Run server
if __name__ == "__main__":
from mcp.server.stdio import stdio_server
stdio_server(app)
Example 2: GitHub Integration (Intermediate)
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
import requests
import os
class GitHubMCPServer:
"""MCP server for GitHub integration"""
def __init__(self, token: str):
self.token = token
self.base_url = "https://api.github.com"
self.headers = {
"Authorization": f"token {token}",
"Accept": "application/vnd.github.v3+json"
}
def create_server(self):
app = Server("github-server")
@app.list_resources()
async def list_resources():
"""List user's repositories"""
response = requests.get(
f"{self.base_url}/user/repos",
headers=self.headers
)
repos = response.json()
return [
Resource(
uri=f"github://repo/{repo['full_name']}",
name=repo['name'],
description=repo['description'] or "No description",
mimeType="application/json"
)
for repo in repos
]
@app.read_resource()
async def read_resource(uri: str):
"""Read repository information"""
repo_name = uri.split("github://repo/")[1]
response = requests.get(
f"{self.base_url}/repos/{repo_name}",
headers=self.headers
)
return TextContent(
type="text",
text=response.text
)
@app.list_tools()
async def list_tools():
return [
Tool(
name="create_issue",
description="Create a GitHub issue",
inputSchema={
"type": "object",
"properties": {
"repo": {"type": "string"},
"title": {"type": "string"},
"body": {"type": "string"}
},
"required": ["repo", "title"]
}
),
Tool(
name="search_code",
description="Search code in repositories",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "create_issue":
response = requests.post(
f"{self.base_url}/repos/{arguments['repo']}/issues",
headers=self.headers,
json={
"title": arguments["title"],
"body": arguments.get("body", "")
}
)
return f"Created issue #{response.json()['number']}"
elif name == "search_code":
response = requests.get(
f"{self.base_url}/search/code",
headers=self.headers,
params={"q": arguments["query"]}
)
results = response.json()
return f"Found {results['total_count']} results"
return app
# Usage
github_server = GitHubMCPServer(token=os.getenv("GITHUB_TOKEN"))
app = github_server.create_server()
if __name__ == "__main__":
from mcp.server.stdio import stdio_server
stdio_server(app)
Example 3: Database MCP Server (Advanced)
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
import psycopg2
import json
class DatabaseMCPServer:
"""MCP server for PostgreSQL database"""
def __init__(self, connection_string: str):
self.conn_string = connection_string
self.conn = psycopg2.connect(connection_string)
def create_server(self):
app = Server("database-server")
@app.list_resources()
async def list_resources():
"""List database tables"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
""")
tables = cursor.fetchall()
return [
Resource(
uri=f"postgres://table/{table[0]}",
name=table[0],
mimeType="application/sql",
description=f"Database table: {table[0]}"
)
for table in tables
]
@app.read_resource()
async def read_resource(uri: str):
"""Read table data"""
table_name = uri.split("postgres://table/")[1]
cursor = self.conn.cursor()
cursor.execute(f"SELECT * FROM {table_name} LIMIT 100")
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
data = [dict(zip(columns, row)) for row in rows]
return TextContent(
type="text",
text=json.dumps(data, indent=2, default=str)
)
@app.list_tools()
async def list_tools():
return [
Tool(
name="query",
description="Execute SQL query",
inputSchema={
"type": "object",
"properties": {
"sql": {"type": "string"}
},
"required": ["sql"]
}
),
Tool(
name="insert",
description="Insert data into table",
inputSchema={
"type": "object",
"properties": {
"table": {"type": "string"},
"data": {"type": "object"}
},
"required": ["table", "data"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
cursor = self.conn.cursor()
if name == "query":
# Only allow SELECT queries for safety
sql = arguments["sql"].strip()
if not sql.upper().startswith("SELECT"):
return "Error: Only SELECT queries allowed"
cursor.execute(sql)
results = cursor.fetchall()
return json.dumps(results, default=str)
elif name == "insert":
table = arguments["table"]
data = arguments["data"]
columns = ", ".join(data.keys())
values = ", ".join(["%s"] * len(data))
sql = f"INSERT INTO {table} ({columns}) VALUES ({values})"
cursor.execute(sql, list(data.values()))
self.conn.commit()
return "Data inserted successfully"
return app
# Usage
db_server = DatabaseMCPServer("postgresql://localhost/mydb")
app = db_server.create_server()
Example 4: Multi-Server Client (Production)
from mcp.client import Client
from mcp.client.stdio import stdio_client
import asyncio
class MCPOrchestrator:
"""Orchestrate multiple MCP servers"""
def __init__(self):
self.servers = {}
async def add_server(self, name: str, command: str, args: list):
"""Add MCP server"""
client = await stdio_client(command, args)
await client.initialize()
self.servers[name] = client
async def list_all_resources(self):
"""List resources from all servers"""
all_resources = {}
for name, client in self.servers.items():
resources = await client.list_resources()
all_resources[name] = resources
return all_resources
async def search_resources(self, query: str):
"""Search across all resources"""
results = []
for name, client in self.servers.items():
resources = await client.list_resources()
for resource in resources:
if query.lower() in resource.name.lower():
results.append({
"server": name,
"resource": resource
})
return results
async def execute_workflow(self, steps: list):
"""Execute multi-step workflow across servers"""
results = []
for step in steps:
server = self.servers[step["server"]]
if step["action"] == "read":
result = await server.read_resource(step["uri"])
elif step["action"] == "tool":
result = await server.call_tool(
step["tool"],
step["arguments"]
)
results.append(result)
return results
# Usage
async def main():
orchestrator = MCPOrchestrator()
# Add servers
await orchestrator.add_server(
"filesystem",
"python",
["filesystem_server.py"]
)
await orchestrator.add_server(
"github",
"python",
["github_server.py"]
)
await orchestrator.add_server(
"database",
"python",
["database_server.py"]
)
# List all resources
resources = await orchestrator.list_all_resources()
print("Available resources:", resources)
# Execute workflow
workflow = [
{
"server": "database",
"action": "read",
"uri": "postgres://table/users"
},
{
"server": "github",
"action": "tool",
"tool": "create_issue",
"arguments": {
"repo": "myorg/myrepo",
"title": "User sync needed",
"body": "Sync users from database"
}
}
]
results = await orchestrator.execute_workflow(workflow)
print("Workflow results:", results)
if __name__ == "__main__":
asyncio.run(main())
Example 5: Custom Tool Server (Expert)
from mcp.server import Server
from mcp.types import Tool, TextContent
import subprocess
import json
class DevToolsMCPServer:
"""MCP server for development tools"""
def create_server(self):
app = Server("devtools-server")
@app.list_tools()
async def list_tools():
return [
Tool(
name="run_tests",
description="Run project tests",
inputSchema={
"type": "object",
"properties": {
"path": {"type": "string"},
"pattern": {"type": "string"}
}
}
),
Tool(
name="lint_code",
description="Lint code files",
inputSchema={
"type": "object",
"properties": {
"files": {
"type": "array",
"items": {"type": "string"}
}
}
}
),
Tool(
name="format_code",
description="Format code with black",
inputSchema={
"type": "object",
"properties": {
"path": {"type": "string"}
}
}
),
Tool(
name="git_status",
description="Get git repository status",
inputSchema={
"type": "object",
"properties": {
"repo_path": {"type": "string"}
}
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "run_tests":
result = subprocess.run(
["pytest", arguments.get("path", "."),
"-k", arguments.get("pattern", "")],
capture_output=True,
text=True
)
return f"Tests: {result.stdout}\n{result.stderr}"
elif name == "lint_code":
files = arguments.get("files", [])
result = subprocess.run(
["pylint"] + files,
capture_output=True,
text=True
)
return f"Lint results:\n{result.stdout}"
elif name == "format_code":
result = subprocess.run(
["black", arguments.get("path", ".")],
capture_output=True,
text=True
)
return f"Formatted: {result.stdout}"
elif name == "git_status":
result = subprocess.run(
["git", "status", "--short"],
cwd=arguments.get("repo_path", "."),
capture_output=True,
text=True
)
return f"Git status:\n{result.stdout}"
return app
# Usage
devtools_server = DevToolsMCPServer()
app = devtools_server.create_server()
TODO: Implement file server with search
class FileServerMCP: def list_resources(self): """List all files in directory""" pass
def read_resource(self, uri):
"""Read file content"""
pass
def search_files(self, query):
"""Search files by content"""
pass
### Exercise 2: API Wrapper
**Goal**: Wrap REST API as MCP server
```python
# TODO: Create MCP server for weather API
class WeatherMCPServer:
def get_current_weather(self, location):
"""Get current weather"""
pass
def get_forecast(self, location, days):
"""Get weather forecast"""
pass
Exercise 3: Multi-Server Integration
Goal: Combine multiple MCP servers
# TODO: Create orchestrator for multiple servers
class MCPHub:
def add_server(self, name, server):
"""Add server to hub"""
pass
def route_request(self, request):
"""Route to appropriate server"""
pass
Exercise 4: Custom Protocol Extension
Goal: Extend MCP with custom features
# TODO: Add streaming support to MCP
class StreamingMCPServer:
def stream_resource(self, uri):
"""Stream large resource"""
pass
Exercise 5: Security Layer
Goal: Add authentication and authorization
# TODO: Implement security for MCP server
class SecureMCPServer:
def authenticate(self, credentials):
"""Verify user credentials"""
pass
def authorize(self, user, resource):
"""Check resource permissions"""
pass
Assignments & Practice
Assignment 1: Personal Knowledge Base
Difficulty: ⭐⭐⭐☆☆
Build MCP server for personal notes:
- Index markdown files
- Full-text search
- Tag management
- Link extraction
- Claude Desktop integration
Deliverables:
- MCP server code
- Configuration file
- Usage documentation
- Demo video
Assignment 2: Development Assistant
Difficulty: ⭐⭐⭐⭐☆
Create MCP server for coding:
- Git integration
- Test runner
- Code formatter
- Documentation generator
- CI/CD integration
Deliverables:
- Complete MCP server
- Tool documentation
- Integration guide
- Example workflows
Assignment 3: Enterprise Integration
Difficulty: ⭐⭐⭐⭐⭐
Build production MCP system:
- Multiple data sources
- Authentication/authorization
- Audit logging
- Rate limiting
- Monitoring/metrics
Deliverables:
- Production-ready code
- Security documentation
- Deployment guide
- Performance benchmarks
Server not connecting
Error: Failed to connect to MCP server
**Solutions**:
```python
# 1. Check server is running
ps aux | grep mcp-server
# 2. Verify configuration
cat ~/.config/claude/claude_desktop_config.json
# 3. Check logs
tail -f ~/.config/claude/logs/mcp-server.log
# 4. Test server directly
python mcp_server.py --test
Issue 2: Resource Not Found
Symptoms:
# Resource URI not resolving
Error: Resource not found: file:///path/to/file
Solutions:
# 1. Validate URI format
def validate_uri(uri):
if not uri.startswith(('file://', 'http://', 'data://')):
raise ValueError(f"Invalid URI: {uri}")
# 2. Check permissions
import os
if not os.access(path, os.R_OK):
raise PermissionError(f"Cannot read: {path}")
# 3. Normalize paths
from pathlib import Path
path = Path(uri.replace("file://", "")).resolve()
Issue 3: Tool Execution Errors
Symptoms:
# Tool fails silently
Tool executed but no result
Solutions:
# 1. Add error handling
@app.call_tool()
async def call_tool(name: str, arguments: dict):
try:
result = execute_tool(name, arguments)
return result
except Exception as e:
return f"Error: {str(e)}"
# 2. Validate arguments
def validate_arguments(schema, arguments):
# Check required fields
for field in schema.get("required", []):
if field not in arguments:
raise ValueError(f"Missing required field: {field}")
# 3. Add logging
import logging
logging.info(f"Executing tool: {name} with {arguments}")
Hack 1: Resource caching
class CachedMCPServer: def init(self): self.cache = {}
async def read_resource(self, uri):
if uri in self.cache:
return self.cache[uri]
content = await fetch_resource(uri)
self.cache[uri] = content
return content
Hack 2: Lazy loading
class LazyMCPServer: async def list_resources(self): # Return metadata only, load content on demand return [ Resource(uri=uri, name=name, lazy=True) for uri, name in self.get_resource_list() ]
Hack 3: Resource composition
async def compose_resources(uris): """Combine multiple resources""" contents = [] for uri in uris: content = await read_resource(uri) contents.append(content) return "\n\n".join(contents)
Hack 4: Tool chaining
async def chain_tools(tools): """Execute tools in sequence""" result = None for tool in tools: result = await call_tool( tool["name"], {**tool["args"], "input": result} ) return result
Hack 5: Dynamic server discovery
class ServerRegistry: def discover_servers(self): """Auto-discover MCP servers""" servers = [] for path in os.listdir("/usr/local/bin"): if path.startswith("mcp-server-"): servers.append(path) return servers
---
<!--REF_OPEN:Related Concepts-->
### Prerequisites
- Agents - MCP enables agent tool use
- [RAG](02-Intermediate--12-RAG.html) - MCP provides data access
- [LLMs](01-Foundations--06-Large-Language-Models.html) - MCP connects LLMs to data
### Related Topics
- [Vector Database](02-Intermediate--11-Vector-Database.html) - Store MCP resources
- Microservices - Similar architectural patterns
- GraphQL - Alternative data access pattern
---
<!--REF_CLOSE-->
<!--REF_OPEN:Resources-->
### 📚 Official Documentation
1. **MCP Specification**
- https://spec.modelcontextprotocol.io/
2. **MCP SDK Documentation**
- https://github.com/anthropics/mcp
3. **Claude Desktop Integration**
- https://claude.ai/docs/mcp
### 🎥 Video Tutorials
1. **"Introducing Model Context Protocol"** - Anthropic
- https://www.youtube.com/watch?v=MCP_intro
2. **"Building MCP Servers"** - Tutorial Series
- https://www.youtube.com/playlist?list=MCP_tutorials
### 🛠️ Tools & Libraries
1. **MCP Python SDK**
- https://github.com/anthropics/mcp-python
2. **MCP TypeScript SDK**
- https://github.com/anthropics/mcp-typescript
3. **Community MCP Servers**
- https://github.com/topics/mcp-server
### 📖 Tutorials & Guides
1. **Getting Started with MCP**
- https://docs.anthropic.com/mcp/quickstart
2. **Building Custom MCP Servers**
- https://docs.anthropic.com/mcp/custom-servers
3. **MCP Best Practices**
- https://docs.anthropic.com/mcp/best-practices
---
5. **What is MCP? - Daily Dose of DS** #MyTopic
- https://www.dailydoseofds.com/what-is-mcp/
- Excellent beginner-friendly explanation
- Real-world use cases and examples
- Recommended by community
<!--REF_CLOSE-->
## Progress Tracking
- [ ] Understand MCP architecture
- [ ] Install MCP SDK
- [ ] Create basic MCP server
- [ ] Integrate with Claude Desktop
- [ ] Build custom tools
- [ ] Connect to external APIs
- [ ] Implement security
- [ ] Deploy production server
---
<div style="text-align:center; margin-top:30px; padding:20px; background:linear-gradient(135deg, #667eea 0%, #764ba2 100%); border-radius:10px; color:white;">
<h2>🔌 Master Model Context Protocol!</h2>
<p>Connect AI to any data source with the universal protocol</p>
</div>