📅 Day 21⏱ 60 min🔥 Ascend☁️ AWS

AWS Services MCP
S3 · DynamoDB · Lambda

The real power of MCP in an AWS shop: give Claude the ability to list buckets, query DynamoDB tables, invoke Lambda functions, and manage your cloud — all through natural language, with IAM least-privilege and audit logging baked in.

The difference between a demo and a production AWS MCP server is three things: IAM roles, error handling, and secrets rotation. This day covers all three with working boto3 code you can deploy today.
📋 Today's topics
🏗️ Architecture

AWS MCP Architecture & IAM Design

The golden rule: your MCP server should only have the IAM permissions it actually uses. Never use AdministratorAccess or PowerUserAccess. Create a dedicated IAM role for each MCP server with only the actions it needs, scoped to specific resource ARNs.

Your MCP server runs as an ECS task or Lambda function. Attach an IAM task/execution role with specific policies. Boto3 automatically uses the role's credentials — no hardcoded keys.

MCP ServerIAM Actions NeededResource Scope
S3 MCPs3:ListBuckets, s3:GetObject, s3:PutObject, s3:GeneratePresignedUrlarn:aws:s3:::your-prefix-*
DynamoDB MCPdynamodb:Query, dynamodb:GetItem, dynamodb:PutItem, dynamodb:UpdateItemarn:aws:dynamodb:region:acct:table/your-*
Lambda MCPlambda:InvokeFunction, lambda:GetFunctionarn:aws:lambda:region:acct:function:mcp-allowed-*
CloudWatch MCPlogs:GetLogEvents, logs:FilterLogEvents, cloudwatch:GetMetricDataarn:aws:logs:region:acct:log-group:/aws/*
🪣 S3 Tools

S3 MCP Server — Full Implementation

A production S3 MCP server gives Claude the ability to list buckets, read/write objects, and generate time-limited presigned URLs — without ever exposing your AWS credentials to the conversation.

from fastmcp import FastMCP
import boto3, json
from botocore.exceptions import ClientError
from pydantic import constr, conint

mcp = FastMCP("AWSS3Server")
s3  = boto3.client("s3")  # Uses ECS task role automatically

@mcp.tool()
async def s3_list_objects(
    bucket: str,
    prefix: str = "",
    max_keys: conint(ge=1, le=1000) = 100
) -> str:
    """List objects in an S3 bucket with optional prefix filter."""
    try:
        resp = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=max_keys)
        objects = [
            {"key": o["Key"], "size_kb": round(o["Size"]/1024,1), "modified": str(o["LastModified"])}
            for o in resp.get("Contents", [])
        ]
        return json.dumps({"bucket": bucket, "count": len(objects), "objects": objects}, indent=2)
    except ClientError as e:
        return f"Error: {e.response['Error']['Code']} — {e.response['Error']['Message']}"

@mcp.tool()
async def s3_get_object(bucket: str, key: str, max_bytes: int = 50000) -> str:
    """Read an S3 object (text files, JSON, CSV). Truncates at max_bytes."""
    try:
        obj = s3.get_object(Bucket=bucket, Key=key)
        content = obj["Body"].read(max_bytes).decode("utf-8", errors="replace")
        size = obj["ContentLength"]
        truncated = size > max_bytes
        return json.dumps({"key": key, "size_bytes": size, "truncated": truncated, "content": content})
    except ClientError as e:
        return f"Error: {e.response['Error']['Message']}"

@mcp.tool()
async def s3_presigned_url(bucket: str, key: str, expires_in: int = 3600) -> str:
    """Generate a time-limited presigned URL for direct S3 access (max 7 days)."""
    expires_in = min(expires_in, 604800)
    url = s3.generate_presigned_url("get_object", Params={"Bucket":bucket,"Key":key}, ExpiresIn=expires_in)
    return json.dumps({"url": url, "expires_in_seconds": expires_in})
🗄️ DynamoDB

DynamoDB MCP Server

DynamoDB tools give Claude the ability to query items, run conditional writes, and perform transactional updates. Always use parameterized expressions — never string-format user input into FilterExpressions.

dynamo = boto3.resource("dynamodb")

@mcp.tool()
async def dynamo_query(table_name: str, pk_value: str, sk_prefix: str = "") -> str:
    """Query a DynamoDB table by partition key with optional sort key prefix."""
    from boto3.dynamodb.conditions import Key
    table = dynamo.Table(table_name)
    condition = Key("PK").eq(pk_value)
    if sk_prefix:
        condition &= Key("SK").begins_with(sk_prefix)
    resp = table.query(KeyConditionExpression=condition, Limit=100)
    return json.dumps({"count": resp["Count"], "items": resp["Items"]}, default=str, indent=2)

@mcp.tool()
async def dynamo_put_item(table_name: str, item_json: str) -> str:
    """Write an item to DynamoDB. item_json must be a valid JSON object."""
    import json
    item = json.loads(item_json)
    table = dynamo.Table(table_name)
    table.put_item(Item=item)
    return f"✅ Item written to {table_name}"
✅
Always use boto3 condition expressions

Never format user input into FilterExpression strings. Use Key() and Attr() condition objects — they automatically handle escaping and prevent injection attacks.

⚡ Lambda

Lambda Invoker MCP Tool

The Lambda invoker tool lets Claude trigger any of your existing Lambda functions by name. This is powerful — your Lambda functions can do anything, and the MCP tool just proxies the call. Add an allowlist so only approved functions can be invoked.

lambda_client = boto3.client("lambda")
ALLOWED_FUNCTIONS = {"data-processor", "report-generator", "notification-sender"}

@mcp.tool()
async def lambda_invoke(function_name: str, payload_json: str = "{}") -> str:
    """Invoke an approved Lambda function and return its response."""
    if function_name not in ALLOWED_FUNCTIONS:
        return f"❌ Function '{function_name}' not in allowlist. Allowed: {ALLOWED_FUNCTIONS}"
    resp = lambda_client.invoke(
        FunctionName=function_name,
        InvocationType="RequestResponse",
        Payload=payload_json.encode()
    )
    result = json.loads(resp["Payload"].read())
    return json.dumps({"function": function_name, "status": resp["StatusCode"], "result": result}, indent=2)
🔒 IAM Policy

IAM Least-Privilege Policy

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3MCPAccess",
      "Effect": "Allow",
      "Action": ["s3:ListBucket","s3:GetObject","s3:PutObject"],
      "Resource": [
        "arn:aws:s3:::your-data-bucket",
        "arn:aws:s3:::your-data-bucket/*"
      ]
    },
    {
      "Sid": "DynamoMCPAccess",
      "Effect": "Allow",
      "Action": ["dynamodb:Query","dynamodb:GetItem","dynamodb:PutItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:ACCOUNT:table/prod-*"
    },
    {
      "Sid": "LambdaMCPInvoke",
      "Effect": "Allow",
      "Action": ["lambda:InvokeFunction"],
      "Resource": [
        "arn:aws:lambda:us-east-1:ACCOUNT:function:data-processor",
        "arn:aws:lambda:us-east-1:ACCOUNT:function:report-generator"
      ]
    }
  ]
}
🌐 Full AWS Assistant

Real-World: Full AWS Assistant

Combine all tools into one MCP server and give it to Claude Code. A developer can now ask: "Show me all S3 objects in the reports bucket from this week, pick the largest one, read its contents, summarize the key findings, and save the summary back to s3://reports/summaries/latest.json."

Claude orchestrates: s3_list_objects → s3_get_object → natural language summarization → s3_put_object. Four tool calls, zero manual AWS Console work.

💡
Use STS AssumeRole for cross-account access

For MCP servers that need to access multiple AWS accounts, use STS assume_role() with a short-lived session (900s TTL) instead of separate credentials per account.

🧠 Knowledge Check — Day 21

4 questions on AWS MCP integration

Q1/4
Why should you never use AdministratorAccess for your MCP server's IAM role?
AIt costs more
BIt violates least-privilege — a compromised MCP server would have full AWS account access
CAdministratorAccess doesn't allow S3 access
Dboto3 doesn't support administrator roles
✅ B. The blast radius of a compromised MCP server with AdministratorAccess is your entire AWS account. Least-privilege limits damage to only the specific resources and actions the server legitimately needs.
Q2/4
How does a boto3 client running in an ECS task automatically get AWS credentials?
AFrom hardcoded keys in the Dockerfile
BFrom the ECS task IAM role via the EC2 metadata service — no explicit credentials needed
CFrom environment variables set manually
DFrom AWS CLI configuration files
✅ B. ECS automatically injects credentials from the task IAM role via the container credentials endpoint. boto3's credential chain picks these up automatically — no hardcoded keys needed, and they rotate automatically.
Q3/4
What is a DynamoDB condition expression used for in MCP tools?
ATo authenticate requests
BSafe, parameterized query conditions that prevent injection attacks
CTo compress query results
DTo add TTL to items
✅ B. boto3's Key() and Attr() condition objects create parameterized expressions that prevent injection. Never format user input directly into a FilterExpression string — it's the DynamoDB equivalent of SQL injection.
Q4/4
Why should the Lambda invoker MCP tool use an allowlist of function names?
ALambda has a limit of 10 functions
BTo prevent the AI from invoking destructive or sensitive functions not intended for MCP use
CFor performance optimization
DAWS requires allowlists for Lambda invocation
✅ B. Without an allowlist, Claude could invoke any Lambda the role has access to — including functions designed for admin tasks, data deletion, or cost-heavy operations. An allowlist provides defense-in-depth beyond IAM.