#!/usr/bin/env python3 """Anthropic Managed Agents (beta): create agent -> environment -> session -> user.message -> stream SSE -> cleanup. STATUS: LIVE_VERIFIED 2026-09-18 — the identical HTTP sequence (env → agent → session with string-cents budget → user.message → SSE stream → cleanup) ran via scripts/live.py (tmp-live/platform-anthropic/agents-probes-2.json); this file itself was not executed. Model: claude-haiku-4-5-20251001 Beta header: anthropic-beta: managed-agents-2026-04-01 Budget: max budget minimal -- session budget capped at 5 US cents ({"amount": "5", "currency": "USD"}), prompt "Reply with OK.", no tools (nothing executes in the sandbox). Names prefixed "atlas-platform-agent". Uses scripts.live.anthropic_request (auth injected from .env, every call logged to reports/live-requests.jsonl, keys masked). SSE is read with stream=True, which returns a line iterator over the HTTP response. Every resource created here is removed in `finally`: DELETE session (after interrupt if still running), POST /v1/agents/{id}/archive (agents have no delete endpoint), DELETE /v1/environments/{id} (falls back to archive). Run: .venv/bin/python examples/anthropic/agents/create_agent_session_stream.py Docs: https://platform.claude.com/docs/en/managed-agents/quickstart https://platform.claude.com/docs/en/managed-agents/events-and-streaming """ from __future__ import annotations import json import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(ROOT)) from scripts.live import anthropic_request # noqa: E402 BETA = "managed-agents-2026-04-01" MODEL = "claude-haiku-4-5-20251001" TS = int(time.time()) # Haiku 4.5 list price (USD per 1M tokens) used only for the est_cost_usd log field. IN_PRICE, OUT_PRICE = 1.0, 5.0 def ant(method: str, path: str, body=None, note: str = "", **kw): st, resp, _ = anthropic_request(method, path, body, beta=BETA, note=note, **kw) if st >= 300: raise RuntimeError(f"{method} {path} -> {st}: {json.dumps(resp)[:500] if not isinstance(resp, bytes) else resp[:500]!r}") return resp def main() -> int: agent_id = env_id = session_id = None try: # 1) Agent: versioned resource; no tools so nothing runs in the sandbox. agent = ant("POST", "/v1/agents", { "name": f"atlas-platform-agent-{TS}", "model": MODEL, "system": "You are a terse test agent. Reply with OK.", "metadata": {"origin": "api-atlas"}, }, note="managed-agents create agent") agent_id = agent["id"] print("agent:", agent_id, "version", agent["version"]) # 2) Environment: cloud sandbox with limited egress (least privilege). env = ant("POST", "/v1/environments", { "name": f"atlas-platform-agent-env-{TS}", "config": {"type": "cloud", "networking": {"type": "limited", "allowed_hosts": ["example.com"]}}, }, note="managed-agents create environment") env_id = env["id"] print("environment:", env_id) # 3) Session with a hard 5-cent list-cost budget (only attachable at creation). session = ant("POST", "/v1/sessions", { "agent": agent_id, "environment_id": env_id, "title": f"atlas-platform-agent-session-{TS}", "budget": {"type": "limit", "max_list_cost": {"amount": "5", "currency": "USD"}}, }, note="managed-agents create session") session_id = session["id"] print("session:", session_id, "status", session["status"]) # 4) Open the SSE stream BEFORE sending the message (only later events are delivered). st, lines, _hdrs = anthropic_request( "GET", f"/v1/sessions/{session_id}/events/stream", beta=BETA, extra_headers={"accept": "text/event-stream"}, content_type=None, stream=True, timeout=200, note="managed-agents stream events") if st >= 300: raise RuntimeError(f"stream -> {st}: {lines}") # 5) Send the user message (returns as soon as the event is queued). ant("POST", f"/v1/sessions/{session_id}/events", { "events": [{"type": "user.message", "content": [{"type": "text", "text": "Reply with OK."}]}], }, note="managed-agents send user.message") # 6) Consume the stream until the session goes idle (or errors / 180 s deadline). deadline = time.time() + 180 seen: list[str] = [] usage = None for line in lines: if time.time() > deadline: print("timeout waiting for session.status_idle") break if not line.startswith("data:"): continue ev = json.loads(line[5:].strip()) seen.append(ev["type"]) if ev["type"] == "agent.message": print("[agent.message]", "".join(c.get("text", "") for c in ev["content"] if c.get("type") == "text")) elif ev["type"] == "session.usage": usage = ev.get("usage") print("[session.usage] list_cost", (usage or {}).get("list_cost"), "active_seconds", (usage or {}).get("active_seconds")) elif ev["type"] == "session.status_idle": print("[session.status_idle] stop_reason", ev["stop_reason"]["type"]) break elif ev["type"] == "session.error": print("[session.error]", ev["error"].get("type"), ev["error"].get("retry_status", {}).get("type")) break else: print(f"[{ev['type']}]") print("event sequence:", " -> ".join(seen)) # 7) Final snapshot; log an estimated cost from token usage. fin = ant("GET", f"/v1/sessions/{session_id}", note="managed-agents get session") u = fin.get("usage") or {} est = (u.get("input_tokens", 0) * IN_PRICE + u.get("output_tokens", 0) * OUT_PRICE) / 1e6 print(json.dumps({"status": fin["status"], "usage": u, "est_cost_usd_tokens_only": round(est, 6)}, indent=2)) return 0 finally: # Cleanup: every resource this script created. try: if session_id: s = ant("GET", f"/v1/sessions/{session_id}") if s["status"] == "running": ant("POST", f"/v1/sessions/{session_id}/events", {"events": [{"type": "user.interrupt"}]}) for _ in range(30): time.sleep(2) if ant("GET", f"/v1/sessions/{session_id}")["status"] != "running": break ant("DELETE", f"/v1/sessions/{session_id}", note="cleanup delete session") print("deleted session", session_id) if agent_id: ant("POST", f"/v1/agents/{agent_id}/archive", note="cleanup archive agent") print("archived agent", agent_id) if env_id: try: ant("DELETE", f"/v1/environments/{env_id}", note="cleanup delete environment") print("deleted environment", env_id) except RuntimeError: ant("POST", f"/v1/environments/{env_id}/archive", note="cleanup archive environment") print("archived environment", env_id) except Exception as exc: # noqa: BLE001 print("cleanup error:", exc) if __name__ == "__main__": raise SystemExit(main())