Your Agent (Claude / Bedrock) ↓ MCP Client (asks: what tools are available?) ↓ MCP Server (returns: tool schemas + executes calls) ↓ Your actual APIs, databases, AWS services Your Agent (Claude / Bedrock) ↓ MCP Client (asks: what tools are available?) ↓ MCP Server (returns: tool schemas + executes calls) ↓ Your actual APIs, databases, AWS services Your Agent (Claude / Bedrock) ↓ MCP Client (asks: what tools are available?) ↓ MCP Server (returns: tool schemas + executes calls) ↓ Your actual APIs, databases, AWS services pip install mcp boto3 fastmcp pip install mcp boto3 fastmcp pip install mcp boto3 fastmcp 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() 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() 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() # Add to Claude Code's MCP servers claude mcp add aws-tools -- python /path/to/aws_mcp_server.py # Add to Claude Code's MCP servers claude mcp add aws-tools -- python /path/to/aws_mcp_server.py # Add to Claude Code's MCP servers claude mcp add aws-tools -- python /path/to/aws_mcp_server.py How many files are in my logs-bucket-prod S3 bucket? How many files are in my logs-bucket-prod S3 bucket? How many files are in my logs-bucket-prod S3 bucket? 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"] 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"] 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"] fastmcp==0.9.0 boto3==1.35.0 fastmcp==0.9.0 boto3==1.35.0 fastmcp==0.9.0 boto3==1.35.0 # 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 # 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 # 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 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"]) 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"]) 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"]) 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) 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) 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) # 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. """ # 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. """ # 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. """ 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"}) 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"}) 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"}) - query_dynamodb — lets Claude query a DynamoDB table using natural language - get_s3_summary — lets Claude list and summarize files in an S3 bucket - 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