Build a Secure AI Agent Platform on Your Laptop: A Code Walkthrough

In my last post I wrote about what AWS's Secure AI Agents on Amazon EKS workshop teaches: keep the agent small and put every safeguard in the platform around it. Several readers asked the obvious next question: can I build it myself?

The workshop can't help you there. It only runs in temporary accounts at AWS events, and its sandbox needs bare-metal EKS nodes for Firecracker. So I rebuilt every idea from scratch on a laptop, with free tools, and tested each piece as I went. Everything in this post ran on my machine; where something failed or surprised me, I've left it in, because those are the parts worth learning from.

Here is what replaces what:

You'll need Docker, uv for Python, and Ollama if you want a free local model. I built and tested everything on Ubuntu 22.04, then replayed the whole post from scratch in an empty folder to make sure nothing is missing. macOS uses the same tools (there's an Apple Silicon build of the gateway), and on Windows, use WSL2. Budget about an hour. If you'd rather clone than copy, the same code is on GitHub at felixmakinda/secure-agents-lab, with a quickstart in its README.

The shape of the project#

secure-agents-lab/
├── shop/seed.py              # sample data (SQLite)
├── identity/idp.py           # a pocket identity provider: keys and persona tokens
├── gateway/config.yaml       # agentgateway: the model gate and the tool gate
├── orders_mcp/server.py      # order tools, served over MCP
├── code_broker/server.py     # run_python: model-written code, sandboxed
├── sandbox/                  # the throwaway container image
│   ├── Dockerfile
│   └── runner.py
├── agent/
│   ├── tools.py              # tools that live inside the agent
│   ├── agent.py              # the model client and the agent
│   └── server.py             # POST /chat, streamed over Server-Sent Events
├── chat.py                   # talk to the agent as a persona
├── check_gate.py             # prove the tool gate
├── check_sandbox.py          # prove the sandbox
└── run.sh                    # start and stop everything

Notice what the agent folder doesn't contain: no database code, no cloud credentials and no business tools. By the end it will hold one tool of its own (today's date, for a reason you'll see) and learn everything else at runtime.

Step 0: Set up#

Create the project and pin the versions I tested with. The MCP Python SDK moved to version 2 recently and renamed things, so older tutorials won't work against it; these pins keep you on what this post was written for.

mkdir secure-agents-lab && cd secure-agents-lab
uv init --python 3.12 --bare
uv add "strands-agents[openai,otel]==1.62.0" "mcp==2.1.1" "pyjwt[crypto]==2.15.1" \
       "fastapi==0.161.1" "uvicorn==0.54.0" "httpx==0.28.1" \
       "opentelemetry-instrumentation-httpx==0.66b1"

Download the agentgateway binary (v1.6.0) and check it. Use agentgateway-darwin-arm64 on an Apple Silicon Mac:

mkdir -p bin
curl -L -o bin/agentgateway \
  https://github.com/agentgateway/agentgateway/releases/download/v1.6.0/agentgateway-linux-amd64
curl -sL https://github.com/agentgateway/agentgateway/releases/download/v1.6.0/agentgateway-linux-amd64.sha256
sha256sum bin/agentgateway          # macOS: shasum -a 256 bin/agentgateway. The hashes must match.
chmod +x bin/agentgateway && bin/agentgateway --version

On a slow connection my first download was silently truncated, and the "binary" crashed with a segmentation fault. The checksum is not a formality.

Finally, a model. For a free local one:

ollama pull qwen2.5:3b

A 3-billion-parameter model is enough for order lookups and, with some help, simple pandas. It is slow on a CPU: on my laptop a single answer took 40 seconds to 3 minutes. With a GPU, an Apple Silicon Mac or a cloud key (Step 3 shows how) it's a few seconds.

Step 1: Some data to talk about#

Five fixed orders, one per status, so every conversation path has something to find, plus a quarter of random sales for the analysis step. A fixed random seed means your numbers will match mine.

# shop/seed.py
"""Create shop.db: a few fixed orders to talk about, plus a quarter of sales to analyse."""
import random
import sqlite3
from datetime import date, timedelta
from pathlib import Path

DB = Path(__file__).with_name("shop.db")

PRODUCTS = [
    ("Laptop Pro 15", 1299.99, 4),
    ("Mechanical Keyboard", 89.99, 0),
    ("Wireless Mouse", 29.99, 57),
    ("Noise-Cancelling Headphones", 199.99, 12),
    ("USB-C Dock", 149.99, 8),
]
REGIONS = ["West", "East", "South", "Central"]

# One order per status, so every conversation path has something to find.
FIXED = [
    ("ORD-1001", "Laptop Pro 15", 1, "shipped", "2026-03-28", "West"),
    ("ORD-1002", "Wireless Mouse", 2, "delivered", "2026-03-20", "East"),
    ("ORD-1003", "USB-C Dock", 1, "processing", "2026-03-30", "South"),
    ("ORD-1004", "Noise-Cancelling Headphones", 1, "cancelled", "2026-03-15", "Central"),
    ("ORD-1005", "Mechanical Keyboard", 1, "returned", "2026-03-02", "West"),
]


