Model Context Protocol: Composable Tools
The Model Context Protocol (MCP) is a standard for building tool ecosystems that agents can easily compose and reuse. Instead of hardcoding tools into each agent, MCP allows you to define tools once (as standardized server implementations) and attach them to any agent. An MCP server exposes resources (data) and tools (functions) via a standard interface. An agent (or human using an AI assistant) connects to MCP servers and gains access to all their capabilities. This article teaches you to build MCP servers, integrate them with Python agents, and create scalable tool ecosystems.
MCP solves a critical problem: tool duplication. If you build five agents, each might implement a search_database tool slightly differently. With MCP, one standardized database_server serves all agents, ensuring consistency and reducing maintenance burden.
Understanding MCP Architecture
MCP has three components:
- Client — An AI agent or human assistant that wants to use tools
- Server — A standalone process exposing tools and resources via the MCP protocol
- Protocol — A JSON-RPC standard for client-server communication
Here's the flow:
Agent (Client)
│
├─ Connects to MCP Server A (Weather tools)
├─ Connects to MCP Server B (Database tools)
└─ Connects to MCP Server C (Math tools)
Each agent sees all servers' tools as if they were local.
MCP servers communicate over stdio, HTTP, or WebSocket. The most common setup is stdio: the agent spawns a server process and talks to it via standard input/output.
Building an MCP Server
Here's a minimal MCP server that exposes a calculate_stats tool:
import json
import sys
from typing import Any
class MCPServer:
"""Simple MCP server exposing tools."""
def __init__(self):
self.tools = [
{
"name": "calculate_stats",
"description": "Calculate statistics on a list of numbers",
"input_schema": {
"type": "object",
"properties": {
"numbers": {
"type": "array",
"items": {"type": "number"},
"description": "List of numbers to analyze"
},
"stat": {
"type": "string",
"enum": ["mean", "median", "stddev", "min", "max"],
"description": "Which statistic to compute"
}
},
"required": ["numbers", "stat"]
}
},
{
"name": "list_available_stats",
"description": "List all available statistics functions",
"input_schema": {
"type": "object",
"properties": {},
"required": []
}
}
]
def list_tools(self) -> list:
"""Return list of available tools."""
return self.tools
def call_tool(self, tool_name: str, arguments: dict) -> Any:
"""Execute a tool."""
if tool_name == "calculate_stats":
numbers = arguments.get("numbers", [])
stat = arguments.get("stat")
import statistics
if stat == "mean":
return statistics.mean(numbers)
elif stat == "median":
return statistics.median(numbers)
elif stat == "stddev":
return statistics.stdev(numbers) if len(numbers) > 1 else 0
elif stat == "min":
return min(numbers)
elif stat == "max":
return max(numbers)
elif tool_name == "list_available_stats":
return ["mean", "median", "stddev", "min", "max"]
return None
def handle_request(self, request: dict) -> dict:
"""Handle an MCP request."""
method = request.get("method")
params = request.get("params", {})
if method == "tools/list":
return {"tools": self.list_tools()}
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
result = self.call_tool(tool_name, arguments)
return {"result": result}
return {"error": "Unknown method"}
def run(self):
"""Main server loop: read requests from stdin, write responses to stdout."""
for line in sys.stdin:
try:
request = json.loads(line)
response = self.handle_request(request)
# Add message ID if present
if "id" in request:
response["id"] = request["id"]
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {"error": str(e)}
print(json.dumps(error_response))
sys.stdout.flush()
if __name__ == "__main__":
server = MCPServer()
server.run()
To use this server, save it as stats_server.py and run it:
python stats_server.py
Connecting an Agent to an MCP Server
Now create an agent that connects to this MCP server:
import subprocess
import json
import anthropic
class MCPAgentClient:
"""Agent that uses MCP servers."""
def __init__(self, server_command: str):
"""
Args:
server_command: Command to start the MCP server, e.g., "python stats_server.py"
"""
self.server_process = subprocess.Popen(
server_command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
self.request_id = 0
self.client = anthropic.Anthropic()
def send_request(self, method: str, params: dict = None) -> dict:
"""Send a request to the MCP server."""
self.request_id += 1
request = {
"id": self.request_id,
"method": method,
"params": params or {}
}
# Write to server
self.server_process.stdin.write(json.dumps(request) + "\n")
# Read response
response_line = self.server_process.stdout.readline()
return json.loads(response_line)
def get_tools(self) -> list:
"""Get list of tools from MCP server."""
response = self.send_request("tools/list")
return response.get("tools", [])
def call_tool(self, tool_name: str, arguments: dict) -> Any:
"""Call a tool on the MCP server."""
response = self.send_request("tools/call", {
"name": tool_name,
"arguments": arguments
})
return response.get("result")
def run_agent(self, user_message: str) -> str:
"""Run the agent with MCP-provided tools."""
tools = self.get_tools()
messages = [{"role": "user", "content": user_message}]
for iteration in range(10):
response = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
tools=tools,
messages=messages
)
if response.stop_reason == "end_turn":
return next(
(block.text for block in response.content if hasattr(block, 'text')),
"Finished"
)
# Handle tool calls
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = self.call_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result) if not isinstance(result, str) else result
})
messages.append({"role": "user", "content": tool_results})
return "Max iterations"
def cleanup(self):
"""Shut down the MCP server."""
self.server_process.terminate()
# Usage
agent_client = MCPAgentClient("python stats_server.py")
try:
result = agent_client.run_agent("Calculate the mean of [1, 2, 3, 4, 5]")
print(result)
finally:
agent_client.cleanup()
Composing Multiple MCP Servers
Agents often need tools from multiple domains. MCP makes this seamless:
class MultiMCPAgent:
"""Agent that uses multiple MCP servers."""
def __init__(self, servers: dict):
"""
Args:
servers: Dict of {server_name: server_command}
"""
self.server_clients = {}
for name, command in servers.items():
self.server_clients[name] = MCPAgentClient(command)
def get_all_tools(self) -> list:
"""Gather tools from all MCP servers."""
all_tools = []
for name, client in self.server_clients.items():
tools = client.get_tools()
for tool in tools:
# Add server prefix to tool names to avoid collisions
tool["name"] = f"{name}:{tool['name']}"
all_tools.append(tool)
return all_tools
def call_tool(self, tool_name: str, arguments: dict) -> Any:
"""Route tool call to the appropriate MCP server."""
server_name, actual_tool_name = tool_name.split(":", 1)
if server_name not in self.server_clients:
raise ValueError(f"Unknown server: {server_name}")
client = self.server_clients[server_name]
return client.call_tool(actual_tool_name, arguments)
def run_agent(self, user_message: str) -> str:
"""Run agent with all available tools."""
tools = self.get_all_tools()
messages = [{"role": "user", "content": user_message}]
client = anthropic.Anthropic()
for iteration in range(10):
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
tools=tools,
messages=messages
)
if response.stop_reason == "end_turn":
return next(
(block.text for block in response.content if hasattr(block, 'text')),
"Finished"
)
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = self.call_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result) if not isinstance(result, str) else result
})
messages.append({"role": "user", "content": tool_results})
return "Max iterations"
def cleanup(self):
"""Shut down all MCP servers."""
for client in self.server_clients.values():
client.cleanup()
# Usage: compose stats server with a weather server
agent = MultiMCPAgent({
"stats": "python stats_server.py",
"weather": "python weather_server.py"
})
result = agent.run_agent("Calculate stats for [1, 2, 3] and get weather for NYC")
agent.cleanup()
Key Takeaways
- MCP is a standard protocol for composing tools across multiple server implementations
- An MCP server exposes tools via a standard interface; agents connect and use all exposed tools
- Building reusable MCP servers eliminates tool duplication across agents
- Multiple MCP servers can be combined, with tool names prefixed to avoid collisions
- MCP enables scalable tool ecosystems where teams share standardized tool implementations
Frequently Asked Questions
How do I handle long-running tools in MCP servers?
Use async I/O in the server. For expensive operations, return a job ID and let the agent poll for results. Or stream results back to the agent incrementally.
Can MCP servers talk to each other?
Yes, but be careful of cycles (Server A calls Server B which calls Server A). For orchestration, have the agent coordinate, not the servers.
Is MCP secure?
By default, MCP assumes the client and server are on the same machine. For remote MCP servers, use authentication and TLS. Never expose an MCP server on the public internet without security.
Can I use MCP with OpenAI's API?
MCP is API-agnostic. Implement the MCP-to-OpenAI adapter in the agent client, similar to how we did it for Claude.