MCP Protocol¶
Overview¶
The Model Context Protocol (MCP) is the standard way for agents to discover, access, and use tools.
It solves a fundamental problem: how do we let agents safely use any tool or resource?
What is MCP?¶
The Problem MCP Solves¶
# Without MCP (chaotic):
agent = Agent()
agent.tools = [
GoogleSearchAPI(),
GitHubAPI(),
SlackAPI(),
CustomDatabase(),
FileSystem(),
#... each with different interfaces
]
# Each tool has different error handling, auth, rate limiting
# Agent must know about all of them
# Hard to add new tools
# Difficult to sandbox
# With MCP (standardized):
mcp_server = MCPServer()
mcp_server.register_resource('web_search', GoogleSearchAPI())
mcp_server.register_resource('github', GitHubAPI())
mcp_server.register_resource('slack', SlackAPI())
agent.connect_to_mcp(mcp_server)
# Agent discovers tools automatically
# Unified interface for all resources
# Easy to sandbox and control
-
MCP Architecture¶
Request/Response Pattern¶
class MCPServer:
"""MCP Server that agents connect to"""
def __init__(self):
self.resources = {}
self.tools = {}
def register_tool(self, name: str, tool):
"""Register a tool with the server"""
tool_definition = {
'name': name,
'description': tool.description,
'inputSchema': {
'type': 'object',
'properties': tool.input_params,
'required': tool.required_params
}
}
self.tools[name] = tool_definition
def handle_call_tool(self, name: str, arguments: dict):
"""Execute tool call"""
if name not in self.tools:
return {'error': f'Tool {name} not found'}
tool = self.tools[name]['impl']
try:
result = tool(**arguments)
return {'result': result, 'success': True}
except Exception as e:
return {'error': str(e), 'success': False}
-
Tool Discovery¶
Automatic Tool Advertisement¶
class MCPClient:
"""Agent side: MCP Client"""
def discover_tools(self):
"""Ask MCP server what tools are available"""
# Send request to MCP server
request = {
'jsonrpc': '2.0',
'id': 1,
'method': 'tools/list',
'params': {}
}
response = self.send_request(request)
# Server responds with all available tools
available_tools = response['result']['tools']
# Agent now knows what it can do
for tool in available_tools:
self.register_available_tool(tool)
return available_tools
def use_tool(self, tool_name: str, args: dict):
"""Call a tool through MCP"""
request = {
'jsonrpc': '2.0',
'id': 2,
'method': 'tools/call',
'params': {
'name': tool_name,
'arguments': args
}
}
response = self.send_request(request)
return response['result']
Real-World Integrations¶
Google Drive + MCP¶
class GoogleDriveMCP:
"""MCP integration for Google Drive"""
def __init__(self, credentials):
self.drive = GoogleDriveService(credentials)
def register_with_mcp(self, server):
"""Register all Drive operations"""
server.register_tool('list_files', self.list_files)
server.register_tool('read_file', self.read_file)
server.register_tool('write_file', self.write_file)
server.register_tool('delete_file', self.delete_file)
server.register_tool('search_files', self.search_files)
def list_files(self, folder_id: str = None):
"""List files in folder"""
results = self.drive.files().list(
q=f"'{folder_id}' in parents" if folder_id else None
).execute()
return results['files']
Security Model¶
Sandboxing Tools¶
class SecureMCPServer:
"""MCP server with security"""
def __init__(self):
self.permissions = {} # User → allowed tools
def authorize_tool_call(self, user_id: str, tool_name: str):
"""Check if user can call tool"""
if user_id not in self.permissions:
return False
allowed_tools = self.permissions[user_id]
return tool_name in allowed_tools
def handle_call_with_auth(self, user_id: str, tool_name: str, args: dict):
"""Secure tool execution"""
# Check permission
if not self.authorize_tool_call(user_id, tool_name):
return {'error': 'Permission denied'}
# Rate limit
if self.exceeded_rate_limit(user_id, tool_name):
return {'error': 'Rate limit exceeded'}
# Execute
result = self.execute_tool(tool_name, args)
# Audit log
self.log_access(user_id, tool_name, result)
return result
3 Warnings¶
Warning 1: Tool Explosion¶
# WRONG
# Register 100 tools with agent
for tool in all_available_tools:
register(tool)
# Agent overwhelmed, can't choose
# Tokens wasted listing tools
# RIGHT
# Register only relevant tools
relevant_tools = filter_tools_by_capability(user_capability)
for tool in relevant_tools:
register(tool)
# Agent focused on what it needs
Warning 2: No Rate Limiting¶
# WRONG
agent = Agent()
agent.can_call_tools()
# No rate limiting!
# Agent hammers API
# RIGHT
secure_server = SecureMCPServer()
secure_server.set_rate_limit('user_1', 'search', 100_per_hour)
agent.connect_to_secure_server(secure_server)
Warning 3: Leaking Tool Results¶
# WRONG
result = agent.call_tool()
return result # Full output to client
# Might expose sensitive data
# RIGHT
result = agent.call_tool()
sanitized = sanitize_output(result)
return sanitized
-
Last Updated: August 9, 2026