def main() -> None:
    DB.unlink(missing_ok=True)
    db = sqlite3.connect(DB)
    db.executescript(
        """
        CREATE TABLE products (name TEXT PRIMARY KEY, price REAL, stock INTEGER);
        CREATE TABLE orders (
            order_id TEXT PRIMARY KEY, product TEXT, quantity INTEGER, total REAL,
            status TEXT, order_date TEXT, region TEXT, return_reason TEXT
        );
        """
    )
    db.executemany("INSERT INTO products VALUES (?, ?, ?)", PRODUCTS)
    price = {name: p for name, p, _ in PRODUCTS}

    rows = [(oid, prod, qty, round(price[prod] * qty, 2), status, day, region, None)
            for oid, prod, qty, status, day, region in FIXED]

    # 135 random Q1 2026 orders. A fixed seed keeps everyone's numbers identical.
    rng = random.Random(42)
    start = date(2026, 1, 1)
    for i in range(135):
        prod, p, _ = rng.choice(PRODUCTS)
        qty = rng.choice([1, 1, 1, 2, 3])
        day = start + timedelta(days=rng.randrange(90))
        rows.append((f"ORD-{2000 + i}", prod, qty, round(p * qty, 2), "delivered",
                     day.isoformat(), rng.choice(REGIONS), None))

    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?)", rows)
    db.commit()
    print(f"Created {DB.name}: {len(PRODUCTS)} products, {len(rows)} orders")


if __name__ == "__main__":
    main()

uv run python shop/seed.py
# Created shop.db: 5 products, 140 orders

Step 2: A pocket identity provider#

In the workshop, Amazon Cognito signed users in and put their role in a cognito:groups claim. On a laptop, this script plays Cognito: it makes an RSA signing key, publishes the public half as a JWKS file for the gateway, and mints a one-hour token per persona.

# identity/idp.py
"""A pocket identity provider, standing in for Amazon Cognito.

    python identity/idp.py keys              # once: signing key + public JWKS
    python identity/idp.py token support     # a 1-hour token for a persona

The gateway only ever sees the public half (jwks.json). The private key
stays here, the way it stays inside Cognito.
"""
import json
import sys
import time
from pathlib import Path

import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from jwt.algorithms import RSAAlgorithm

HERE = Path(__file__).parent
PRIVATE = HERE / "private.pem"
JWKS = HERE / "jwks.json"
ISSUER = "https://idp.shop.local"
AUDIENCE = "shop-tools"
KID = "shop-key-1"

PERSONAS = {
    "support": {"sub": "sam@shop.local", "groups": ["support-associate"]},
    "analyst": {"sub": "ana@shop.local", "groups": ["sales-analyst"]},
}


def make_keys() -> None:
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    PRIVATE.write_bytes(
        key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
    )
    PRIVATE.chmod(0o600)
    public = json.loads(RSAAlgorithm.to_jwk(key.public_key()))
    public.update({"kid": KID, "use": "sig", "alg": "RS256"})
    JWKS.write_text(json.dumps({"keys": [public]}, indent=2))
    print(f"wrote {PRIVATE.name} (keep secret) and {JWKS.name} (give to the gateway)")


def mint(persona: str) -> str:
    now = int(time.time())
    claims = {**PERSONAS[persona], "iss": ISSUER, "aud": AUDIENCE, "iat": now, "exp": now + 3600}
    return jwt.encode(claims, PRIVATE.read_bytes(), algorithm="RS256", headers={"kid": KID})


if __name__ == "__main__":
    if sys.argv[1:] == ["keys"]:
        make_keys()
    elif len(sys.argv) == 3 and sys.argv[1] == "token" and sys.argv[2] in PERSONAS:
        print(mint(sys.argv[2]))
    else:
        sys.exit(f"usage: idp.py keys | idp.py token {{{','.join(PERSONAS)}}}")

uv run python identity/idp.py keys
# wrote private.pem (keep secret) and jwks.json (give to the gateway)
uv run python identity/idp.py token analyst     # prints a JWT

The split matters. The gateway only ever receives jwks.json, the public key, so it can check tokens but never create them. Paste a token into jwt.io and you'll see "groups": ["sales-analyst"]. That one claim decides which tools the analyst gets.

Step 3: The model gate#

Here is the whole gateway configuration. One binary runs two gates on two ports: models on 4000, tools on 3000. We'll go through the model half first.

# gateway/config.yaml
# yaml-language-server: $schema=https://agentgateway.dev/schema/config
#
# One agentgateway, two gates:
#   :4000  the model gate  (OpenAI-compatible, aliases, caller keys)
#   :3000  the tool gate   (MCP, user JWTs, per-tool rules)

config:
  tracing:
    otlpEndpoint: http://localhost:4317   # Jaeger, gRPC
    randomSampling: true

gateways:
  models:
    port: 4000
    bindAddress: 127.0.0.1   # loopback only: nothing on your Wi-Fi can reach it
  tools:
    port: 3000
    bindAddress: 127.0.0.1

llm:
  gateways: models
  policies:
    # The gap the workshop left open: callers must present a key.
    apiKey:
      mode: strict
      keys:
      - key: "$AGENT_GATEWAY_KEY"
        metadata:
          caller: shop-agent
  models:
  # Agents ask for an alias. Swap what it points at here, and no agent changes.
  - name: shop-fast
    provider: ollama
    params:
      model: qwen2.5:3b
  # To use a cloud model instead, change provider and params here (see the post).

mcp:
  gateways: tools
  policies:
    # Outermost time limit. The broker gives up at 40s and the code at 20s,
    # so each layer fails cleanly before the one outside it.
    timeout:
      requestTimeout: 60s
    # Every MCP call needs a valid token from our identity provider.
    jwtAuth:
      mode: strict
      issuer: https://idp.shop.local
      audiences: [shop-tools]
      jwks:
        file: ./identity/jwks.json
    # One rule per allowed tool. Anything not listed is denied, and hidden from tools/list.
    mcpAuthorization:
      rules:
      - 'mcp.tool.name == "lookup_order"'
      - 'mcp.tool.name == "initiate_return" && "support-associate" in jwt.groups'
      - 'mcp.tool.name == "run_python" && "sales-analyst" in jwt.groups'
      # check_inventory: deliberately unmapped, so nobody gets it.
  targets:
  - name: orders
    mcp:
      host: http://localhost:8101/mcp
  - name: code
    mcp:
      host: http://localhost:8102/mcp

Four decisions in the model half:

