MCP: Model Context Protocol on AWS with a FastAPI Example

Scenario

A company wants to expose internal tools and knowledge sources to AI assistants through MCP while keeping everything controlled, auditable, and secure.

Problem and Business Context

Teams often connect AI assistants to internal systems with ad-hoc scripts. That creates three common risks:

• unbounded tool permissions
• weak auditability of who invoked what
• brittle integration patterns that are hard to govern

MCP gives a standardized way for assistants to discover tools and invoke them. On AWS, the goal is to deploy MCP as an explicit control plane with authentication, authorization, and audit trails.

Architecture Choices and Trade-offs

Option A: API Gateway + Lambda (FastAPI + Mangum) + DynamoDB (recommended for low traffic)

• Pros: low baseline cost, scale-to-zero, simple deployment
• Cons: cold starts and request timeout constraints

Option B: ECS Fargate MCP service + ALB

• Pros: stable latency, long-lived connections for SSE
• Cons: higher fixed monthly cost

Option C: EKS multi-tenant MCP gateway

• Pros: highest flexibility
• Cons: significant operational overhead

For most internal assistant workloads, Option A is the best first production deployment.

AWS Reference Architecture

Control Principles

• Every MCP tool has an owner, scope, and risk level.
• High-risk tools require a human-approval branch (Step Functions Wait for task token).
• Tool execution identity is role-based, never shared access keys.
• Every tool call is logged with request context, policy decision, and output classification.

Step-by-Step Tutorial

1) Bootstrap variables

export AWS_REGION=us-east-1
export PROJECT=mcp-internal
export ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
export TABLE_NAME=${PROJECT}-audit
export FN_NAME=${PROJECT}-server

$env:AWS_REGION = "us-east-1"
$env:PROJECT = "mcp-internal"
$env:ACCOUNT_ID = (aws sts get-caller-identity --query Account --output text)
$env:TABLE_NAME = "$($env:PROJECT)-audit"
$env:FN_NAME = "$($env:PROJECT)-server"

2) Create audit table and config parameters

aws dynamodb create-table \
  --table-name "$TABLE_NAME" \
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=ts,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=ts,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification Enabled=true

aws ssm put-parameter \
  --name "/${PROJECT}/allowed_tools" \
  --type String \
  --value '["search_policies","get_ticket_status","list_incidents"]' \
  --overwrite

aws secretsmanager create-secret \
  --name "${PROJECT}/db-readonly" \
  --secret-string '{"username":"readonly_user","password":"REPLACE_ME","host":"db.internal","port":5432,"dbname":"internal"}'

aws dynamodb create-table `
  --table-name $env:TABLE_NAME `
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=ts,AttributeType=S `
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=ts,KeyType=RANGE `
  --billing-mode PAY_PER_REQUEST `
  --sse-specification Enabled=true

aws ssm put-parameter `
  --name "/$($env:PROJECT)/allowed_tools" `
  --type String `
  --value '["search_policies","get_ticket_status","list_incidents"]' `
  --overwrite

3) IAM least-privilege policy for MCP runtime

mcp-runtime-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem"],
      "Resource": "arn:aws:dynamodb:*:*:table/mcp-internal-audit"
    },
    {
      "Effect": "Allow",
      "Action": ["ssm:GetParameter"],
      "Resource": "arn:aws:ssm:*:*:parameter/mcp-internal/*"
    },
    {
      "Effect": "Allow",
      "Action": ["secretsmanager:GetSecretValue"],
      "Resource": "arn:aws:secretsmanager:*:*:secret:mcp-internal/*"
    },
    {
      "Effect": "Allow",
      "Action": ["states:StartExecution"],
      "Resource": "arn:aws:states:*:*:stateMachine:mcp-approval-*"
    }
  ]
}

Attach it as an inline or managed policy to the Lambda execution role.

4) FastAPI MCP server example

This example uses a simple HTTP shape while keeping MCP-like semantics (tools/list, tools/call).

mcp_server/main.py

import json
import os
from datetime import datetime, timezone
from typing import Any

import boto3
from fastapi import FastAPI, HTTPException, Request
from mangum import Mangum
from pydantic import BaseModel

app = FastAPI(title="Internal MCP Server")

ssm = boto3.client("ssm")
ddb = boto3.resource("dynamodb")

AUDIT_TABLE = os.environ["AUDIT_TABLE"]
ALLOWED_TOOLS_PARAM = os.environ["ALLOWED_TOOLS_PARAM"]

table = ddb.Table(AUDIT_TABLE)


class ToolCall(BaseModel):
    name: str
    arguments: dict[str, Any]


def get_allowed_tools() -> set[str]:
    raw = ssm.get_parameter(Name=ALLOWED_TOOLS_PARAM)["Parameter"]["Value"]
    return set(json.loads(raw))


def audit(event_type: str, actor: str, payload: dict[str, Any]) -> None:
    table.put_item(Item={
        "pk": f"ACTOR#{actor}",
        "ts": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        "payload": payload,
    })


def search_policies(arguments: dict[str, Any]) -> dict[str, Any]:
    keyword = arguments.get("keyword", "")
    return {"matches": [f"Policy section related to: {keyword}"]}


def get_ticket_status(arguments: dict[str, Any]) -> dict[str, Any]:
    ticket_id = arguments.get("ticket_id", "unknown")
    return {"ticket_id": ticket_id, "status": "in_progress"}


TOOL_IMPL = {
    "search_policies": search_policies,
    "get_ticket_status": get_ticket_status,
}


@app.get("/mcp/tools/list")
def tools_list() -> dict:
    tools = [
        {"name": "search_policies", "description": "Search internal policy snippets"},
        {"name": "get_ticket_status", "description": "Read ticket status"},
        {"name": "list_incidents", "description": "List latest incidents (read-only)"}
    ]
    return {"tools": tools}


@app.post("/mcp/tools/call")
async def tools_call(req: Request, call: ToolCall) -> dict:
    actor = req.headers.get("x-employee-id", "unknown")
    allowed = get_allowed_tools()

    if call.name not in allowed:
        audit("tool_denied", actor, {"tool": call.name, "reason": "not_allowlisted"})
        raise HTTPException(status_code=403, detail="Tool is not allowed")

    if call.name not in TOOL_IMPL:
        audit("tool_missing", actor, {"tool": call.name})
        raise HTTPException(status_code=404, detail="Tool implementation not found")

    result = TOOL_IMPL[call.name](call.arguments)
    audit("tool_called", actor, {"tool": call.name, "args": call.arguments})
    return {"result": result}


handler = Mangum(app)

5) Package and deploy

cat > requirements.txt << 'TXT'
fastapi==0.115.0
mangum==0.17.0
boto3==1.35.0
TXT

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt -t package
cp -r mcp_server package/
cd package && zip -r ../mcp.zip . && cd ..

aws lambda create-function \
  --function-name "$FN_NAME" \
  --runtime python3.12 \
  --handler mcp_server.main.handler \
  --role "arn:aws:iam::${ACCOUNT_ID}:role/${PROJECT}-lambda-role" \
  --zip-file fileb://mcp.zip \
  --timeout 30 \
  --memory-size 512 \
  --environment "Variables={AUDIT_TABLE=$TABLE_NAME,ALLOWED_TOOLS_PARAM=/$PROJECT/allowed_tools}"

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt -t package
Copy-Item -Recurse mcp_server package
Compress-Archive -Path package\* -DestinationPath mcp.zip -Force

aws lambda create-function `
  --function-name $env:FN_NAME `
  --runtime python3.12 `
  --handler mcp_server.main.handler `
  --role "arn:aws:iam::$($env:ACCOUNT_ID):role/$($env:PROJECT)-lambda-role" `
  --zip-file fileb://mcp.zip `
  --timeout 30 `
  --memory-size 512 `
  --environment "Variables={AUDIT_TABLE=$($env:TABLE_NAME),ALLOWED_TOOLS_PARAM=/$($env:PROJECT)/allowed_tools}"

