Running agents in production: undo, cost, observability, hosting, security

options = ClaudeAgentOptions(cwd=PROJECT, enable_file_checkpointing=True, permission_mode="acceptEdits",
                             extra_args={"replay-user-messages": None},    # (1) needed to receive the UUIDs
                             max_turns=4)
async with ClaudeSDKClient(options) as client:
    await client.query("Append a line '## Changelog' followed by '- demo edit' to README.md")
    async for message in client.receive_response():
        if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
            checkpoint_id = message.uuid                                  # (2) first user message = "before anything happened"
        if isinstance(message, ResultMessage):
            session_id = message.session_id

async with ClaudeSDKClient(ClaudeAgentOptions(cwd=PROJECT, enable_file_checkpointing=True, resume=session_id)) as client:
    await client.query("")                                                # (3) empty prompt just opens the connection
    async for _ in client.receive_response():
        await client.rewind_files(checkpoint_id)                          # (4) restore
        break

File changed by agent: True
Tail after edit: 'up with a known SQL injection bug\n\n## Changelog\n- demo edit\n'
Rewound to checkpoint 53cb3f96-3508-4865-91c5-454a190b25ac
File identical to original after rewind: True

if isinstance(message, AssistantMessage) and not message.parent_tool_use_id:   # (1) main loop only
    if message.message_id not in seen_ids:                                     # (2) count each API response once
        seen_ids.add(message.message_id); steps += 1
        u = message.usage or {}
        input_tokens += u.get("input_tokens", 0)
elif isinstance(message, ResultMessage):
    u = message.usage or {}                                                    # (3) totals for the main loop
    for model, mu in (message.model_usage or {}).items():                      # (4) per model, including subagents
        print(f"{model}: cost=${mu['costUSD']:.4f} contextWindow={mu['contextWindow']}")

step 1: message_id=msg_011Cg5Rkrj... input=2 cache_read=0 cache_create=24941 stop=None
step 2: message_id=msg_011Cg5Rm2X... input=2 cache_read=24941 cache_create=129 stop=None

result.usage: input=4 output=85 cache_read=24941 cache_create=25070
result.total_cost_usd: $0.1061  (client-side estimate)
result.duration_ms=3877 duration_api_ms=3655 num_turns=2
result.model_usage[claude-sonnet-5]: cost=$0.1061 contextWindow=1000000 costBasis=list

async with ClaudeSDKClient(options) as client:
    await client.query("Reply with exactly: ready")
    async for message in client.receive_response():
        if isinstance(message, SystemMessage) and message.subtype == "init":
            d = message.data            # claude_code_version, model, permissionMode, tools, slash_commands, skills, mcp_servers, plugins
    usage  = await client.get_context_usage()
    status = await client.get_mcp_status()
    await client.set_model("claude-haiku-5-5")
    await client.query("Reply with exactly: done")

init.claude_code_version: 2.1.281
init.model: claude-sonnet-5
init.tools: [..., 'mcp__claude-code-docs__query_docs_filesystem_claude_code_docs', 'mcp__claude-code-docs__search_claude_code_docs', 'mcp__claude-code-docs__submit_feedback']
init.slash_commands: ['security-check', 'verify', 'code-review', 'clear', 'compact', 'context', 'effort', 'model', 'usage', ...]
init.mcp_servers: [{'name': 'claude-code-docs', 'status': 'connected', 'source': 'dynamic'}]

get_context_usage: 25361/1000000 tokens (3.0%), model=claude-sonnet-5, autocompact=True
  categories: [('System tools', 11446), ('MCP tools (deferred)', 1721), ('System tools (deferred)', 16341), ('Memory files', 64), ('Skills', 5576), ('Messages', 8275), ('Autocompact buffer', 33000), ('Free space', 941639)]
get_mcp_status: claude-code-docs -> connected tools=3

Turn 1 model: claude-sonnet-5
Turn 2 model (after set_model): claude-haiku-5-5

received = {"traces": [], "metrics": [], "logs": []}

class Collector(BaseHTTPRequestHandler):
    """Accepts OTLP over HTTP with JSON encoding and stores the payloads."""
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        signal = self.path.rsplit("/", 1)[-1]                    # /v1/traces -> traces
        if signal in received:
            received[signal].append(json.loads(body))
        self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
        self.wfile.write(b"{}")