1. Aliases. The agent asks for shop-fast. The gateway decides what that means. Today it's a local Qwen model; tomorrow you change two lines and every agent follows, with no rebuild.
2. Caller keys. The workshop's gateway accepted calls from any pod, which meant anything in the cluster could spend model tokens. Here the gate is strict: no key, no model.
3. Loopback only. Without bindAddress, agentgateway listens on every network interface. On a laptop in a café, that would put your gateway on the café's Wi-Fi. 127.0.0.1 keeps it on your machine.
4. The agent never sees a provider key. If you switch to OpenAI or Bedrock, the provider credential lives in the gateway's environment, not the agent's.

To use a cloud model, replace the shop-fast entry. For OpenAI (then export OPENAI_API_KEY=... before starting the gateway):

  - name: shop-fast
    provider: openAI
    params:
      model: gpt-4.1-mini
      apiKey: "$OPENAI_API_KEY"

For Amazon Bedrock, with your usual AWS credentials in the environment:

  - name: shop-fast
    provider: bedrock
    params:
      model: us.amazon.nova-2-lite-v1:0
      awsRegion: us-east-1

A gotcha that cost me ten minutes: agentgateway expands $VARIABLES everywhere in the file, including inside YAML comments. I had a commented-out OpenAI example in my config and the gateway refused to start with environment variable not found: OPENAI_API_KEY. Keep examples out of the file, or out of reach of the $.

We'll start everything together in Step 8, but you can already prove the model gate on its own:

AGENT_GATEWAY_KEY=sk-shop-agent-local ./bin/agentgateway -f gateway/config.yaml &

# No key: refused
curl -s -o /dev/null -w "%{http_code}\n" localhost:4000/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"shop-fast","messages":[{"role":"user","content":"Say hi in 3 words"}]}'
# 401

# With the key: answered, and the alias resolved
curl -s localhost:4000/v1/chat/completions -H "Authorization: Bearer sk-shop-agent-local" \
  -H 'Content-Type: application/json' \
  -d '{"model":"shop-fast","messages":[{"role":"user","content":"Say hi in 3 words"}]}'
# ... "content": "Hello there!" ... "model": "qwen2.5:3b" ...

We asked for shop-fast and the response says qwen2.5:3b. That's the alias rewrite, visible from outside.

Step 4: Tools on the network#

The order tools move out of the agent into their own MCP server. Anything that speaks MCP can discover and call them, and, more importantly, a gateway can stand between the agent and every call.

# orders_mcp/server.py
"""Order tools, served over MCP. The agent discovers these at runtime."""
import sqlite3
from pathlib import Path

from mcp.server.mcpserver import MCPServer
from mcp.server.transport_security import TransportSecuritySettings

DB = Path(__file__).parent.parent / "shop" / "shop.db"

mcp = MCPServer("shop-orders")


def _db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB)
    conn.row_factory = sqlite3.Row
    return conn


@mcp.tool()
def lookup_order(order_id: str) -> dict:
    """Look up an order's status, product, total, date and region by its ID, e.g. ORD-1001."""
    with _db() as db:
        row = db.execute("SELECT * FROM orders WHERE order_id = ?", (order_id.strip().upper(),)).fetchone()
    return dict(row) if row else {"error": f"No order with ID {order_id}."}


@mcp.tool()
def check_inventory(product_name: str) -> dict:
    """Check how many units of a product are in stock."""
    with _db() as db:
        row = db.execute(
            "SELECT name, stock FROM products WHERE name LIKE ?", (f"%{product_name.strip()}%",)
        ).fetchone()
    return dict(row) if row else {"error": f"No product matching {product_name!r}."}


@mcp.tool()
def initiate_return(order_id: str, reason: str) -> dict:
    """Start a return for a delivered order. Ask the customer for a reason first."""
    with _db() as db:
        row = db.execute("SELECT status FROM orders WHERE order_id = ?", (order_id.upper(),)).fetchone()
        if not row:
            return {"error": f"No order with ID {order_id}."}
        if row["status"] != "delivered":
            return {"error": f"Only delivered orders can be returned; this one is {row['status']}."}
        db.execute(
            "UPDATE orders SET status = 'return_requested', return_reason = ? WHERE order_id = ?",
            (reason, order_id.upper()),
        )
    return {"order_id": order_id.upper(), "status": "return_requested"}


if __name__ == "__main__":
    # Listens on localhost only: the gateway is the one way in.
    mcp.run(
        "streamable-http",
        host="127.0.0.1",
        port=8101,
        # Keep DNS-rebinding protection on; the gateway forwards "Host: localhost".
        transport_security=TransportSecuritySettings(
            allowed_hosts=["localhost", "localhost:*", "127.0.0.1", "127.0.0.1:*"]
        ),
    )

Three details:

• MCPServer is the new name for FastMCP. In version 2 of the MCP Python SDK, from mcp.server.fastmcp import FastMCP raises an error that points you to the migration guide. Same decorator, new import.
• It listens on 127.0.0.1 only. The gateway is the one way in.
• DNS-rebinding protection stays on. The workshop switched it off. When I left it on, the server refused the gateway with 421 Misdirected Request, because the gateway forwards Host: localhost without a port and the SDK's default allow-list expects one. Adding exactly the hosts we trust fixes it without lowering the guard.

The broker in Step 7 is a second MCP server built the same way.

Step 5: The tool gate#

Look back at the mcp: half of the gateway config. Two policies do all the work:

• jwtAuth in strict mode checks every request's token against jwks.json, the issuer and the audience. No token, or a forged one, gets 401 before any tool runs.
• mcpAuthorization lists one rule per allowed tool, written in CEL. A rule can read the token's claims (jwt.groups) and the tool being touched (mcp.tool.name). Once any rule exists, everything not allowed is denied. check_inventory has no rule, so nobody gets it.

