Building Dynamic AI Systems with MCP
MCP gives a model a standardized bridge to live data and executable functions — the basis for a modular, tool-augmented system rather than a closed text generator. The best way to internalize that architecture is to build a miniature of it. This reference implements the protocol’s core ideas in plain Python: the data structures, a server, a client, and the async patterns that hold it together.
The code below is a teaching model, deliberately built from first principles to expose the moving parts. It is not the production wire protocol — for that, and for the roles this mirrors, see MCP Foundations and Architecture.
The three data structures
Structured information flows between an AI system and its environment through three types:
- Resource — external data or a document (a file, a database record).
- Tool — an executable capability the model can invoke (an API call, an analysis function).
- Message — one unit of communication; a sequence of them forms the interaction history.
from dataclasses import dataclass, asdict
from typing import Dict, List, Any, Optional, Callable
from datetime import datetime
@dataclass
class Resource:
uri: str
name: str
description: str
mime_type: str
content: Any = None
@dataclass
class Tool:
name: str
description: str
parameters: Dict[str, Any]
handler: Optional[Callable] = None
@dataclass
class Message:
role: str
content: str
timestamp: str = None
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
Architecture
Two components carry the system: a server that registers and executes resources and tools, and a client that connects to servers, queries them, and keeps a running context of every interaction. That local context is what makes the exchange stateful — the client remembers what it fetched and called.
The server
import asyncio
class MCPServer:
def __init__(self, name: str):
self.name = name
self.resources: Dict[str, Resource] = {}
self.tools: Dict[str, Tool] = {}
self.capabilities = {"resources": True, "tools": True, "prompts": True, "logging": True}
def register_resource(self, resource: Resource) -> None:
self.resources[resource.uri] = resource
def register_tool(self, tool: Tool) -> None:
self.tools[tool.name] = tool
async def get_resource(self, uri: str) -> Optional[Resource]:
await asyncio.sleep(0.1) # Simulate I/O
return self.resources.get(uri)
async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
if tool_name not in self.tools:
raise ValueError(f"Tool '{tool_name}' not found")
tool = self.tools[tool_name]
if tool.handler:
return await tool.handler(**arguments)
return {"status": "executed", "tool": tool_name, "args": arguments}
def list_resources(self) -> List[Dict[str, str]]:
return [{"uri": r.uri, "name": r.name, "description": r.description} for r in self.resources.values()]
def list_tools(self) -> List[Dict[str, Any]]:
return [{"name": t.name, "description": t.description, "parameters": t.parameters} for t in self.tools.values()]
The list_resources and list_tools methods are the discovery surface — the equivalent of resources/list and tools/list on a real server.
The client
class MCPClient:
def __init__(self, client_id: str):
self.client_id = client_id
self.connected_servers: Dict[str, MCPServer] = {}
self.context: List[Message] = []
def connect_server(self, server: MCPServer) -> None:
self.connected_servers[server.name] = server
async def query_resources(self, server_name: str) -> List[Dict[str, str]]:
if server_name not in self.connected_servers:
raise ValueError(f"Not connected to server: {server_name}")
return self.connected_servers[server_name].list_resources()
async def fetch_resource(self, server_name: str, uri: str) -> Optional[Resource]:
server = self.connected_servers[server_name]
resource = await server.get_resource(uri)
if resource:
self.add_to_context(Message(role="system", content=f"Fetched resource: {resource.name}"))
return resource
async def call_tool(self, server_name: str, tool_name: str, **kwargs) -> Any:
server = self.connected_servers[server_name]
result = await server.execute_tool(tool_name, kwargs)
self.add_to_context(Message(role="system", content=f"Tool '{tool_name}' executed"))
return result
def add_to_context(self, message: Message) -> None:
self.context.append(message)
def get_context(self) -> List[Dict[str, Any]]:
return [asdict(msg) for msg in self.context]
Every fetch and call appends a Message, so get_context() yields a complete, replayable record of the session.
Example tool handlers
Async functions stand in for real external operations and get registered as tools:
import random
async def analyze_sentiment(text: str) -> Dict[str, Any]:
await asyncio.sleep(0.2)
sentiments = ["positive", "negative", "neutral"]
return {"text": text, "sentiment": random.choice(sentiments), "confidence": round(random.uniform(0.7, 0.99), 2)}
async def summarize_text(text: str, max_length: int = 100) -> Dict[str, str]:
await asyncio.sleep(0.15)
summary = text[:max_length] + "..." if len(text) > max_length else text
return {"original_length": len(text), "summary": summary}
async def search_knowledge(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
await asyncio.sleep(0.25)
mock_results = [{"title": f"Result {i+1} for '{query}'", "score": round(random.uniform(0.5, 1.0), 2)} for i in range(top_k)]
return sorted(mock_results, key=lambda x: x["score"], reverse=True)
Putting it together
A full run exercises the protocol end to end:
- Initialize an
MCPServer. - Register resources (documents) and tools (
analyze_sentimentand friends). - Connect an
MCPClientto the server. - Interact — list resources, fetch a specific one, and call tools with arguments.
- Inspect context — the client’s history now holds a stateful record of everything that happened.
Why the design holds up
- Modularity — resources and tools plug in independently, so the system extends without rewrites.
- External context — resources feed structured, outside data into the model on demand.
- Dynamic action — tools let the model do things, not just describe them.
- Async by default — the
asynccore overlaps I/O, which is what makes the pattern scale to systems talking to many APIs and data sources at once.