OTEL_ENV = {
    "CLAUDE_CODE_ENABLE_TELEMETRY": "1",
    "CLAUDE_CODE_ENHANCED_TELEMETRY_BETA": "1",             # (1) traces are beta
    "OTEL_TRACES_EXPORTER": "otlp", "OTEL_METRICS_EXPORTER": "otlp", "OTEL_LOGS_EXPORTER": "otlp",
    "OTEL_EXPORTER_OTLP_PROTOCOL": "http/json",             # (2) JSON for the demo; use http/protobuf against a real collector
    "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318",
    "OTEL_SERVICE_NAME": "tutorial-agent",
    "OTEL_RESOURCE_ATTRIBUTES": "deployment.environment=demo,tenant.id=acme",
    "OTEL_METRIC_EXPORT_INTERVAL": "1000", "OTEL_LOGS_EXPORT_INTERVAL": "1000", "OTEL_TRACES_EXPORT_INTERVAL": "1000",  # (3) flush fast
    "CLAUDE_CODE_OTEL_DIAG_STDERR": "1",                    # (4) surface exporter errors on stderr
    "NO_PROXY": "127.0.0.1,localhost",                      # (5) don't send localhost traffic through a corporate proxy
}

async for message in query(
    prompt="Read README.md and reply with its first heading only.",
    options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read"], max_turns=3,
                               env=OTEL_ENV, stderr=stderr_lines.append),      # Python merges env; TS needs { ...process.env, ...otelEnv }
):
    ...
await asyncio.sleep(8)                                                          # let the final export intervals fire

[result] success: '# Sample project' cost=$0.0186

received: 0 trace batches, 4 metric batches, 4 log batches

spans (name -> count): {}

metrics: ['claude_code.active_time.total', 'claude_code.cost.usage', 'claude_code.session.count', 'claude_code.token.usage']

log events (name -> count): {'managed_settings_resolved': 1, 'user_prompt': 1, 'plugin_loaded': 2, 'tool_decision': 2, 'api_request': 3, 'tool_result': 2, 'assistant_response': 1}

# ephemeral entrypoint
async for message in query(prompt=os.environ["TASK_PROMPT"], options=ClaudeAgentOptions(max_turns=20)):
    print(message)

# hybrid: hydrate from a shared store
async for message in query(prompt=user_input, options=ClaudeAgentOptions(resume=session_id, session_store=session_store)):
    ...

from claude_agent_sdk import fold_session_summary
from claude_agent_sdk.testing import run_session_store_conformance

class SqliteSessionStore:
    """Two required methods (append, load) plus the optional ones, on one SQLite file."""

    async def append(self, key, entries):
        pk, sid, sub = key["project_key"], key["session_id"], key.get("subpath") or ""
        with self.db:
            seen = {uuids already stored for this key}                            # (1) a retried batch may re-deliver entries
            seq = (max stored seq) + 1
            for e in entries:
                if e.get("uuid") in seen: continue
                self.db.execute("INSERT INTO entries VALUES (?,?,?,?,?,?)", (pk, sid, sub, seq, e.get("uuid"), json.dumps(e)))
                seq += 1
            if not sub:                                                           # (2) summary sidecar for the main transcript only
                prev = {stored summary for (pk, sid)} or None
                summary = fold_session_summary(prev, key, entries)                # (3) SDK helper: fold new entries in
                self.db.execute("INSERT OR REPLACE INTO summaries VALUES (?,?,?,?)", (pk, sid, now_ms, json.dumps(summary["data"])))

    async def load(self, key):
        rows = SELECT entry ... ORDER BY seq
        return [json.loads(r[0]) for r in rows] if rows else None                 # (4) None, not [], when unknown

    async def list_sessions(self, project_key): ...        # [{"session_id", "mtime"}], newest first
    async def list_session_summaries(self, project_key): ...
    async def delete(self, key): ...                       # main key: cascade to subkeys and the summary
    async def list_subkeys(self, key): ...

# 1. the SDK's contract tests, each on a fresh empty store
await run_session_store_conformance(lambda: SqliteSessionStore(":memory:"))

# 2. host A writes a session into the shared store (its own config dir = its own local transcripts)
store = SqliteSessionStore(db_path)
async for m in query(prompt="Remember the code word OSPREY. Reply only: noted",
                     options=ClaudeAgentOptions(cwd=PROJECT, tools=[], max_turns=2, session_store=store,
                                                env={"CLAUDE_CONFIG_DIR": host_a})):
    if isinstance(m, ResultMessage): session_id = m.session_id