The gateway also merges our two MCP servers into one endpoint and prefixes each tool with its server's name, so the agent sees orders_lookup_order and code_run_python. The rules still match the plain names.

This script asks the gate what each persona can see, then tries to call a tool that persona can't see:

# check_gate.py
"""Ask the tool gate what each persona can see, then try a tool it can't see."""
import subprocess
import sys

from strands.tools.mcp import MCPClient


def token(persona: str) -> str:
    return subprocess.check_output([sys.executable, "identity/idp.py", "token", persona], text=True).strip()


def client(tok: str | None) -> MCPClient:
    headers = {"Authorization": f"Bearer {tok}"} if tok else {}
    return MCPClient(url="http://localhost:3000/mcp", headers=headers)


try:
    with client(None) as c:
        c.list_tools_sync()
    print("no token  -> allowed?!")
except Exception as e:
    print("no token  -> refused:", type(e).__name__)

for persona in ("support", "analyst"):
    with client(token(persona)) as c:
        names = sorted(t.tool_name for t in c.list_tools_sync())
        print(f"{persona:8}  -> sees {names}")
        hidden = "code_run_python" if persona == "support" else "orders_initiate_return"
        args = {"code": "print(1)"} if hidden == "code_run_python" else {"order_id": "ORD-1002", "reason": "test"}
        result = c.call_tool_sync("t1", hidden, args)
        print(f"{'':8}     calls {hidden} anyway -> {result['status']}: {result['content'][0].get('text', '')[:90]}")

no token  -> refused: MCPClientInitializationError
support   -> sees ['orders_initiate_return', 'orders_lookup_order']
             calls code_run_python anyway -> error: Tool execution failed: Unknown tool: code_run_python
analyst   -> sees ['code_run_python', 'orders_lookup_order']
             calls orders_initiate_return anyway -> error: Tool execution failed: Unknown tool: orders_initiate_return

Two lessons are in those five lines. First, a denied tool doesn't produce a 403; it disappears. The support persona's list simply doesn't contain run_python, so the model never knows it exists. Second, hiding isn't the only defence: calling a hidden tool directly is refused too. The gateway checks every call, not just the list.

That second point made me correct something. In my workshop post I wrote that removing someone's access doesn't reach a conversation already in progress, because the agent caches its tool list. So I tested it: open a session as the analyst, delete the run_python rule from the config (the gateway reloads it automatically), and call the tool again from the same session.

before:                      ['code_run_python', 'orders_lookup_order']
after revoke, same session -> error Tool execution failed: Unknown tool: code_run_python
after revoke, tools/list  -> ['orders_lookup_order']

Revocation is enforced on the very next call. What the cache really delays is the opposite direction: a newly granted tool won't appear until the agent fetches its list again, in a new conversation. That's a much better failure mode, and I've updated the earlier post.

Step 6: The agent#

The agent is three small files, in the same shape as the workshop's.

tools.py, the agent's only local tool.

# agent/tools.py
"""Tools that live inside the agent process. Everything else arrives over MCP."""
from datetime import date

from strands import tool


@tool
def today() -> str:
    """Return today's date (YYYY-MM-DD). Use it for anything relative, like "last month"."""
    return date.today().isoformat()

In the workshop, the model told a customer their parcel would arrive "in about 3 weeks from today". It had no idea what today was. Later, asked for "last month's sales", it quietly charted the newest month in the data instead. A one-line tool, plus today's date in the system prompt, removes that whole class of mistake.

agent.py, the model client and the agent.

# agent/agent.py
"""The customer-service agent. It holds no cloud credentials and imports no business tools."""
import os
from datetime import date

from strands import Agent
from strands.models.openai import OpenAIModel
from strands.tools.mcp import MCPClient

from tools import today

MODEL_BASE_URL = os.environ.get("MODEL_BASE_URL", "http://localhost:4000/v1")
MODEL_API_KEY = os.environ["MODEL_API_KEY"]          # a gateway key, not a provider key
MODEL_ID = os.environ.get("MODEL_ID", "shop-fast")   # an alias the gateway resolves
TOOLS_URL = os.environ.get("TOOLS_URL", "http://localhost:3000/mcp")

SYSTEM_PROMPT = """You are the customer-service assistant for Shop.
Use your tools to answer; never invent order details. If a tool is missing,
say you can't help with that. For sales questions, write pandas code for
run_python and report the printed numbers. Today is {today}."""


def build_model() -> OpenAIModel:
    return OpenAIModel(
        client_args={"base_url": MODEL_BASE_URL, "api_key": MODEL_API_KEY},
        model_id=MODEL_ID,
        params={"max_tokens": 1024, "temperature": 0.2},
    )


def build_agent(user_token: str) -> tuple[Agent, MCPClient]:
    """One agent per conversation, carrying that user's token on every tool call."""
    tools = MCPClient(url=TOOLS_URL, headers={"Authorization": f"Bearer {user_token}"})
    agent = Agent(
        model=build_model(),
        system_prompt=SYSTEM_PROMPT.format(today=date.today().isoformat()),
        tools=[today, tools],
        callback_handler=None,
    )
    return agent, tools


if __name__ == "__main__":
    # Quick CLI check:  MODEL_API_KEY=... python agent.py "$(python ../identity/idp.py token support)"
    import sys

    agent, _ = build_agent(sys.argv[1])
    print(agent("Where is my order ORD-1001?"))

The important function is build_agent(user_token). It builds one agent per conversation, and its MCPClient sends that user's token on every tool call. The agent never decides what a user may do. It carries the user's identity to the gate, and the gate decides. Note too that MODEL_API_KEY is a key for our gateway, not for any model provider.

server.py, the HTTP front door.

# agent/server.py
"""HTTP front door for the agent: POST /chat streams tokens back as Server-Sent Events."""
import json
import os

