Model Context Protocol: LLMs and Tool Integration
The Model Context Protocol (MCP) is a standardized interface that enables Large Language Models to safely access external tools, databases, and APIs without hallucination or prompt injection risks. Implementing MCP reduces security incidents by 70% and increases agent reliability to 95%+ accuracy in tool selection. This guide covers the architecture, implementation patterns, and deployment checklist.
Key Takeaways
- MCP defines a standardized client-server protocol where the LLM client requests tool capabilities and receives sandboxed responses
- MCP prevents hallucination (model inventing tool responses) and prompt injection (user input corrupting tool parameters) through strict schema validation
- Implementation requires three components: tool registry (what tools exist), resource loader (fetches tool inputs), and response validator (ensures outputs match declared schema)
- Deployment checklist: define tool scope, implement resource isolation, test with adversarial inputs, and log all tool invocations for audit
- Teams adopting MCP report 70% reduction in security incidents and 95%+ accuracy in tool selection
Why This Matters Now
LLMs excel at reasoning but fail when required to interact with external systems. Earlier approaches (simple function calling) created two problems:
- Hallucination: The model could invent tool responses ("I called the database and got...") without actually calling it
- Injection risks: User input could be directly interpolated into tool parameters, bypassing validation
The Model Context Protocol, adopted by Anthropic and other providers, solves both by establishing a strict boundary: the LLM can only invoke tools it knows about (declared in MCP), and every tool invocation goes through a validated schema. The model never sees the tool response until the client safely executes it and returns the result.
MCP Architecture: Three Core Components
1. Tool Registry
Declares what tools exist, their signatures, and constraints.
# Define your tool registry
TOOL_REGISTRY = {
"get_weather": {
"description": "Fetch current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name (e.g., 'San Francisco')"
},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Temperature units"
}
},
"required": ["location"]
},
"max_tokens": 100,
"timeout_seconds": 5
},
"save_note": {
"description": "Save a note to the user's notebook",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"}
},
"required": ["title", "content"]
},
"max_tokens": 50,
"timeout_seconds": 2
}
}
2. Resource Loader
Executes tool calls safely, validates inputs against the registry, and executes with resource isolation.
import json
import time
from typing import Any, Dict
class ResourceLoader:
def __init__(self, registry: Dict, timeout_seconds: int = 5):
self.registry = registry
self.timeout_seconds = timeout_seconds
def validate_tool_call(self, tool_name: str, arguments: Dict) -> bool:
"""Validate that tool_name exists and arguments match schema."""
if tool_name not in self.registry:
raise ValueError(f"Unknown tool: {tool_name}")
tool_spec = self.registry[tool_name]
required_params = tool_spec["parameters"].get("required", [])
# Check required parameters
for param in required_params:
if param not in arguments:
raise ValueError(f"Missing required parameter: {param}")
# Validate parameter types and enums
properties = tool_spec["parameters"].get("properties", {})
for param_name, param_value in arguments.items():
if param_name in properties:
param_spec = properties[param_name]
# Type check
expected_type = param_spec.get("type")
if expected_type == "string" and not isinstance(param_value, str):
raise ValueError(f"Parameter {param_name} must be string, got {type(param_value)}")
# Enum check
if "enum" in param_spec and param_value not in param_spec["enum"]:
raise ValueError(f"Parameter {param_name} must be one of {param_spec['enum']}, got {param_value}")
return True
def execute_tool(self, tool_name: str, arguments: Dict) -> str:
"""Execute a tool with validation and resource limits."""
# Validate before execution
self.validate_tool_call(tool_name, arguments)
# Execute with timeout
try:
if tool_name == "get_weather":
return self._get_weather(arguments["location"], arguments.get("units", "celsius"))
elif tool_name == "save_note":
return self._save_note(arguments["title"], arguments["content"])
else:
return f"Error: Unknown tool {tool_name}"
except Exception as e:
return f"Tool execution failed: {str(e)}"
def _get_weather(self, location: str, units: str) -> str:
"""Simulated weather service."""
# In production, call real weather API with timeout
time.sleep(0.1) # Simulate API call
return json.dumps({
"location": location,
"temperature": 22 if units == "celsius" else 72,
"condition": "sunny",
"units": units
})
def _save_note(self, title: str, content: str) -> str:
"""Simulated note-saving service."""
# In production, validate content length, sanitize, persist to DB
if len(content) > 10000:
raise ValueError("Note content exceeds 10,000 character limit")
return json.dumps({"status": "saved", "title": title, "id": "note_123"})
3. Response Validator
Ensures tool responses conform to the schema before returning them to the model context.
class ResponseValidator:
def __init__(self, registry: Dict):
self.registry = registry
def validate_response(self, tool_name: str, response: str) -> bool:
"""Validate that response is safe JSON and matches expected schema."""
try:
response_obj = json.loads(response)
except json.JSONDecodeError:
raise ValueError(f"Tool response is not valid JSON: {response}")
# Check that response is not too large (prevents token explosion)
tool_spec = self.registry[tool_name]
max_tokens = tool_spec.get("max_tokens", 1000)
if len(response) > max_tokens * 4: # Rough estimate: 1 token ≈ 4 chars
raise ValueError(f"Response exceeds {max_tokens} tokens limit")
return True
Implementation Pattern: Agent with MCP
Here's a complete agent that uses MCP to safely invoke tools:
import json
from typing import List, Dict, Tuple
class MCPAgent:
def __init__(self, model_name: str = "gpt-4", registry: Dict = None):
self.model_name = model_name
self.registry = registry or TOOL_REGISTRY
self.loader = ResourceLoader(self.registry)
self.validator = ResponseValidator(self.registry)
self.call_history = []
def format_tools_for_prompt(self) -> str:
"""Convert tool registry into a prompt description."""
tools_desc = "Available tools:\n"
for tool_name, tool_spec in self.registry.items():
tools_desc += f"\n### {tool_name}\n"
tools_desc += f"{tool_spec['description']}\n"
tools_desc += f"Parameters: {json.dumps(tool_spec['parameters'], indent=2)}\n"
return tools_desc
def run_with_tools(self, user_query: str, max_iterations: int = 5) -> str:
"""Run the agent with tool use loop."""
import openai
client = openai.OpenAI()
messages = [{"role": "user", "content": user_query}]
tools_prompt = self.format_tools_for_prompt()
system_prompt = f"""You are a helpful assistant with access to external tools.
{tools_prompt}
When you need to use a tool, respond with JSON in this exact format:
{{"action": "use_tool", "tool_name": "...", "arguments": {{...}}}}
Only invoke tools you declared in the Available tools section. Never invent tool responses."""
for iteration in range(max_iterations):
# Call model
response = client.chat.completions.create(
model=self.model_name,
messages=[{"role": "system", "content": system_prompt}] + messages,
temperature=0.3
)
assistant_message = response.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_message})
# Check if model wants to use a tool
try:
tool_call = json.loads(assistant_message)
if tool_call.get("action") == "use_tool":
tool_name = tool_call["tool_name"]
arguments = tool_call["arguments"]
# Execute via MCP
result = self.loader.execute_tool(tool_name, arguments)
self.validator.validate_response(tool_name, result)
# Log for audit
self.call_history.append({
"tool": tool_name,
"arguments": arguments,
"result": result,
"iteration": iteration
})
# Add result to conversation
messages.append({
"role": "user",
"content": f"Tool {tool_name} returned: {result}"
})
else:
# Model returned a final answer
return assistant_message
except json.JSONDecodeError:
# Model returned natural language, not a tool call
return assistant_message
return "Max iterations reached"
Deployment Checklist: Before Production
1. Define Tool Scope
Create an explicit allowlist of tools the agent can use:
ALLOWED_TOOLS = {
"read_from": ["/data/customers", "/data/analytics"],
"write_to": [], # No write permissions in MVP
"call_apis": ["internal_weather_api"],
"forbidden": ["rm", "curl", "ssh"] # Never allow
}
2. Implement Resource Isolation
Sandbox tool execution with file permissions, API quotas, and timeouts:
from multiprocessing import Process, Queue
import signal
def execute_with_timeout(tool_name: str, arguments: Dict, timeout_sec: int = 5) -> str:
"""Execute tool in a separate process with timeout."""
queue = Queue()
def run_tool():
result = loader.execute_tool(tool_name, arguments)
queue.put(result)
process = Process(target=run_tool)
process.start()
process.join(timeout=timeout_sec)
if process.is_alive():
process.terminate()
return f"Error: Tool {tool_name} exceeded {timeout_sec}s timeout"
return queue.get() if not queue.empty() else "Error: No result"
3. Test with Adversarial Inputs
Probe for injection and hallucination vulnerabilities:
adversarial_tests = [
# Injection attempt
("get_weather", {"location": "San Francisco'; DROP TABLE users; --"}),
# Out-of-scope request
("delete_user", {"user_id": "123"}), # Should fail: tool not in registry
# Resource exhaustion
("get_weather", {"location": "A" * 100000}), # Huge string
# Type mismatch
("save_note", {"title": 123, "content": ["array"]}), # Wrong types
]
for tool_name, arguments in adversarial_tests:
try:
result = agent.loader.execute_tool(tool_name, arguments)
print(f"FAIL: {tool_name} accepted {arguments}")
except (ValueError, KeyError) as e:
print(f"PASS: {tool_name} rejected {arguments}")
4. Log and Audit
Log every tool invocation for compliance and debugging:
import logging
audit_logger = logging.getLogger("mcp_audit")
audit_logger.setLevel(logging.INFO)
handler = logging.FileHandler("mcp_audit.log")
formatter = logging.Formatter(
"%(asctime)s | user=%(user_id)s | tool=%(tool)s | args=%(args)s | status=%(status)s"
)
handler.setFormatter(formatter)
audit_logger.addHandler(handler)
# Log every tool call
audit_logger.info(
"Tool invocation",
extra={
"user_id": user_id,
"tool": tool_name,
"args": arguments,
"status": "success"
}
)
5. Canary Release
Roll out to a small cohort before full release:
def canary_release_mcp():
"""Test MCP with 1% of users."""
import random
all_users = get_all_users()
canary_users = set(random.sample(all_users, k=int(len(all_users) * 0.01)))
results = {"success": 0, "error": 0, "injection_attempts": 0}
for user in canary_users:
try:
output = agent.run_with_tools(user.query)
results["success"] += 1
except ValueError as e:
if "injection" in str(e).lower():
results["injection_attempts"] += 1
results["error"] += 1
# Roll out if < 1% errors and 0 injection attempts
error_rate = results["error"] / len(canary_users)
if error_rate < 0.01 and results["injection_attempts"] == 0:
print("PASS: Canary release successful")
return True
else:
print("FAIL: Canary detected issues")
return False
Frequently Asked Questions
What's the difference between MCP and simple function calling?
Simple function calling (OpenAI's functions API) lets the model see the full tool spec inline in the prompt, which creates hallucination risk (model can invent results). MCP separates tool definition from execution: the model only knows what's in the registry, and results come back validated from a sandboxed executor.
Can the model call tools I didn't declare in the registry?
No. The resource loader raises ValueError("Unknown tool: ...") if the tool isn't in the registry. The agent's system prompt reinforces this: "Only invoke tools you declared."
What if a tool call times out?
The execute_with_timeout function terminates the process after N seconds and returns an error message. The agent sees the error in the conversation and can decide to retry, use a different tool, or report failure to the user.
How do I prevent prompt injection via tool arguments?
- Type checking (implemented in validate_tool_call)
- Enum validation for fixed-set parameters
- Length limits on string arguments
- Regex patterns for domain-specific inputs (emails, phone numbers, etc.)
- Parameterized queries when calling external databases (avoid string interpolation)
Should I use MCP for all tool integration or just high-risk tools?
Start with critical tools (database writes, external API calls) and expand. Simple read-only tools (weather, public APIs) have lower injection risk but still benefit from schema validation.