# 3. host B (a different config dir, so NO local transcript) resumes from the store
async for m in query(prompt="What was the code word? One word.",
                     options=ClaudeAgentOptions(cwd=PROJECT, tools=[], max_turns=2, session_store=store,
                                                resume=session_id, env={"CLAUDE_CONFIG_DIR": host_b})):
    ...

=== conformance suite ===
conformance: PASSED

host A: session 7841b307-51d9-4f01-852f-57f0d485d90f -> 'noted'
store now holds 16 transcript entries for that session (project_key='-tmp-demo-project')
summary sidecar: {"is_sidechain": false, "created_at": 1792135295404, "cwd": "/tmp/.../demo/project", "first_prompt": "Remember the code word OSPREY. Rep
host B: resumed 7841b307-51d9-4f01-852f-57f0d485d90f -> 'OSPREY'
host B had no local transcript: []

def confine_to(root: Path):
    """PreToolUse hook: deny any file tool whose path resolves outside the tenant's working directory."""
    async def hook(input_data, tool_use_id, context):
        target = input_data["tool_input"].get("file_path") or input_data["tool_input"].get("path")
        if target and not Path(target).resolve().is_relative_to(root.resolve()):
            print(f"  [hook] DENY {input_data['tool_name']}({target})")
            return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny",
                                           "permissionDecisionReason": f"{target} is outside the tenant workspace"}}
        return {}
    return hook

async def ask(work, config, prompt, sources, **extra):
    options = ClaudeAgentOptions(
        cwd=str(work),
        setting_sources=sources,                                      # (1) [] = no host/project settings, no CLAUDE.md
        env={"CLAUDE_CONFIG_DIR": str(config),                        # (2) per-tenant config dir: transcripts, state
             "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"},                 # (3) auto memory would load regardless of (1)
        max_turns=3, **extra)
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            return (m.result or "").strip(), m.session_id, len(m.permission_denials)

# each tenant: work/CLAUDE.md says "The internal project codename is BLUEBIRD" (acme) / "REDWOOD" (globex)
q = "Without using any tools: what is the internal project codename? Reply with the codename only, or 'unknown'."
await ask(a_work, a_config, q, sources=[], tools=[])                  # 1. isolation
await ask(a_work, a_config, q, sources=["project"], tools=[])         # 2. opt in to the project's CLAUDE.md
probe = f"Read the file {a_work}/notes.txt and quote its contents in one line. If the read fails, say exactly why."
await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"])                            # 3. cwd only
await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"],
          hooks={"PreToolUse": [HookMatcher(matcher="Read|Edit|Write|Glob|Grep", hooks=[confine_to(b_work)])]})  # 3b. hook

=== 1. settings isolation: setting_sources=[] means CLAUDE.md is NOT injected ===
acme   -> 'unknown'
globex -> 'unknown'

=== 2. the same tenants with project settings loaded on purpose (setting_sources=['project']) ===
acme   -> 'BLUEBIRD'
globex -> 'REDWOOD'

=== 3. file isolation: does cwd alone bound what tools can reach? ===
globex, cwd only          -> I was able to read the file, but I want to flag something before quoting it: the path points to
`/tmp/ws-acme-xcw1vyki/...`, a different workspace sandbox (acme) than the one I'm running in (...globex). The file's content
is literally labeled "acme private notes" — i.e., it appears to be another tenant's private data that happens to be reachable
from this environment, not something that should cross into this session. [...]   (denials=0)

=== 3b. file isolation done right: a PreToolUse hook (or the OS sandbox) confines paths to the tenant root ===
  [hook] DENY Read(/tmp/ws-acme-xcw1vyki/work/notes.txt)
globex, with confine hook -> The read failed: the path `/tmp/ws-acme-xcw1vyki/work/notes.txt` is outside my current tenant
workspace (`/tmp/ws-globex-s6_uwjwa/work`), so the Read tool's access-control hook blocked it with the error
"is outside the tenant workspace."   (denials=1)

=== 4. transcript isolation: each tenant's sessions live under its own CLAUDE_CONFIG_DIR ===
acme: 2 transcript(s) under /tmp/ws-acme-xcw1vyki/config
globex: 4 transcript(s) under /tmp/ws-globex-s6_uwjwa/config
tenant B's config dir contains tenant A's session: False