import httpx
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import Response, StreamingResponse
from pydantic import BaseModel

from agent import build_agent

# Traces: every model call and tool call, if a collector is configured.
if os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT"):
    from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
    from strands.telemetry import StrandsTelemetry

    StrandsTelemetry().setup_otlp_exporter()
    HTTPXClientInstrumentor().instrument()  # adds traceparent to model calls, so gateway spans join our trace

CHARTS_URL = os.environ.get("CHARTS_URL", "http://localhost:8102/charts")

app = FastAPI()
sessions: dict[str, tuple] = {}  # session_id -> (agent, mcp_client, token)


class ChatRequest(BaseModel):
    session_id: str
    message: str


def _session(session_id: str, token: str):
    """A conversation is pinned to the token that started it."""
    found = sessions.get(session_id)
    if found and found[2] != token:
        raise HTTPException(403, "This conversation belongs to another user.")
    if not found:
        agent, client = build_agent(token)
        sessions[session_id] = found = (agent, client, token)
    return found[0]


def _charts_since(agent, start: int) -> list[str]:
    """Pull chart IDs out of the tool results added during this turn."""
    ids = []
    for message in agent.messages[start:]:
        for block in message.get("content", []):
            for item in block.get("toolResult", {}).get("content", []):
                text = item.get("text", "")
                if '"chart_id"' in text:
                    try:
                        ids.append(json.loads(text)["chart_id"])
                    except (ValueError, KeyError):
                        pass
    return ids


@app.post("/chat")
async def chat(req: ChatRequest, authorization: str = Header()):
    token = authorization.removeprefix("Bearer ").strip()
    agent = _session(req.session_id, token)
    start = len(agent.messages)

    async def generate():
        announced: set[str] = set()
        async for event in agent.stream_async(req.message):
            if "data" in event:
                yield f"data: {json.dumps({'token': event['data']})}\n\n"
            elif "current_tool_use" in event:
                use = event["current_tool_use"]
                if use.get("toolUseId") not in announced:  # streamed in pieces; announce once
                    announced.add(use.get("toolUseId"))
                    yield f"data: {json.dumps({'tool': use.get('name')})}\n\n"
        for chart_id in _charts_since(agent, start):
            yield f"data: {json.dumps({'chart': f'/charts/{chart_id}'})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")


@app.get("/charts/{chart_id}")
async def chart(chart_id: str):
    """The image travels here, beside the conversation, never through the model."""
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{CHARTS_URL}/{chart_id}")
    if r.status_code != 200:
        raise HTTPException(404)
    return Response(r.content, media_type="image/png")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8080)

POST /chat streams the reply as Server-Sent Events: text tokens, a tool event when a tool starts, and a chart event when the sandbox produced an image. Three details are deliberate:

• A conversation is pinned to the token that started it. Reuse someone else's session_id with a different token and you get 403.
• Tool events are deduplicated. Strands streams a tool call in pieces while the model writes its arguments; without the announced set you'd see the same tool announced a dozen times.
• Charts never pass through the model. The broker gives the model a short chart_id; this server fetches the image bytes on a side route and hands them to the client.

chat.py, a terminal client.

# chat.py
"""Talk to the agent as a persona:  python chat.py support   |   python chat.py analyst"""
import json
import subprocess
import sys
import uuid
from pathlib import Path

import httpx

AGENT = "http://localhost:8080"
persona = sys.argv[1] if len(sys.argv) > 1 else "support"
token = subprocess.check_output([sys.executable, "identity/idp.py", "token", persona], text=True).strip()
session = uuid.uuid4().hex  # a new conversation, so tools are discovered afresh
Path("received").mkdir(exist_ok=True)

print(f"Signed in as {persona}. Ctrl+C to quit.")
while True:
    try:
        message = input("\nyou> ").strip()
    except (EOFError, KeyboardInterrupt):
        break
    if not message:
        continue
    print("agent> ", end="", flush=True)
    with httpx.stream("POST", f"{AGENT}/chat", timeout=300,
                      headers={"Authorization": f"Bearer {token}"},
                      json={"session_id": session, "message": message}) as r:
        for line in r.iter_lines():
            if not line.startswith("data: ") or line == "data: [DONE]":
                continue
            event = json.loads(line[6:])
            if "token" in event:
                print(event["token"], end="", flush=True)
            elif "tool" in event:
                print(f"\n  [tool: {event['tool']}]", flush=True)
            elif "chart" in event:
                png = httpx.get(AGENT + event["chart"]).content
                path = Path("received") / (event["chart"].split("/")[-1] + ".png")
                path.write_bytes(png)
                print(f"\n  [chart saved to {path}]", flush=True)
    print()

Step 7: A sandbox for model-written code#

An analyst asks "what were total sales by region in March?" No fixed tool answers that, and pasting a hundred orders into the prompt is slow, expensive and unreliable: models are poor at long arithmetic. Writing groupby('region')['total'].sum() and running it is exact. But model-written code is untrusted code, so it runs in a container that is created for one job and destroyed after.

The image has pandas and matplotlib, runs as the unprivileged nobody user, and sets MPLCONFIGDIR. In the workshop, matplotlib's "cache directory is not writable" warning leaked into every tool result and the model read it as input. One environment variable removes it.

# sandbox/Dockerfile
FROM python:3.12-slim
RUN pip install --no-cache-dir pandas==2.2.3 matplotlib==3.9.2
# Matplotlib wants a writable cache dir; point it at the tmpfs so its
# warning never ends up in the tool result (and the model's prompt).
ENV MPLCONFIGDIR=/work/.mpl MPLBACKEND=Agg
COPY runner.py /runner.py
USER 65534:65534
WORKDIR /work
ENTRYPOINT ["python", "/runner.py"]

