Intermediate
Understanding MCP: Architecture and Concepts
Understanding MCP: Architecture and Concepts
The Model Context Protocol (MCP) is a standardized architecture for connecting language models to external data sources and tools. Unlike ad-hoc integrations, MCP provides a systematic way to extend AI capabilities while maintaining security and standardization. This lesson explores MCP’s architecture, core concepts, and how it enables seamless integration.
Why Model Context Protocol
Traditional approaches to extending AI capabilities are fragmented:
- Tight Coupling: Custom integration code for each data source
- Maintenance Burden: Changes in external systems require code updates
- Security Challenges: Managing credentials and access control scattered across codebase
- Limited Reusability: Integration code specific to one application
MCP solves these problems by providing:
- Standard Interface: Consistent protocol for all integrations
- Decoupled Architecture: Client and server operate independently
- Composable Resources: Combine multiple data sources seamlessly
- Security First: Built-in authentication and authorization mechanisms
MCP Architecture Overview
MCP follows a client-server architecture where clients (like Claude) communicate with servers (exposing resources and tools):
┌─────────────────────────────────┐
│ Client (Claude) │
│ - Sends requests to servers │
│ - Processes responses │
│ - Orchestrates multi-server │
│ interactions │
└──────────────┬──────────────────┘
│
┌──────┴──────┐
│ MCP │
│ Protocol │
└──────┬──────┘
│
┌────────┼────────┐
│ │ │
┌─────▼──┐ ┌──▼────┐ ┌─▼──────┐
│Server 1│ │Server2│ │Server 3│
│ (DB) │ │(API) │ │(Files) │
└────────┘ └───────┘ └────────┘Core Components
Transport Layer: Handles communication between client and server. Supports:
- Standard input/output (stdio)
- Server-sent events (SSE) over HTTP
- WebSockets for real-time communication
Request-Response Model: Clients send requests, servers respond:
{
"jsonrpc": "2.0",
"id": 1,
"method": "resources/list",
"params": {}
}Resources: Named, addressable pieces of data or capabilities:
- Documents
- Database records
- API endpoints
- File paths
Tools: Functions that servers expose for client invocation.
Prompts: Pre-written instructions or templates that servers provide.
MCP Communication Protocol
JSON-RPC 2.0 Foundation
MCP uses JSON-RPC 2.0 for request-response communication:
import json
from typing import Dict, Any
class MCPMessage:
"""Represents an MCP protocol message."""
def __init__(self, method: str, params: Dict[str, Any] = None, msg_id: int = 1):
self.method = method
self.params = params or {}
self.id = msg_id
def to_json(self) -> str:
"""Convert to JSON-RPC format."""
return json.dumps({
"jsonrpc": "2.0",
"id": self.id,
"method": self.method,
"params": self.params
})
# Example: Request to list available resources
list_resources_msg = MCPMessage("resources/list")
print(list_resources_msg.to_json())
# Output: {"jsonrpc": "2.0", "id": 1, "method": "resources/list", "params": {}}Initialization Handshake
Clients and servers exchange capabilities during initialization:
class ServerCapabilities:
"""Define what a server can provide."""
def __init__(self):
self.capabilities = {
"resources": {
"subscribe": False # Can client subscribe to resource updates?
},
"tools": {},
"prompts": {}
}
def to_dict(self) -> Dict:
return {
"serverInfo": {
"name": "example-server",
"version": "1.0.0"
},
"capabilities": self.capabilities
}
# Server initialization response
server_init = ServerCapabilities()
print(json.dumps(server_init.to_dict(), indent=2))Resources: Data Access Layer
Resources represent accessible data or information that servers provide:
Resource Definition
@dataclass
class Resource:
"""Represents a resource provided by server."""
uri: str # Unique identifier, e.g., "sqlite:///database.db/users"
name: str
description: str
resource_type: str = "text/plain"
class ResourceServer:
"""Server that provides resources."""
def __init__(self):
self.resources = [
Resource(
uri="sqlite:///company.db/employees",
name="Employee Database",
description="Access employee records and information"
),
Resource(
uri="file:///documents/policies.md",
name="Company Policies",
description="Employee handbook and policies"
),
Resource(
uri="https://api.company.com/announcements",
name="Company Announcements",
description="Latest company announcements and news"
)
]
def list_resources(self) -> list:
"""Return available resources."""
return [{
"uri": r.uri,
"name": r.name,
"description": r.description,
"mimeType": r.resource_type
} for r in self.resources]
def read_resource(self, uri: str) -> str:
"""Read content of a resource."""
for resource in self.resources:
if resource.uri == uri:
# In real implementation, fetch actual data
return f"Content of {resource.name}"
raise ValueError(f"Resource not found: {uri}")Resource URIs and Hierarchy
Resources use URI format for global addressing:
def parse_resource_uri(uri: str) -> Dict[str, str]:
"""Parse MCP resource URI."""
# Format: scheme://authority/path
# Examples:
# - sqlite:///database.db/users
# - file:///home/user/documents/report.pdf
# - http://api.example.com/v1/data
import urllib.parse
parsed = urllib.parse.urlparse(uri)
return {
"scheme": parsed.scheme, # e.g., "sqlite", "file", "http"
"authority": parsed.netloc, # e.g., "", "api.example.com"
"path": parsed.path, # e.g., "/database.db/users"
"query": parsed.query
}
# Examples
print(parse_resource_uri("sqlite:///database.db/users"))
# {'scheme': 'sqlite', 'authority': '', 'path': '/database.db/users', 'query': ''}
print(parse_resource_uri("http://api.example.com/v1/data?limit=10"))
# {'scheme': 'http', 'authority': 'api.example.com', 'path': '/v1/data', 'query': 'limit=10'}Tools: Function Invocation
Tools allow clients to invoke server functions:
Tool Definition
class Tool:
"""Defines a tool available on server."""
def __init__(self, name: str, description: str, input_schema: Dict):
self.name = name
self.description = description
self.input_schema = input_schema
def to_dict(self) -> Dict:
return {
"name": self.name,
"description": self.description,
"inputSchema": self.input_schema
}
class ToolServer:
"""Server exposing tools."""
def __init__(self):
self.tools = [
Tool(
name="search_documents",
description="Search through company documents",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"limit": {
"type": "integer",
"description": "Maximum results"
}
},
"required": ["query"]
}
),
Tool(
name="create_issue",
description="Create a ticket in issue tracker",
input_schema={
"type": "object",
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"]
}
},
"required": ["title", "description"]
}
)
]
def list_tools(self) -> list:
"""Return available tools."""
return [t.to_dict() for t in self.tools]
def call_tool(self, name: str, arguments: Dict) -> str:
"""Execute a tool."""
if name == "search_documents":
query = arguments.get("query")
limit = arguments.get("limit", 10)
return f"Found documents matching '{query}' (limited to {limit})"
elif name == "create_issue":
title = arguments.get("title")
return f"Created issue: {title}"
else:
raise ValueError(f"Unknown tool: {name}")Tool Invocation Flow
class MCPClient:
"""Client that uses MCP to invoke tools."""
def __init__(self, server: ToolServer):
self.server = server
def discover_tools(self) -> list:
"""Discover what tools server provides."""
return self.server.list_tools()
def invoke_tool(self, name: str, arguments: Dict) -> str:
"""Invoke a tool on the server."""
# Validate tool exists
tools = {t["name"]: t for t in self.discover_tools()}
if name not in tools:
raise ValueError(f"Tool not found: {name}")
tool = tools[name]
# Validate arguments match schema
# (In production, use jsonschema library)
# Invoke tool
return self.server.call_tool(name, arguments)
# Usage
server = ToolServer()
client = MCPClient(server)
# Discover tools
print("Available tools:")
for tool in client.discover_tools():
print(f" - {tool['name']}: {tool['description']}")
# Invoke a tool
result = client.invoke_tool("search_documents", {"query": "budget", "limit": 5})
print(result)Prompts: Pre-Written Instructions
Prompts allow servers to provide templates or instructions:
class Prompt:
"""Server-provided prompt template."""
def __init__(self, name: str, description: str, template: str):
self.name = name
self.description = description
self.template = template
self.arguments = []
def to_dict(self) -> Dict:
return {
"name": self.name,
"description": self.description
}
class PromptServer:
"""Server providing prompt templates."""
def __init__(self):
self.prompts = [
Prompt(
name="code_review",
description="Template for code review guidelines",
template="""Review this code for:
1. Correctness and logic
2. Performance considerations
3. Security vulnerabilities
4. Code style and readability
Code to review:
{code}"""
),
Prompt(
name="bug_report",
description="Template for comprehensive bug reports",
template="""Please provide:
- Steps to reproduce
- Expected behavior
- Actual behavior
- System information
Report: {description}"""
)
]
def list_prompts(self) -> list:
"""List available prompts."""
return [p.to_dict() for p in self.prompts]
def get_prompt(self, name: str, arguments: Dict) -> str:
"""Get rendered prompt."""
for prompt in self.prompts:
if prompt.name == name:
return prompt.template.format(**arguments)
raise ValueError(f"Prompt not found: {name}")Transport Mechanisms
Stdio Transport
Simple transport using standard input/output:
import sys
import json
class StdioMCPServer:
"""MCP server using stdio."""
def __init__(self):
self.running = True
def run(self):
"""Read requests from stdin, send responses to stdout."""
while self.running:
try:
line = sys.stdin.readline()
if not line:
break
request = json.loads(line)
response = self.handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {
"code": -32603,
"message": str(e)
}
}
print(json.dumps(error_response))
sys.stdout.flush()
def handle_request(self, request: Dict) -> Dict:
"""Handle an MCP request."""
method = request.get("method")
params = request.get("params", {})
msg_id = request.get("id")
if method == "resources/list":
return self._list_resources(msg_id)
elif method == "tools/list":
return self._list_tools(msg_id)
else:
return {
"jsonrpc": "2.0",
"id": msg_id,
"error": {"code": -32601, "message": "Method not found"}
}
def _list_resources(self, msg_id: int) -> Dict:
"""Handle resources/list request."""
return {
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"resources": [
{
"uri": "memory:///notes",
"name": "Notes",
"description": "User notes"
}
]
}
}
def _list_tools(self, msg_id: int) -> Dict:
"""Handle tools/list request."""
return {
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"tools": []
}
}Best Practices
1. Resource Organization
Group related resources with meaningful URI hierarchies:
database:///company/employees
database:///company/departments
database:///company/projects2. Clear Tool Specifications
Define tools with precise input schemas and descriptions.
3. Error Handling
Return proper JSON-RPC error responses with meaningful messages.
Key Takeaway
MCP provides a standardized, composable architecture for connecting language models to external systems. Understanding resources, tools, and transports enables building scalable, maintainable AI integrations without tight coupling or scattered integration logic.
Exercises
-
Build a Resource Server: Create an MCP server that exposes multiple resources (files, database records, API endpoints).
-
Tool Implementation: Implement a tool server with 3-5 tools that perform different operations.
-
Stdio Transport: Write a complete stdio-based MCP server that handles multiple request types.
-
Resource URIs: Design a comprehensive URI hierarchy for a fictional company’s resources.
-
Integration: Integrate your server with an MCP-compatible client and test resource access and tool invocation.