MCP (Model Context Protocol) is the most important AI infrastructure pattern of 2026. Anthropic built it, the Linux Foundation now owns it, and AWS just made it a first-class citizen in Bedrock AgentCore.
97 million SDK downloads. 13,000+ servers built by the community. And as of this month, AWS is deploying them as managed services inside your existing cloud infrastructure.
This is the tutorial I wish existed when I started. Not theory. Actual working code that deploys a real MCP server connected to AWS services in under 30 minutes.
What MCP Actually Is (In One Paragraph)
MCP is the protocol that lets AI agents use external tools reliably. Without it, your agent either hardcodes tool integrations (brittle, unmaintainable) or hallucinates function calls that don't exist.
With MCP, you define tools once as a server. Any agent — Claude, Cursor, your own custom Bedrock agent — can discover and use those tools via a standardized interface. Think USB-C for AI tools. Build once, plug in anywhere.
Your Agent (Claude / Bedrock)
↓
MCP Client (asks: what tools are available?)
↓
MCP Server (returns: tool schemas + executes calls)
↓
Your actual APIs, databases, AWS services
What We're Building
A working MCP server that exposes two AWS tools:
query_dynamodb— lets Claude query a DynamoDB table using natural language
get_s3_summary— lets Claude list and summarize files in an S3 bucket
Then we'll connect it to a Bedrock agent and watch Claude use both tools autonomously.
Prerequisites: Python 3.11+, AWS credentials configured, boto3 installed.
Step 1 — Install the MCP SDK
pip install mcp boto3 fastmcp
FastMCP is the Python framework that makes building MCP servers significantly less painful than raw MCP. It handles the protocol layer so you write tools, not JSON-RPC boilerplate.
Step 2 — Build the MCP Server
Create aws_mcp_server.py:
import boto3
import json
from fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("AWS Tools Server")
# AWS clients
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
s3_client = boto3.client("s3", region_name="us-east-1")
@mcp.tool()
def query_dynamodb(table_name: str, key_name: str, key_value: str) -> str:
"""
Query a DynamoDB table by primary key.
Use this when the user wants to look up specific records from a database.
Args:
table_name: The DynamoDB table to query
key_name: The primary key attribute name
key_value: The value to look up
"""
try:
table = dynamodb.Table(table_name)
response = table.get_item(
Key={key_name: key_value}
)
item = response.get("Item")
if not item:
return json.dumps({
"found": False,
"message": f"No record found for {key_name}={key_value} in {table_name}"
})
return json.dumps({
"found": True,
"table": table_name,
"record": item
}, default=str)
except Exception as e:
return json.dumps({"error": str(e), "table": table_name})
@mcp.tool()
def get_s3_summary(bucket_name: str, prefix: str = "") -> str:
"""
List and summarize files in an S3 bucket.
Use this when the user asks about files, documents, or data stored in S3.
Args:
bucket_name: The S3 bucket to inspect
prefix: Optional folder prefix to filter results (e.g., 'reports/' or 'data/2026/')
"""
try:
paginator = s3_client.get_paginator("list_objects_v2")
pages = paginator.paginate(
Bucket=bucket_name,
Prefix=prefix,
PaginationConfig={"MaxItems": 50}
)
files = []
total_size = 0
for page in pages:
for obj in page.get("Contents", []):
files.append({
"key": obj["Key"],
"size_kb": round(obj["Size"] / 1024, 2),
"last_modified": obj["LastModified"].isoformat()
})
total_size += obj["Size"]
return json.dumps({
"bucket": bucket_name,
"prefix": prefix or "(root)",
"file_count": len(files),
"total_size_kb": round(total_size / 1024, 2),
"files": files[:20], # Return first 20 for context window efficiency
"note": f"Showing {min(20, len(files))} of {len(files)} files"
}, default=str)
except Exception as e:
return json.dumps({"error": str(e), "bucket": bucket_name})
if __name__ == "__main__":
# Run as stdio MCP server (for local testing with Claude Desktop / Claude Code)
mcp.run()
The @mcp.tool() decorator does the heavy lifting — it generates the JSON schema from your Python type hints and docstring. Claude uses the docstring to decide when to call each tool. Write it from Claude's perspective: "Use this when the user wants to..."
Step 3 — Test Locally With Claude Code
Before deploying to Bedrock, test the MCP server locally. Add it to your Claude Code MCP config:
# Add to Claude Code's MCP servers
claude mcp add aws-tools -- python /path/to/aws_mcp_server.py
Restart Claude Code, then ask:
How many files are in my logs-bucket-prod S3 bucket?
You should see Claude invoke get_s3_summary automatically. If it works locally, it'll work on Bedrock.
Step 4 — Deploy to Bedrock AgentCore Runtime
AWS Bedrock AgentCore Runtime lets you deploy MCP servers as managed services — serverless, auto-scaling, with session isolation handled for you. This is the new way to run MCP in production.
4a — Create a Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY aws_mcp_server.py .
# AgentCore expects MCP servers to run on port 8080
EXPOSE 8080
# Run as HTTP MCP server for AgentCore
CMD ["python", "aws_mcp_server.py", "--transport", "http", "--port", "8080"]
requirements.txt:
fastmcp==0.9.0
boto3==1.35.0
4b — Push to ECR
# Create ECR repo
aws ecr create-repository --repository-name aws-tools-mcp-server
# Get ECR login
aws ecr get-login-password --region us-east-1 | \
docker login --username AWS \
--password-stdin 123456789.dkr.ecr.us-east-1.amazonaws.com
# Build and push
docker build -t aws-tools-mcp-server .
docker tag aws-tools-mcp-server:latest \
123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest
docker push 123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest
4c — Deploy to AgentCore Runtime
import boto3
agentcore = boto3.client("bedrock-agentcore", region_name="us-east-1")
response = agentcore.create_agent_runtime(
agentRuntimeName="aws-tools-mcp",
agentRuntimeArtifact={
"containerConfiguration": {
"containerUri": "123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest"
}
},
networkConfiguration={
"networkMode": "PUBLIC"
}
)
print("AgentCore Runtime ARN:", response["agentRuntimeArn"])
print("Endpoint:", response["agentRuntimeEndpoint"])
AgentCore handles: session isolation (each user gets their own MCP session in a dedicated microVM), automatic scaling, authentication, and 8-hour maximum session support for long-running operations.
Step 5 — Connect to a Bedrock Agent
Now wire the deployed MCP server into a Bedrock agent:
import boto3
import json
bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")
# Your MCP server endpoint from Step 4c
MCP_ENDPOINT = "https://your-agentcore-endpoint.bedrock-agentcore.us-east-1.amazonaws.com"
def run_agent_with_mcp(user_message: str) -> str:
"""
Bedrock agent that uses your deployed MCP server as its tool provider.
"""
messages = [
{"role": "user", "content": [{"text": user_message}]}
]
system = [{
"text": f"""You are an AWS assistant with access to DynamoDB and S3 tools.
Use your tools to answer questions about the user's AWS data.
Always use tools to get real data — never guess or make up values.
MCP Server: {MCP_ENDPOINT}"""
}]
# Tool config pointing to your MCP server
tool_config = {
"tools": [{
"toolSpec": {
"name": "query_dynamodb",
"description": "Query a DynamoDB table by primary key",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"table_name": {"type": "string"},
"key_name": {"type": "string"},
"key_value": {"type": "string"}
},
"required": ["table_name", "key_name", "key_value"]
}
}
}
}, {
"toolSpec": {
"name": "get_s3_summary",
"description": "List and summarize files in an S3 bucket",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"bucket_name": {"type": "string"},
"prefix": {"type": "string"}
},
"required": ["bucket_name"]
}
}
}
}]
}
# Agentic loop
for _ in range(10):
response = bedrock_runtime.converse(
modelId="anthropic.claude-3-sonnet-20240229-v1:0",
system=system,
messages=messages,
toolConfig=tool_config
)
stop_reason = response["stopReason"]
output = response["output"]["message"]
messages.append(output)
if stop_reason == "end_turn":
for block in output["content"]:
if "text" in block:
return block["text"]
elif stop_reason == "tool_use":
tool_results = []
for block in output["content"]:
if "toolUse" not in block:
continue
tool = block["toolUse"]
tool_name = tool["name"]
tool_input = tool["input"]
# Route to correct tool
if tool_name == "query_dynamodb":
result = query_dynamodb(**tool_input)
elif tool_name == "get_s3_summary":
result = get_s3_summary(**tool_input)
else:
result = json.dumps({"error": f"Unknown tool: {tool_name}"})
tool_results.append({
"toolResult": {
"toolUseId": tool["toolUseId"],
"content": [{"text": result}]
}
})
messages.append({"role": "user", "content": tool_results})
return "Agent reached iteration limit"
# Test it
response = run_agent_with_mcp(
"How many files are in my data-lake-prod bucket? "
"And look up user ID 'user_123' in my Users table."
)
print(response)
The Three Things That Break MCP in Production
1 — Vague tool descriptions
Claude decides which tool to call based entirely on the description field. If your description is vague, Claude either skips the tool or calls it when it shouldn't.
# Weak — Claude might not call this when it should
@mcp.tool()
def get_data(table: str, key: str) -> str:
"""Get data from a table."""
# Strong — Claude knows exactly when to use this
@mcp.tool()
def query_dynamodb(table_name: str, key_name: str, key_value: str) -> str:
"""
Query a DynamoDB table by primary key.
Use this when the user wants to look up specific records,
find customer data, retrieve order information, or access
any structured data stored in DynamoDB.
"""
2 — Returning too much data
MCP tool results flow back into the agent's context window. A tool that returns 10,000 rows from DynamoDB will burn your context window in one call.
Always limit your returns: paginate aggressively, return summaries not raw dumps, use limit parameters in every database query.
3 — Not handling tool failures
If your tool raises an exception, the entire agentic loop breaks. Every tool should return structured JSON even on failure:
try:
result = do_the_thing()
return json.dumps({"success": True, "data": result})
except Exception as e:
return json.dumps({"success": False, "error": str(e), "tool": "tool_name"})
The agent can read the error message and respond gracefully. An uncaught exception just crashes.
Stateful MCP: AWS's New Feature (March 2026)
AWS just shipped stateful MCP server support in AgentCore Runtime. This is significant.
Previously, each MCP call was stateless — the server had no memory between tool invocations. Now, each user session gets a dedicated microVM with session context preserved using an Mcp-Session-Id header.
This enables:
Elicitation — the server can ask the user follow-up questions mid-tool execution
Sampling — the server can request Claude to generate content as part of a tool's operation
Progress notifications — real-time updates for long-running operations
For long-running tasks (ML training jobs, large data exports, multi-hour simulations), this changes the architecture completely. The server can now maintain state across a multi-hour operation without requiring the agent session to stay open.
What to Build Next
With this foundation working, the natural next steps:
Add more AWS tools — CloudWatch metrics querier, RDS query executor, Lambda invoker, Step Functions status checker. Each becomes a tool your agent can use.
Add Bedrock Guardrails — wrap your MCP server with content filtering and PII detection that operates outside the agent's reasoning loop.
Multi-agent coordination — one coordinator agent that routes requests to specialist subagents, each with their own focused MCP server and tool set.
AgentCore Gateway — instead of embedding tool schemas in your agent, register your MCP server with AgentCore Gateway and let multiple agents discover the same tools via a central registry.
The full production version of this architecture — agentic loops, multi-agent systems, MCP server builds, Bedrock Guardrails, CloudWatch observability — is covered hands-on in the CCA-001: Claude Certified Architect track. Real Bedrock sandboxes, automated validation, no AWS account needed.
If you want to build the architecture in this article in a real environment without worrying about AWS billing, that's the path.
👉 cloudedventures.com/labs/track/claude-certified-architect-cca-001
What are you building with MCP? Drop a comment — especially if you hit a specific architecture problem I can help with.
SOCIAL SHARE CARD GENERATOR