6) Publish via API Gateway with JWT auth

API_ID=$(aws apigatewayv2 create-api --name "${PROJECT}-api" --protocol-type HTTP --target "arn:aws:lambda:${AWS_REGION}:${ACCOUNT_ID}:function:${FN_NAME}" --query ApiId --output text)

aws lambda add-permission \
  --function-name "$FN_NAME" \
  --statement-id apigw-mcp \
  --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:${AWS_REGION}:${ACCOUNT_ID}:${API_ID}/*/*"

AUTH_ID=$(aws apigatewayv2 create-authorizer \
  --api-id "$API_ID" \
  --authorizer-type JWT \
  --name mcp-jwt \
  --identity-source '$request.header.Authorization' \
  --jwt-configuration Audience=mcp-internal,Issuer=https://id.example.com \
  --query AuthorizerId --output text)

Set route auth type to JWT so all calls must carry valid employee tokens.

7) Security controls

• Use WAF managed rules on API Gateway stage.
• Put Lambda in VPC only if private data access is required; otherwise keep it outside VPC to reduce cost/latency.
• Deny dangerous tools by default; use explicit allowlist.
• Separate read-only and write-capable tools into different MCP servers and roles.
• Require human approval for write tools (Step Functions branch + signed approval record).

8) Monitoring and auditability

• CloudWatch Logs retention policy: 30-90 days hot, archive to S3 for long retention.
• Create metric filters:
• tool_denied spikes
• 5xx from API Gateway
• 429 rate limiting
• Alarm into SNS + PagerDuty/Slack integration.

Example alarm command:

aws cloudwatch put-metric-alarm \
  --alarm-name "${PROJECT}-api-5xx" \
  --namespace "AWS/ApiGateway" \
  --metric-name "5xx" \
  --dimensions Name=ApiId,Value="$API_ID" Name=Stage,Value='$default' \
  --period 60 --evaluation-periods 5 --threshold 5 --statistic Sum \
  --comparison-operator GreaterThanOrEqualToThreshold \
  --alarm-actions "arn:aws:sns:${AWS_REGION}:${ACCOUNT_ID}:platform-alerts"

9) Cost optimization

• Lambda + HTTP API is usually cheaper than always-on containers for bursty internal traffic.
• Use caching for static tool metadata (/mcp/tools/list).
• Keep DynamoDB PAY_PER_REQUEST until call volume stabilizes.
• Add concurrency limits to protect downstream systems.
• Configure AWS Budgets and alerts for API Gateway, Lambda, and data stores.

Pricing reminder: always verify current prices at:

• https://aws.amazon.com/api-gateway/pricing/
• https://aws.amazon.com/lambda/pricing/
• https://aws.amazon.com/dynamodb/pricing/

10) Production checklist

• JWT auth enforced and tested with invalid token cases
• Tool allowlist comes from SSM/Secrets, not source code
• Per-tool IAM role boundaries documented
• High-risk tools have approval workflow
• Audit records immutable and queryable
• CloudWatch alarms + on-call integration tested
• Runbooks for auth outage, downstream outage, and rollback completed
• Pen-test/red-team findings tracked and remediated

Final takeaway

MCP on AWS should be treated as a secure integration platform, not just a protocol endpoint. FastAPI + Lambda + API Gateway gives a practical low-cost starting point, while IAM boundaries, audit trails, and approval flows make the solution production-safe.