Loading…
Imagine you have multiple AI assistants, each specialized in different tasks - one for coding, another for data analysis, a third for creative writing, and so on. How do they share information? How do they maintain context across conversations? How do they access the same tools and data sources consistently?
This is where the Model Context Protocol (MCP) comes in - a revolutionary open standard that enables seamless communication and context sharing between AI applications, data sources, and tools. Think of MCP as the "universal language" that allows different AI systems to work together harmoniously, much like how HTTP enables web browsers to communicate with web servers.
In this lesson, we'll explore how MCP is transforming the landscape of agentic AI by providing a standardized way for agents to access resources, maintain context, and collaborate effectively.
By the end of this comprehensive lesson, you will be able to:
Before MCP, building AI applications that needed to access external data or tools faced several challenges:
Model Context Protocol (MCP) is an open standard that provides:
MCP consists of three main components that work together to enable seamless AI communication:
What they are: Applications or AI systems that request resources and use tools.
Key Characteristics:
Examples:
# Example MCP Client
class MCPClient:
def __init__(self):
self.connection = None
self.context = ConversationContext()
async def connect_to_server(self, server_url):
self.connection = await MCPConnection.connect(server_url)
async def request_resource(self, resource_uri):
return await self.connection.get_resource(resource_uri)
async def use_tool(self, tool_name, arguments):
return await self.connection.call_tool(tool_name, arguments)
What they are: Programs that provide access to specific resources, tools, or data sources.
Key Characteristics:
Examples:
# Example MCP Server
class MCPServer:
def __init__(self):
self.resources = {}
self.tools = {}
self.auth_manager = AuthManager()
def register_resource(self, uri, handler):
self.resources[uri] = handler
def register_tool(self, name, handler):
self.tools[name] = handler
async def handle_request(self, request):
# Authenticate request
if not await self.auth_manager.authenticate(request):
return ErrorResponse("Unauthorized")
# Route to appropriate handler
if request.type == "resource":
return await self.handle_resource_request(request)
elif request.type == "tool":
return await self.handle_tool_request(request)
What they are: Runtime environments that manage connections between clients and servers.
Key Characteristics:
Examples:
┌─────────────┐ 1. Connect ┌─────────────┐ 2. Request ┌─────────────┐
│ Client │ ───────────────► │ Host │ ───────────────► │ Server │
│ │ │ │ │ │
│ - AI App │ │ - Router │ │ - Database │
│ - Interface │ │ - Monitor │ │ - API │
│ - Context │ │ - Logger │ │ - Tools │
└─────────────┘ └─────────────┘ └─────────────┘
▲ │ │
│ 4. Response │ 3. Process │
│ ◄────────────────────────── │ ◄────────────────────────── │
│ │ │
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ User │ │ Result │ │ Data │
│ Interaction │ │ Processing │ │ Retrieval │
└─────────────┘ └─────────────┘ └─────────────┘
MCP uses JSON-RPC 2.0 as its base protocol, with specific message types for different operations:
{
"jsonrpc": "2.0",
"id": "req_123",
"method": "resources/read",
"params": {
"uri": "database://users/123"
}
}
{
"jsonrpc": "2.0",
"id": "req_123",
"result": {
"contents": [
{
"uri": "database://users/123",
"mimeType": "application/json",
"text": "{\"id\": 123, \"name\": \"John Doe\", \"email\": \"[email protected]\"}"
}
]
}
}
{
"jsonrpc": "2.0",
"id": "req_123",
"error": {
"code": -32602,
"message": "Invalid params",
"data": "Resource not found: database://users/123"
}
}
resources/list: List available resources
{
"method": "resources/list",
"params": {}
}
resources/read: Read a specific resource
{
"method": "resources/read",
"params": {
"uri": "file:///home/user/document.txt"
}
}
tools/list: List available tools
{
"method": "tools/list",
"params": {}
}
tools/call: Execute a tool
{
"method": "tools/call",
"params": {
"name": "database_query",
"arguments": {
"query": "SELECT * FROM users WHERE active = true",
"limit": 100
}
}
}
prompts/list: List available prompts
{
"method": "prompts/list",
"params": {}
}
prompts/get: Get a specific prompt template
{
"method": "prompts/get",
"params": {
"name": "code_review",
"arguments": {
"language": "python",
"focus": "security"
}
}
}
MCP supports various resource types through URI schemes:
file:///path/to/file.txt
file://hostname/path/to/file.txt
database://connection/table/row
postgresql://user:pass@host:port/database/table
mysql://user:pass@host:port/database/table
https://api.example.com/users/123
api://service/resource/identifier
custom://namespace/resource/identifier
mcp://server/resource/path
An MCP server follows a structured pattern for exposing resources and tools. The server acts as a bridge between AI applications and underlying data sources or services.
Key Server Components:
Server Lifecycle:
MCP clients are responsible for initiating connections and consuming resources from servers. They maintain session state and handle the communication protocol.
Key Client Components:
Client Operations Flow:
# Minimal MCP server example
from mcp.server import Server
server = Server("my-server")
@server.list_resources()
async def list_resources():
return [Resource(uri="data://example", name="Example Data")]
@server.call_tool()
async def handle_tool(name, args):
if name == "process":
return [TextContent(text="Processed successfully")]
# Minimal MCP client example
from mcp.client.session import ClientSession
async def use_mcp_server():
async with ClientSession() as session:
await session.initialize()
resources = await session.list_resources()
result = await session.call_tool("process", {"data": "example"})
Resource Types and Access Patterns:
Access Control Strategies:
Tool Categories:
Input Validation Patterns:
Error Classification:
Recovery Mechanisms:
MCP provides several authentication methods to ensure secure communication:
{
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "my-client",
"version": "1.0.0"
},
"auth": {
"type": "token",
"token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}
}
}
{
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "my-client",
"version": "1.0.0"
},
"auth": {
"type": "api_key",
"key": "sk-1234567890abcdef"
}
}
}
{
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "my-client",
"version": "1.0.0"
},
"auth": {
"type": "oauth",
"access_token": "ya29.a0AfH6SMC...",
"refresh_token": "1//0g...",
"expires_in": 3600
}
}
}
MCP servers can implement fine-grained access control:
class AuthManager:
def __init__(self):
self.permissions = {
"user_123": {
"resources": ["file://workspace/*", "database://users/*"],
"tools": ["calculator", "database_query"],
"actions": ["read", "write"]
},
"admin_456": {
"resources": ["*"],
"tools": ["*"],
"actions": ["*"]
}
}
async def authenticate(self, request):
"""Authenticate the request"""
auth_header = request.headers.get("Authorization")
if not auth_header:
return False
# Validate token/API key
token = auth_header.replace("Bearer ", "")
user_id = await self.validate_token(token)
return user_id is not None
async def authorize(self, user_id, resource, action):
"""Check if user has permission for resource/action"""
if user_id not in self.permissions:
return False
user_perms = self.permissions[user_id]
# Check resource access
if not self._match_resource(user_perms["resources"], resource):
return False
# Check action permission
if action not in user_perms["actions"] and "*" not in user_perms["actions"]:
return False
return True
def _match_resource(self, allowed_resources, requested_resource):
"""Check if requested resource matches allowed patterns"""
for allowed in allowed_resources:
if allowed == "*":
return True
if allowed.endswith("*"):
prefix = allowed[:-1]
if requested_resource.startswith(prefix):
return True
if allowed == requested_resource:
return True
return False
import logging
from datetime import datetime
class SecurityLogger:
def __init__(self):
self.logger = logging.getLogger("mcp_security")
handler = logging.FileHandler("mcp_security.log")
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_access(self, user_id, resource, action, success):
"""Log access attempts"""
self.logger.info(
f"User: {user_id}, Resource: {resource}, "
f"Action: {action}, Success: {success}"
)
def log_authentication(self, user_id, success, ip_address):
"""Log authentication attempts"""
status = "SUCCESS" if success else "FAILED"
self.logger.warning(
f"Auth {status} - User: {user_id}, IP: {ip_address}"
)
import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests=100, time_window=60):
self.max_requests = max_requests
self.time_window = time_window
self.requests = defaultdict(list)
async def check_rate_limit(self, client_id):
"""Check if client has exceeded rate limit"""
now = time.time()
client_requests = self.requests[client_id]
# Remove old requests outside time window
client_requests[:] = [
req_time for req_time in client_requests
if now - req_time < self.time_window
]
# Check if under limit
if len(client_requests) >= self.max_requests:
return False
# Add current request
client_requests.append(now)
return True
AI coding assistants leverage MCP to provide comprehensive development support by integrating with various development tools and resources.
Key Capabilities:
Resource Examples:
git://repository/status - Current repository state and pending changesbuild://project/dependencies - Project dependencies and version informationfile://src/**/*.py - Source code files with metadatatest://results/latest - Most recent test execution resultsTool Examples:
run_tests - Execute test suites with configurable parameterslint_code - Perform code quality checks and automatic fixesbuild_project - Compile and package applicationsdeploy_code - Deploy applications to various environmentsBenefits for Developers:
Data analysis platforms use MCP to provide unified access to diverse data sources and analytical tools.
Data Source Integration:
Analytical Capabilities:
Resource Organization:
postgresql://analytics/{table} - Structured analytics datamongodb://analytics/{collection} - Document-based analytics dataredis://cache/{key} - Cached computation resultss3://data/{dataset} - Large dataset filesTool Categories:
run_query - Execute database queries with optimizationgenerate_report - Create formatted analytical reportsdata_transform - Apply transformations to datasetsexport_data - Export results in various formatsMulti-agent systems use MCP to enable sophisticated collaboration between specialized AI agents.
Collaboration Patterns:
Agent Specializations:
Collaboration Tools:
assign_task - Delegate tasks with priority and deadline managementshare_context - Distribute relevant context to participating agentsrequest_help - Escalate problems requiring specialized expertisecoordinate_results - Merge and synchronize agent outputsCommunication Protocols:
Benefits for Complex Problem Solving:
For enterprise deployments, a gateway pattern provides centralized management of multiple MCP servers, enabling scalable and maintainable architectures.
Gateway Components:
Routing Logic:
Benefits:
Implementing caching significantly improves MCP performance by reducing redundant data access and computation.
Caching Strategies:
Cache Invalidation:
Performance Benefits:
Event-driven architecture enables loose coupling between MCP components and supports scalable, real-time coordination.
Event Types:
Event Processing:
Coordination Benefits:
Testing Framework Components:
Test Categories:
Key Test Scenarios:
Integration Test Strategies:
Test Environment Setup:
Integration Test Types:
Debugging Components:
Debugging Techniques:
Debugging Tools Features:
# Good resource URI design
resources = [
Resource(
uri="api://v1/users/123/profile",
name="User Profile",
description="User profile information",
mimeType="application/json",
metadata={
"version": "1.0",
"last_modified": "2024-01-15T10:30:00Z",
"size": 1024,
"etag": "abc123"
}
)
]
@server.call_tool()
async def robust_tool_handler(name: str, arguments: dict):
"""Robust tool handler with validation"""
try:
# Validate required parameters
required_params = ["query", "limit"]
for param in required_params:
if param not in arguments:
raise ValueError(f"Missing required parameter: {param}")
# Validate parameter types and ranges
query = arguments["query"]
if not isinstance(query, str) or len(query.strip()) == 0:
raise ValueError("Query must be a non-empty string")
limit = arguments["limit"]
if not isinstance(limit, int) or limit < 1 or limit > 1000:
raise ValueError("Limit must be an integer between 1 and 1000")
# Execute tool logic
result = await execute_query(query, limit)
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except ValueError as e:
# Handle validation errors
return [TextContent(
type="text",
text=f"Validation Error: {str(e)}"
)]
except Exception as e:
# Handle unexpected errors
logger.error(f"Unexpected error in {name}: {e}")
return [TextContent(
type="text",
text=f"Internal Error: Unable to process request"
)]
# BAD - Blocking operation in async context
@server.call_tool()
async def slow_tool(name: str, arguments: dict):
time.sleep(10) # Blocks the event loop
return "Done"
# GOOD - Non-blocking operation
@server.call_tool()
async def fast_tool(name: str, arguments: dict):
await asyncio.sleep(10) # Doesn't block the event loop
return "Done"
# BAD - Accumulating data without cleanup
class BadServer:
def __init__(self):
self.cache = {} # Never cleared
async def handle_request(self, request):
self.cache[request.id] = request.data # Grows indefinitely
# GOOD - Implementing cleanup
class GoodServer:
def __init__(self):
self.cache = {}
self.max_cache_size = 1000
async def handle_request(self, request):
# Implement cache eviction
if len(self.cache) >= self.max_cache_size:
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[request.id] = request.data
# BAD - Swallowing exceptions
@server.call_tool()
async def bad_tool(name: str, arguments: dict):
try:
return await risky_operation()
except:
return "Error occurred" # No information about what went wrong
# GOOD - Proper error handling
@server.call_tool()
async def good_tool(name: str, arguments: dict):
try:
return await risky_operation()
except DatabaseError as e:
logger.error(f"Database error in {name}: {e}")
return [TextContent(
type="text",
text=f"Database Error: {str(e)}"
)]
except ValidationError as e:
logger.warning(f"Validation error in {name}: {e}")
return [TextContent(
type="text",
text=f"Validation Error: {str(e)}"
)]
except Exception as e:
logger.error(f"Unexpected error in {name}: {e}")
raise
Question 1: What is the primary purpose of the Model Context Protocol (MCP)? A) To provide a standardized way for AI applications to access resources and tools B) To replace all existing AI frameworks C) To create a new programming language for AI D) To optimize neural network performance
Question 2: Which of the following is NOT a core component of MCP? A) MCP Clients B) MCP Servers C) MCP Hosts D) MCP Routers
Question 3: How does MCP handle authentication? A) Only supports token-based authentication B) Supports multiple authentication methods including tokens, API keys, and OAuth C) Doesn't provide authentication mechanisms D) Requires biometric authentication
Question 4: What is the role of an MCP Host? A) To create AI models B) To manage connections between clients and servers C) To store data permanently D) To train machine learning algorithms
Exercise 1: Design an MCP server for a weather service that provides:
Outline the resources, tools, and authentication mechanisms you would implement.
Exercise 2: Create a simple MCP client that connects to a file system server and:
Exercise 3: Design a security model for an enterprise MCP deployment that includes:
Build a Complete MCP Application: Create a task management system using MCP with the following requirements:
MCP Server that provides:
MCP Client that:
Additional Features:
Deliverables:
You've gained a comprehensive understanding of the Model Context Protocol and how it enables seamless communication between AI systems!
In the next lesson, "Agent Architecture Patterns", we'll explore:
This knowledge will build upon your understanding of MCP to help you design and implement robust agent architectures that can leverage the power of standardized communication protocols.
| Term | Definition |
|---|---|
| MCP Client | Application or AI system that requests resources and uses tools |
| MCP Server | Program that provides access to specific resources, tools, or data sources |
| MCP Host | Runtime environment that manages connections between clients and servers |
| Resource | Data source accessible through MCP (files, database tables, API endpoints) |
| Tool | Executable function or operation provided by an MCP server |
| URI | Uniform Resource Identifier used to identify MCP resources |
| JSON-RPC | Remote procedure call protocol encoded in JSON |
| Authentication | Process of verifying the identity of a client or server |
| Authorization | Process of determining if an authenticated entity has permission to access a resource |
| Context | Information maintained across multiple interactions or operations |
The Model Context Protocol represents a fundamental shift in how AI systems interact with data and tools. Master MCP, and you'll be equipped to build the next generation of interconnected, collaborative AI applications!