async def one(i):
    async for m in query(prompt=f"Session {i}: read src/app.py and say in one sentence what find_user does.",
                         options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read"], max_turns=3, effort="low")):
        if isinstance(m, ResultMessage):
            return i, m.subtype, round(time.time() - t0, 1), m.total_cost_usd or 0

results = await asyncio.gather(*(one(i) for i in range(N)))       # N sessions at once, one subprocess each

session 0: success in 6.5s, $0.0421
session 1: success in 7.2s, $0.0421
session 2: success in 6.3s, $0.0420

3 sessions in parallel: wall=7.2s, peak claude processes=4, peak total RSS=1074 MiB (~268 MiB per session)
planning: agents per host = (host RAM - overhead) / per-session ceiling; at ~268 MiB a 16 GiB host fits ~53 idle-sized sessions

# app.py — one worker process, many sessions; sessions are resumed from the shared store on every request
app = FastAPI()
store = SqliteSessionStore(os.environ["SESSION_DB"])              # swap for Postgres/S3 in production

@app.post("/chat/{tenant}")
async def chat(tenant: str, body: ChatIn):
    work, config = tenant_dirs(tenant)                             # separate cwd + CLAUDE_CONFIG_DIR per tenant (5.2)
    opts = ClaudeAgentOptions(
        cwd=work, setting_sources=[], tools=["Read", "Grep", "Glob"],
        env={"CLAUDE_CONFIG_DIR": config, "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1", **OTEL_ENV},   # telemetry (section 4)
        hooks={"PreToolUse": [HookMatcher(matcher="Read|Glob|Grep", hooks=[confine_to(Path(work))])]},
        session_store=store, resume=body.session_id,               # hybrid pattern (5.1)
        max_turns=20, max_budget_usd=0.50)
    async for m in query(prompt=body.message, options=opts):
        if isinstance(m, ResultMessage):
            return {"session_id": m.session_id, "reply": m.result, "cost_usd": m.total_cost_usd}

# Dockerfile (illustrative)
FROM python:3.13-slim
RUN apt-get update && apt-get install -y --no-install-recommends bubblewrap socat && rm -rf /var/lib/apt/lists/*   # sandbox deps (section 6)
RUN pip install --no-cache-dir claude-agent-sdk fastapi uvicorn
RUN useradd -m agent
USER agent
WORKDIR /srv
COPY --chown=agent app.py .
ENV CLAUDE_CODE_DISABLE_AUTO_MEMORY=1
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]
# run with: --cap-drop ALL --read-only --tmpfs /tmp --pids-limit 256, and HTTPS only through your egress proxy

options = ClaudeAgentOptions(sandbox={
    "enabled": True,
    "autoAllowBashIfSandboxed": True,     # sandboxed commands need no prompt
    "failIfUnavailable": True,            # Python falls back to unsandboxed by default; TS stops by default
    "excludedCommands": ["docker *"],
    "allowUnsandboxedCommands": True,     # model may ask to leave the sandbox -> goes to can_use_tool
    "network": {"allowedDomains": ["api.github.com"], "allowLocalBinding": True},
})

stderr_lines = []
try:
    async for m in query(
        prompt="Run `echo sandbox-test` with bash and reply with its output only.",
        options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Bash"], max_turns=3,
                                   sandbox={"enabled": True, "autoAllowBashIfSandboxed": True, "failIfUnavailable": fail},
                                   stderr=stderr_lines.append),                      # (1) the only place sandbox warnings go
    ):
        if isinstance(m, ResultMessage):
            print(f"result: subtype={m.subtype} terminal_reason={m.terminal_reason!r} errors={m.errors}")
except ResultError as e:                                                             # (2) one-shot query() raises on an error result
    print(f"ResultError raised: {e}")

=== sandbox enabled, failIfUnavailable=True ===
ResultError raised: Claude Code returned an error result: Sandbox required but unavailable: sandbox is enabled but dependencies
are missing: bubblewrap (bwrap) not installed, socat not installed · install missing tools (e.g. apt install bubblewrap socat)
or see https://code.claude.com/docs/en/sandboxing. Set sandbox.failIfUnavailable=false to allow unsandboxed execution. (exit code: 1)
stderr lines mentioning sandbox (2):
    Error: sandbox required but unavailable: sandbox is enabled but dependencies are missing: bubblewrap (bwrap) not installed, ...
    sandbox.failIfUnavailable is set — refusing to start without a working sandbox.

=== sandbox enabled, failIfUnavailable=False (fall back unsandboxed, warn) ===
result: subtype=success terminal_reason='completed' errors=None
result text: 'sandbox-test'
stderr lines mentioning sandbox (2):
    ⚠ Sandbox disabled: sandbox is enabled but dependencies are missing: bubblewrap (bwrap) not installed, socat not installed ...
    Commands will run WITHOUT sandboxing. Network and filesystem restrictions will NOT be enforced.