The runner inside the container reads one job from standard input, so no files are shared with the host at all. It writes the rows and the code into a scratch folder, runs them, and prints one line of JSON.

# sandbox/runner.py
"""Runs inside the throwaway sandbox container.

Reads {"code": ..., "rows": [...]} on stdin, writes orders.json and main.py
into a scratch directory, runs the code, and prints one JSON line back:
{"exit_code", "stdout", "chart_png_b64"}.
"""
import base64
import json
import os
import signal
import subprocess
import sys
from pathlib import Path

# Small models often forget imports, so the usual ones are already there.
PREAMBLE = "import pandas as pd\nimport matplotlib.pyplot as plt\n"

job = json.load(sys.stdin)
work = Path("/work")  # the only writable place: a small in-memory tmpfs
(work / "orders.json").write_text(json.dumps(job["rows"]))
(work / "main.py").write_text(PREAMBLE + job["code"])

# Output goes to a file, not a pipe: stray child processes (a fork bomb, say)
# can hold a pipe open and make us wait forever. The code runs in its own
# process group so we can kill everything it started in one go.
with open(work / "out.txt", "w") as out:
    proc = subprocess.Popen([sys.executable, "main.py"], cwd=work, stdout=out,
                            stderr=subprocess.STDOUT, start_new_session=True)
    timed_out = False
    try:
        proc.wait(timeout=20)
    except subprocess.TimeoutExpired:
        timed_out = True
    try:
        os.killpg(proc.pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    proc.wait()

exit_code = proc.returncode
output = (work / "out.txt").read_text()
if timed_out:
    output += "\nStopped: the code ran for more than 20 seconds."
elif exit_code < 0:
    output += f"\nStopped by signal {-exit_code} (9 usually means it hit the memory limit)."
chart = work / "chart.png"
print(json.dumps({
    "exit_code": exit_code,
    "stdout": output[-4000:],
    "chart_png_b64": base64.b64encode(chart.read_bytes()).decode() if chart.exists() else None,
}))

Two parts of the runner exist because my first version failed:

• The PREAMBLE. The 3B model's first attempt forgot import pandas. Pre-importing the usual libraries is a kindness to small models and costs nothing.
• Output goes to a file, and the code runs in its own process group. My first runner captured output through a pipe. A fork bomb's child processes kept that pipe open after the main process died, so the runner sat waiting for its full timeout. Writing to a file and killing the whole process group afterwards made the same attack end in about a second.

The broker is the MCP server that owns run_python. It fetches a scoped, read-only slice of orders (note mode=ro on the SQLite connection), starts the container with every door shut, and returns a summary.

# code_broker/server.py
"""run_python: model-written pandas, executed in a throwaway, offline container."""
import base64
import json
import os
import re
import sqlite3
import subprocess
import uuid
from pathlib import Path

from mcp.server.mcpserver import MCPServer
from mcp.server.transport_security import TransportSecuritySettings
from starlette.requests import Request
from starlette.responses import FileResponse, Response

ROOT = Path(__file__).parent.parent
DB = ROOT / "shop" / "shop.db"
CHARTS = ROOT / "charts"
CHARTS.mkdir(exist_ok=True)
IMAGE = "shop-sandbox:latest"
RUNTIME = os.environ.get("SANDBOX_RUNTIME")  # e.g. "runsc" for gVisor, "kata-fc" for Firecracker

QUARTERS = {"2026-Q1": ("2026-01-01", "2026-03-31")}

mcp = MCPServer("code-broker")


def _fetch_rows(quarter: str, region: str | None) -> list[dict]:
    """A scoped, read-only slice. The broker can read orders; it can never write them."""
    start, end = QUARTERS[quarter]
    sql = "SELECT order_id, product, quantity, total, status, order_date, region FROM orders WHERE order_date BETWEEN ? AND ?"
    args: list = [start, end]
    if region:
        sql += " AND region = ?"
        args.append(region)
    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    with conn:
        return [dict(r) for r in conn.execute(sql, args)]


def _run_in_sandbox(code: str, rows: list[dict]) -> dict:
    name = f"sandbox-{uuid.uuid4().hex[:8]}"
    cmd = [
        "docker", "run", "--rm", "-i", "--name", name,
        "--network", "none",                 # no route out, not even DNS
        "--read-only",                       # nothing on disk can change...
        "--tmpfs", "/work:rw,size=64m,uid=65534,gid=65534,mode=0700",  # ...except a scratch dir
        "--cap-drop", "ALL", "--security-opt", "no-new-privileges",
        "--pids-limit", "64", "--memory", "512m", "--cpus", "1",
    ]
    if RUNTIME:
        cmd += ["--runtime", RUNTIME]
    cmd.append(IMAGE)
    try:
        proc = subprocess.run(cmd, input=json.dumps({"code": code, "rows": rows}),
                              capture_output=True, text=True, timeout=40)
    except subprocess.TimeoutExpired:
        subprocess.run(["docker", "kill", name], capture_output=True)
        return {"exit_code": -1, "stdout": "Timed out after 40 seconds.", "chart_png_b64": None}
    if proc.returncode != 0 and not proc.stdout.strip():
        return {"exit_code": proc.returncode, "stdout": proc.stderr[-2000:], "chart_png_b64": None}
    return json.loads(proc.stdout.strip().splitlines()[-1])


@mcp.tool()
def run_python(code: str, quarter: str = "2026-Q1", region: str | None = None) -> dict:
    """Run Python (pandas, matplotlib) over one quarter of orders in an offline sandbox.

    The orders are in ./orders.json with fields order_id, product, quantity, total,
    status, order_date (YYYY-MM-DD) and region. pandas (pd) and matplotlib.pyplot
    (plt) are already imported; load the data with pd.read_json("orders.json").
    Print the numbers you need. To show a chart, save it to chart.png.
    Only quarter "2026-Q1" exists; filter to a month in code, for example:

        df = pd.read_json("orders.json", dtype={"order_date": str})
        march = df[df.order_date.str.startswith("2026-03")]
        print(march.groupby("region")["total"].sum().round(2))
    """
    if quarter not in QUARTERS:
        return {"error": f"Unknown quarter {quarter!r}. Available: {', '.join(QUARTERS)}"}
    rows = _fetch_rows(quarter, region)
    result = _run_in_sandbox(code, rows)
    png = result.pop("chart_png_b64", None)
    if png:
        chart_id = f"c_{uuid.uuid4().hex[:12]}"
        (CHARTS / f"{chart_id}.png").write_bytes(base64.b64decode(png))
        result["chart_id"] = chart_id  # the model gets an ID, never the image bytes
    result["row_count"] = len(rows)
    return result


@mcp.custom_route("/charts/{chart_id}", methods=["GET"])
async def get_chart(request: Request) -> Response:
    """Side channel: the agent's server fetches chart bytes here, outside the LLM."""
    chart_id = request.path_params["chart_id"]
    if not re.fullmatch(r"c_[0-9a-f]{12}", chart_id):  # no "../" tricks
        return Response(status_code=404)
    path = CHARTS / f"{chart_id}.png"
    if not path.exists():
        return Response(status_code=404)
    return FileResponse(path, media_type="image/png")


if __name__ == "__main__":
    mcp.run(
        "streamable-http",
        host="127.0.0.1",
        port=8102,
        # Keep DNS-rebinding protection on; the gateway forwards "Host: localhost".
        transport_security=TransportSecuritySettings(
            allowed_hosts=["localhost", "localhost:*", "127.0.0.1", "127.0.0.1:*"]
        ),
    )

Each docker run flag closes one door:

Two more details. The tool's docstring includes a three-line example of filtering to a month. With it, the 3B model got March right; without it, it filtered on the current month. Small models copy examples far better than they follow descriptions. And the chart route checks the ID's exact format, because my first version only checked the prefix, so a crafted c_../../ could have walked out of the charts folder.

Don't take the table on trust. This script throws hostile code at the sandbox through the gateway, as the analyst:

# check_sandbox.py
"""Throw hostile code at run_python and see what the sandbox allows."""
import json
import subprocess
import sys

from strands.tools.mcp import MCPClient

ATTEMPTS = {
    "reach the internet": "import urllib.request\nprint(urllib.request.urlopen('https://example.com', timeout=5).status)",
    "write outside /work": "open('/etc/evil', 'w').write('x')",
    "read secrets from env": "import os\nprint({k: v for k, v in os.environ.items() if 'KEY' in k or 'TOKEN' in k} or 'no secrets here')",
    "who am I": "import os\nprint('uid', os.getuid())",
    "fork bomb": "import os\nwhile True: os.fork()",
    "the real job": (
        "import pandas as pd, matplotlib.pyplot as plt\n"
        "df = pd.read_json('orders.json')\n"
        "s = df.groupby('region')['total'].sum().round(2).sort_values(ascending=False)\n"
        "print(s.to_string())\n"
        "s.plot(kind='bar', title='Q1 2026 sales by region'); plt.tight_layout(); plt.savefig('chart.png')"
    ),
}

tok = subprocess.check_output([sys.executable, "identity/idp.py", "token", "analyst"], text=True).strip()
with MCPClient(url="http://localhost:3000/mcp", headers={"Authorization": f"Bearer {tok}"}) as c:
    for name, code in ATTEMPTS.items():
        r = c.call_tool_sync(name, "code_run_python", {"code": code})
        out = json.loads(r["content"][0]["text"])
        last = out.get("stdout", "").strip().splitlines()[-1:] or [""]
        print(f"{name:22} exit={out.get('exit_code')}  {last[0][:80]}  {('chart_id=' + out['chart_id']) if 'chart_id' in out else ''}")

reach the internet     exit=1  urllib.error.URLError: <urlopen error [Errno -3] Temporary failure in name resol
write outside /work    exit=1  OSError: [Errno 30] Read-only file system: '/etc/evil'
read secrets from env  exit=0  {'GPG_KEY': '7169605F62C751356D054A26A821E680E5FA6305'}
who am I               exit=0  uid 65534
fork bomb              exit=-9  Stopped by signal 9 (9 usually means it hit the memory limit).
the real job           exit=0  South      11809.50  chart_id=c_37c0faffd188

Every attack failed, the real job worked, all six runs took about 12 seconds together, and docker ps -a showed no leftover containers. The one "secret" it found, GPG_KEY, is the public fingerprint the official Python image uses to verify its own download, not a credential. Nothing of ours is in there.

Docker is not Firecracker. A container shares the host's kernel, and a kernel exploit is a real escape route. That's why the workshop used microVMs. The broker reads SANDBOX_RUNTIME, so you can upgrade without touching code: install gVisor and set SANDBOX_RUNTIME=runsc, or on a Kata-enabled host use its Firecracker runtime. For a laptop, the flags above are a sensible floor; for production, use a VM boundary.

Order your timeouts from the inside out. This one bit me under load. agentgateway gives a tool call about 10 seconds by default, while my broker allowed 60. When the machine was busy, the gateway gave up on a slow run and reported upstream call timeout, but the broker's container kept running behind it. The fix is to nest the limits so each layer fails cleanly before the one outside it: the code gets 20 seconds inside the sandbox, the broker gives up at 40, and the gateway's timeout policy (in the config above) allows 60.

Build the image once:

docker build -t shop-sandbox:latest sandbox

Step 8: Run everything and talk to it#

This script starts Jaeger, both MCP servers, the gateway and the agent, writing logs to logs/:

#!/usr/bin/env bash
# run.sh
# ./run.sh up | down | logs     Start or stop every service in the background.
set -euo pipefail
cd "$(dirname "$0")"
mkdir -p logs .pids
export AGENT_GATEWAY_KEY="${AGENT_GATEWAY_KEY:-sk-shop-agent-local}"   # the agent's key for the model gate

start() {  # start <name> <command...>
  local name=$1; shift
  nohup "$@" > "logs/$name.log" 2>&1 &
  echo $! > ".pids/$name"
  echo "started $name (pid $!)"
}

case "${1:-}" in
  up)
    docker run -d --rm --name shop-jaeger \
      -p 127.0.0.1:16686:16686 -p 127.0.0.1:4317:4317 -p 127.0.0.1:4318:4318 \
      jaegertracing/jaeger:2.21.0 >/dev/null && echo "started jaeger (UI on http://localhost:16686)"
    start orders  uv run python orders_mcp/server.py
    start broker  uv run python code_broker/server.py
    start gateway ./bin/agentgateway -f gateway/config.yaml
    sleep 3
    start agent   env MODEL_API_KEY="$AGENT_GATEWAY_KEY" \
                      OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 OTEL_SERVICE_NAME=shop-agent \
                      uv run --directory agent python server.py
    ;;
  down)
    for f in .pids/*; do [ -e "$f" ] && kill "$(cat "$f")" 2>/dev/null || true; rm -f "$f"; done
    docker stop shop-jaeger >/dev/null 2>&1 || true
    echo "stopped"
    ;;
  logs)
    tail -n 20 logs/*.log
    ;;
  *) echo "usage: ./run.sh up | down | logs"; exit 1 ;;
esac

chmod +x run.sh && ./run.sh up
uv run python chat.py support

Here's what my runs looked like, with the local 3B model:

you> Where is my order ORD-1001?
  [tool: orders_lookup_order]
agent> Your order ORD-1001 is currently in the shipped status. It contains a
Laptop Pro 15 and was placed on March 28, 2026. The region for this order is West.

you> Is order ORD-99999 on its way?
  [tool: orders_lookup_order]
agent> I'm sorry, but there is no order with the ID ORD-99999. Could you please
check the order ID and try again?

you> Chart last month's sales by region.
agent> I can help you look up orders from the past month to analyze their regions,
but I don't have a direct function to chart sales data...

The last answer is the gate working: the support persona doesn't have run_python, so the agent says so instead of failing. Now sign in as the analyst:

uv run python chat.py analyst

you> What were total sales by region for March 2026? Use run_python.
  [tool: code_run_python]
agent> ... South: $4,089.73 ... West: $14,469.77 ...

Those match a direct SQL query against the database to the cent. When the model saves a chart, it lands in received/ as a PNG.

Be honest with yourself about the model. Once the timeouts were fixed, the plumbing never failed in my tests; the 3B model sometimes did. It wrote code with a missing import, then code that filtered on the wrong month, before the docstring example fixed it. Asked for a bar chart, it computed the right numbers but wrote ![](chart.png) into its reply instead of saving the file. (I verified the chart path separately: a real tool result, the chart ID picked out by server.py, and the PNG served through the agent.) For anything beyond lookups, point shop-fast at a stronger model. The point of the gateway is that this is a two-line change.

Step 9: See everything in one trace#

Open Jaeger at http://localhost:16686, pick the shop-agent service and open a trace. Here is one request from my run, longest spans first:

shop-agent    POST /chat                       42.06 s
shop-agent    invoke_agent Strands Agents      42.03 s   model: shop-fast    tokens 763 in / 60 out
shop-agent    chat                             26.00 s   model: shop-fast
agentgateway  POST /v1/chat/completions        25.99 s   model: qwen2.5:3b
shop-agent    chat                             16.01 s   model: shop-fast
agentgateway  POST /v1/chat/completions        15.65 s   model: qwen2.5:3b
agentgateway  POST localhost:11434             14.87 s   http.status: 200
shop-agent    execute_tool orders_lookup_order  0.01 s
shop-agent    MCP send tools/call orders_lookup_order   0.01 s

The agent's spans say shop-fast, and the gateway's span nested inside says qwen2.5:3b: the alias rewrite, seen from inside one trace. You can also see where the time goes. The tool call took 10 milliseconds; the model took everything else.

Getting the gateway's spans into the agent's trace needed one line in server.py, HTTPXClientInstrumentor().instrument(), which adds the W3C traceparent header to the agent's model calls. Without it, the gateway's spans showed up as separate traces. The tool gate's spans still arrive as their own traces, because the MCP client uses a different HTTP library. The agent's own MCP send spans cover those calls in the meantime.

What's different from production#

This is a faithful laptop version, not a production system. If you take it further:

• Use a real identity provider (Cognito, Entra ID, Keycloak, Auth0) instead of idp.py. The gateway config barely changes: point jwks at the provider's URL instead of a file.
• Put a VM boundary around the sandbox (Firecracker or gVisor), and keep the broker's Docker access away from anything else. On a laptop the broker can start containers; in a cluster it should only be able to request sandboxes.
• Keep secrets in a secret store, not in environment variables in a script.
• Run it on Kubernetes when you need scale, which is exactly what the AWS workshop shows. The ideas carry over unchanged; only the packaging changes.

Clean up#

./run.sh down
docker image rm shop-sandbox:latest jaegertracing/jaeger:2.21.0

That's the whole platform: an agent that holds no provider credentials, a gateway that decides which model it may use and which tools each person may touch, a sandbox that runs untrusted code with every door closed, and a trace that shows all of it. None of it needed the agent to be clever. It needed the platform to be strict.

This post follows ideas from AWS's Secure AI Agents on Amazon EKS workshop. The code is my own and was written and tested for this post; the companion piece, Securing AI Agents on Amazon EKS, covers the workshop itself.