#!/usr/bin/env python3 """Live probes for the Anthropic advanced-generation domain (API Atlas). Runs every probe once, saves sanitized raw responses under tmp-live/anthropic-advanced/, logs each call to reports/live-requests.jsonl with an estimated cost computed from usage. """ from __future__ import annotations import base64, json, os, sys, time, uuid, zlib, struct from pathlib import Path ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) from scripts import live # noqa: E402 OUT = ROOT / "tmp-live" / "anthropic-advanced" OUT.mkdir(parents=True, exist_ok=True) HAIKU = "claude-haiku-4-5-20251001" SONNET5 = "claude-sonnet-5" SONNET46 = "claude-sonnet-4-6" OPUS5 = "claude-opus-5" # USD per MTok: input, output, 5m write, 1h write, read PRICES = { HAIKU: (1, 5, 1.25, 2, 0.10), SONNET5: (2, 10, 2.5, 4, 0.20), SONNET46: (3, 15, 3.75, 6, 0.30), OPUS5: (5, 25, 6.25, 10, 0.50), } SUMMARY: list[dict] = [] TOTAL = [0.0] def cost(model: str, usage: dict | None) -> float: if not usage or model not in PRICES: return 0.0 i, o, w5, w1, r = PRICES[model] cc = usage.get("cache_creation") or {} w5t = cc.get("ephemeral_5m_input_tokens") w1t = cc.get("ephemeral_1h_input_tokens", 0) or 0 if w5t is None: w5t = usage.get("cache_creation_input_tokens") or 0 c = (usage.get("input_tokens", 0) * i + usage.get("output_tokens", 0) * o + w5t * w5 + w1t * w1 + (usage.get("cache_read_input_tokens") or 0) * r) / 1e6 if usage.get("speed") == "fast": c *= 2 return c def call(name: str, method: str, path: str, body=None, *, beta: str | None = None, model: str | None = None, stream: bool = False, data: bytes | None = None, content_type: str | None = "application/json", note: str = "") -> tuple[int, object, dict]: key = os.environ["ANTHROPIC_API_KEY"] headers = {"x-api-key": key, "anthropic-version": live.ANTHROPIC_VERSION} if content_type: headers["Content-Type"] = content_type if beta: headers["anthropic-beta"] = beta if body is not None: data = json.dumps(body).encode() t0 = time.time() st, out, hdrs = live._request(live.ANTHROPIC_BASE + path, method, headers, data, 180, stream) events = None if stream and st == 200: events = [l for l in out if l] out = events dt = time.time() - t0 usage = None if isinstance(out, dict): usage = out.get("usage") elif isinstance(out, list): # SSE lines for l in out: if l.startswith("data:"): try: d = json.loads(l[5:].strip()) except Exception: continue if d.get("type") == "message_start": usage = dict(d["message"].get("usage") or {}) elif d.get("type") == "message_delta" and usage is not None: usage["output_tokens"] = d.get("usage", {}).get("output_tokens", usage.get("output_tokens", 0)) mdl = model or (body or {}).get("model") if isinstance(body, dict) else model c = cost(mdl or "", usage) TOTAL[0] += c live.log_request("anthropic", method, path.split("?")[0], st, c, f"atlas-anthropic-advanced {name} {mdl or ''} beta={beta or '-'} {note}") rec = {"name": name, "method": method, "path": path, "status": st, "model": mdl, "beta": beta, "latency_s": round(dt, 2), "est_cost_usd": round(c, 6), "usage": usage, "headers": live.interesting_headers(hdrs)} if isinstance(out, bytes): rec["body_bytes"] = len(out) rec["body_preview"] = out[:200].decode("utf-8", "replace") else: rec["body"] = out live.save_sanitized(rec, OUT / f"{name}.json") SUMMARY.append({k: rec[k] for k in ("name", "status", "model", "beta", "est_cost_usd", "usage", "latency_s")}) err = out.get("error", {}).get("message") if isinstance(out, dict) and st >= 400 else "" print(f"[{st}] {name} {mdl or ''} ${c:.5f} {dt:.1f}s {err[:160] if err else ''}") return st, out, hdrs def filler(seed: str, words: int) -> str: base = ("The quarterly logistics review covers warehouse throughput, carrier performance, customs " "documentation, cold-chain compliance, and regional demand forecasting for the coming period. ") parts = [] n = 0 k = 0 while n < words: parts.append(f"Section {seed}-{k}: " + base) n += len(base.split()) + 2 k += 1 return "\n".join(parts) def tiny_png() -> str: def chunk(t, d): c = struct.pack(">I", len(d)) + t + d return c + struct.pack(">I", zlib.crc32(t + d) & 0xffffffff) raw = b"\x00\xff\x00\x00" # 1 row: filter 0 + RGB red pixel png = (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b"")) return base64.b64encode(png).decode() def tiny_pdf(text: str) -> bytes: stream = f"BT /F1 18 Tf 40 750 Td ({text}) Tj ET".encode() objs = [ b"<< /Type /Catalog /Pages 2 0 R >>", b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >>", b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream", b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", ] out = b"%PDF-1.4\n" offsets = [] for i, o in enumerate(objs, 1): offsets.append(len(out)) out += f"{i} 0 obj\n".encode() + o + b"\nendobj\n" xref = len(out) out += f"xref\n0 {len(objs)+1}\n0000000000 65535 f \n".encode() for off in offsets: out += f"{off:010d} 00000 n \n".encode() out += f"trailer\n<< /Size {len(objs)+1} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode() return out def multipart(fields: dict, filename: str, content: bytes, mime: str) -> tuple[bytes, str]: b = "----atlas" + uuid.uuid4().hex body = b"" for k, v in fields.items(): body += f"--{b}\r\nContent-Disposition: form-data; name=\"{k}\"\r\n\r\n{v}\r\n".encode() body += (f"--{b}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{filename}\"\r\n" f"Content-Type: {mime}\r\n\r\n").encode() + content + f"\r\n--{b}--\r\n".encode() return body, f"multipart/form-data; boundary={b}" def msg(model, content, **kw): body = {"model": model, "max_tokens": kw.pop("max_tokens", 16), "messages": [{"role": "user", "content": content}]} body.update(kw) return body def run(only: set[str] | None = None): def want(tag): return only is None or tag in only # ---------------- (a) prompt caching (Haiku minimum is 4096 tokens) ---------------- if want("cache"): sys_text = filler("A", 3600) # ~4.5k tokens system = [{"type": "text", "text": sys_text, "cache_control": {"type": "ephemeral"}}] b = msg(HAIKU, "Reply with OK.", system=system, max_tokens=8) call("cache_01_write", "POST", "/v1/messages", b) call("cache_02_read", "POST", "/v1/messages", b) # automatic (top-level) cache_control, same system prefix -> expect read of system + write of tail b_auto = msg(HAIKU, "Reply with OK.", system=[{"type": "text", "text": sys_text}], max_tokens=8, cache_control={"type": "ephemeral"}) call("cache_03_auto_top_level", "POST", "/v1/messages", b_auto) # 1h TTL on a fresh prefix sys1h = [{"type": "text", "text": filler("B", 3600), "cache_control": {"type": "ephemeral", "ttl": "1h"}}] call("cache_04_ttl_1h_write", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", system=sys1h, max_tokens=8)) # below minimum: ~600 tokens small = [{"type": "text", "text": filler("C", 450), "cache_control": {"type": "ephemeral"}}] call("cache_05_below_minimum", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", system=small, max_tokens=8)) # caching on tools[] block tools = [{"name": f"tool_{i}", "description": filler(f"T{i}", 700), "input_schema": {"type": "object", "properties": {"q": {"type": "string"}}}} for i in range(6)] tools[-1]["cache_control"] = {"type": "ephemeral"} bt = msg(HAIKU, "Reply with OK. Do not call tools.", tools=tools, max_tokens=8) call("cache_06_tools_write", "POST", "/v1/messages", bt) call("cache_07_tools_read", "POST", "/v1/messages", bt) # pre-warm with max_tokens: 0 call("cache_08_prewarm_max_tokens_0", "POST", "/v1/messages", msg(HAIKU, "warmup", system=system, max_tokens=0)) # invalid ttl value bad = [{"type": "text", "text": sys_text, "cache_control": {"type": "ephemeral", "ttl": "2h"}}] call("cache_09_invalid_ttl", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", system=bad, max_tokens=8)) # 5 breakpoints -> error? five = [{"type": "text", "text": f"Block {i}.", "cache_control": {"type": "ephemeral"}} for i in range(5)] call("cache_10_five_breakpoints", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", system=five, max_tokens=8)) # ---------------- (b) extended thinking on Haiku ---------------- if want("thinking"): th = {"type": "enabled", "budget_tokens": 1024} call("think_01_enabled", "POST", "/v1/messages", msg(HAIKU, "What is 2+2? Reply with the number.", thinking=th, max_tokens=1200)) call("think_02_enabled_stream", "POST", "/v1/messages", msg(HAIKU, "What is 2+2? Reply with the number.", thinking=th, max_tokens=1200, stream=True), stream=True) call("think_03_temperature_conflict", "POST", "/v1/messages", msg(HAIKU, "What is 2+2?", thinking=th, max_tokens=1200, temperature=0.5)) call("think_04_top_k_conflict", "POST", "/v1/messages", msg(HAIKU, "What is 2+2?", thinking=th, max_tokens=1200, top_k=5)) call("think_05_budget_too_small", "POST", "/v1/messages", msg(HAIKU, "What is 2+2?", thinking={"type": "enabled", "budget_tokens": 512}, max_tokens=1200)) call("think_06_budget_ge_max_tokens", "POST", "/v1/messages", msg(HAIKU, "What is 2+2?", thinking={"type": "enabled", "budget_tokens": 2000}, max_tokens=1200)) call("think_07_adaptive_on_haiku", "POST", "/v1/messages", msg(HAIKU, "What is 2+2?", thinking={"type": "adaptive"}, max_tokens=1200)) call("think_08_display_omitted", "POST", "/v1/messages", msg(HAIKU, "What is 2+2? Reply with the number.", thinking={**th, "display": "omitted"}, max_tokens=1200)) call("think_09_forced_tool_choice_conflict", "POST", "/v1/messages", msg(HAIKU, "Call the tool.", thinking=th, max_tokens=1200, tool_choice={"type": "any"}, tools=[{"name": "get_time", "description": "Returns the current time.", "input_schema": {"type": "object", "properties": {}}}])) # tool-use round trip preserving thinking block tool = {"name": "get_time", "description": "Returns the current UTC time.", "input_schema": {"type": "object", "properties": {}}} st, r1, _ = call("think_10_tool_turn1", "POST", "/v1/messages", msg(HAIKU, "Use the get_time tool, then tell me the time in one short sentence.", thinking=th, tools=[tool], max_tokens=1500)) if st == 200 and isinstance(r1, dict): tu = [c for c in r1["content"] if c["type"] == "tool_use"] if tu: msgs = [{"role": "user", "content": "Use the get_time tool, then tell me the time in one short sentence."}, {"role": "assistant", "content": r1["content"]}, {"role": "user", "content": [{"type": "tool_result", "tool_use_id": tu[0]["id"], "content": "12:00:00 UTC"}]}] call("think_11_tool_turn2_pass_back", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 1500, "thinking": th, "tools": [tool], "messages": msgs}) # tampered thinking block tampered = json.loads(json.dumps(msgs)) for c in tampered[1]["content"]: if c["type"] == "thinking": c["thinking"] = c["thinking"] + " (edited)" call("think_12_tool_turn2_tampered", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 1500, "thinking": th, "tools": [tool], "messages": tampered}) # thinking block dropped -> ? dropped = json.loads(json.dumps(msgs)) dropped[1]["content"] = [c for c in dropped[1]["content"] if c["type"] != "thinking"] call("think_13_tool_turn2_thinking_dropped", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 1500, "thinking": th, "tools": [tool], "messages": dropped}) # ---------------- (c) adaptive / interleaved ---------------- if want("adaptive"): call("adaptive_01_sonnet5", "POST", "/v1/messages", msg(SONNET5, "What is 2+2? Reply with the number.", thinking={"type": "adaptive", "display": "summarized"}, max_tokens=600)) call("adaptive_02_sonnet5_disabled", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", thinking={"type": "disabled"}, max_tokens=8)) call("adaptive_03_sonnet5_enabled_rejected", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", thinking={"type": "enabled", "budget_tokens": 1024}, max_tokens=1200)) call("adaptive_04_sonnet5_temperature", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", max_tokens=8, temperature=0.2)) call("adaptive_05_display_updates_no_header", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", thinking={"type": "adaptive", "display": "updates"}, max_tokens=8)) tool = {"name": "get_time", "description": "Returns the current UTC time.", "input_schema": {"type": "object", "properties": {}}} call("interleaved_01_sonnet46_enabled_header", "POST", "/v1/messages", msg(SONNET46, "Reply with OK. Do not call tools.", thinking={"type": "enabled", "budget_tokens": 1024}, tools=[tool], max_tokens=1200), beta="interleaved-thinking-2025-05-14") call("interleaved_02_haiku_header_ignored", "POST", "/v1/messages", msg(HAIKU, "Reply with OK. Do not call tools.", thinking={"type": "enabled", "budget_tokens": 1024}, tools=[tool], max_tokens=1200), beta="interleaved-thinking-2025-05-14") # ---------------- (d) effort ---------------- if want("effort"): for lvl in ("low", "max", "xhigh"): call(f"effort_{lvl}_sonnet5", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", output_config={"effort": lvl}, max_tokens=64)) call("effort_invalid_sonnet5", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", output_config={"effort": "ultra"}, max_tokens=8)) call("effort_low_haiku_unsupported", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", output_config={"effort": "low"}, max_tokens=8)) call("effort_disabled_thinking_max_sonnet5", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", output_config={"effort": "max"}, thinking={"type": "disabled"}, max_tokens=8)) call("effort_per_message_sonnet5_beta", "POST", "/v1/messages", {"model": SONNET5, "max_tokens": 8, "messages": [ {"role": "system", "content": [], "output_config": {"effort": "low"}}, {"role": "user", "content": "Reply with OK."}]}, beta="mid-conversation-output-config-2026-07-01") call("task_budget_sonnet5_unsupported", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", output_config={"task_budget": {"type": "tokens", "total": 20000}}, max_tokens=8), beta="task-budgets-2026-03-13") # ---------------- (e) fast mode ---------------- if want("fast"): call("fast_01_opus5_header", "POST", "/v1/messages", msg(OPUS5, "Reply with OK.", speed="fast", max_tokens=8, thinking={"type": "disabled"}), beta="fast-mode-2026-02-01") call("fast_02_opus5_no_header", "POST", "/v1/messages", msg(OPUS5, "Reply with OK.", speed="fast", max_tokens=8, thinking={"type": "disabled"})) call("fast_03_haiku_unsupported", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", speed="fast", max_tokens=8), beta="fast-mode-2026-02-01") call("fast_04_opus5_speed_standard", "POST", "/v1/messages", msg(OPUS5, "Reply with OK.", speed="standard", max_tokens=8, thinking={"type": "disabled"}), beta="fast-mode-2026-02-01") # ---------------- (f) 1M context header ---------------- if want("ctx1m"): call("ctx1m_01_sonnet5_header", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", max_tokens=8), beta="context-1m-2025-08-07") call("ctx1m_02_haiku_header", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", max_tokens=8), beta="context-1m-2025-08-07") # ---------------- (g) context editing / compaction ---------------- if want("ctxmgmt"): tool = {"name": "lookup", "description": "Looks up a record.", "input_schema": {"type": "object", "properties": {"id": {"type": "string"}}}} hist = [{"role": "user", "content": "Look up records r1 and r2 then reply OK."}] for i in range(3): hist.append({"role": "assistant", "content": [{"type": "tool_use", "id": f"toolu_atlas{i}", "name": "lookup", "input": {"id": f"r{i}"}}]}) hist.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": f"toolu_atlas{i}", "content": "record " + ("x" * 400)}]}) hist.append({"role": "user", "content": "Reply with OK. Do not call tools."}) # consecutive user messages are merged by the API; keep as is cm = {"edits": [{"type": "clear_tool_uses_20250919", "trigger": {"type": "input_tokens", "value": 1}, "keep": {"type": "tool_uses", "value": 1}, "clear_at_least": {"type": "input_tokens", "value": 1}}]} call("ctxedit_01_clear_tool_uses", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 8, "tools": [tool], "messages": hist, "context_management": cm}, beta="context-management-2025-06-27") call("ctxedit_02_count_tokens", "POST", "/v1/messages/count_tokens", {"model": HAIKU, "tools": [tool], "messages": hist, "context_management": cm}, beta="context-management-2025-06-27") call("ctxedit_03_no_header", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 8, "tools": [tool], "messages": hist, "context_management": cm}) call("ctxedit_04_clear_thinking_haiku", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", max_tokens=8, context_management={"edits": [{"type": "clear_thinking_20251015", "keep": {"type": "thinking_turns", "value": 1}}]}), beta="context-management-2025-06-27") comp = {"edits": [{"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 50000}}]} call("compact_01_sonnet5_accepted", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", max_tokens=8, context_management=comp), beta="compact-2026-01-12") call("compact_02_haiku_unsupported", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", max_tokens=8, context_management=comp), beta="compact-2026-01-12") call("compact_03_trigger_too_low", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", max_tokens=8, context_management={"edits": [{"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 1000}}]}), beta="compact-2026-01-12") call("compact_04_on_demand_summarize", "POST", "/v1/messages", {"model": SONNET5, "max_tokens": 400, "compaction": {"type": "summarize"}, "messages": [{"role": "user", "content": "My name is Ada and I like teal."}, {"role": "assistant", "content": "Nice to meet you, Ada."}]}, beta="compact-2026-09-04") # ---------------- (h) structured outputs ---------------- if want("structured"): schema = {"type": "object", "properties": {"name": {"type": "string"}, "age": {"type": "integer"}, "city": {"type": "string"}}, "required": ["name", "age", "city"], "additionalProperties": False} fmt = {"type": "json_schema", "schema": schema} prompt = "Ada is 36 and lives in Lyon. Extract the person." call("so_01_json_schema", "POST", "/v1/messages", msg(HAIKU, prompt, output_config={"format": fmt}, max_tokens=100)) call("so_02_json_schema_stream", "POST", "/v1/messages", msg(HAIKU, prompt, output_config={"format": fmt}, max_tokens=100, stream=True), stream=True) bad = {"type": "object", "patternProperties": {"^x-": {"type": "string"}}, "additionalProperties": False} call("so_03_unsupported_patternProperties", "POST", "/v1/messages", msg(HAIKU, prompt, output_config={"format": {"type": "json_schema", "schema": bad}}, max_tokens=50)) bad2 = {"type": "object", "properties": {"age": {"type": "integer", "minimum": 0}}, "required": ["age"], "additionalProperties": False} call("so_04_unsupported_minimum", "POST", "/v1/messages", msg(HAIKU, prompt, output_config={"format": {"type": "json_schema", "schema": bad2}}, max_tokens=50)) call("so_05_count_tokens", "POST", "/v1/messages/count_tokens", {"model": HAIKU, "messages": [{"role": "user", "content": prompt}], "output_config": {"format": fmt}}) call("so_06_with_citations_conflict", "POST", "/v1/messages", msg(HAIKU, [{"type": "document", "source": {"type": "text", "media_type": "text/plain", "data": "Ada is 36."}, "citations": {"enabled": True}}, {"type": "text", "text": prompt}], output_config={"format": fmt}, max_tokens=50)) call("so_07_legacy_output_format", "POST", "/v1/messages", msg(HAIKU, prompt, output_format=fmt, max_tokens=100)) call("so_08_strict_tool", "POST", "/v1/messages", msg(HAIKU, "Record Ada, 36, Lyon with the tool.", max_tokens=200, tools=[{"name": "record_person", "description": "Records a person.", "strict": True, "input_schema": schema}], tool_choice={"type": "tool", "name": "record_person"})) # ---------------- (i) citations ---------------- if want("citations"): doc = {"type": "document", "source": {"type": "text", "media_type": "text/plain", "data": "The sky is blue. Grass is green."}, "title": "Colors", "citations": {"enabled": True}} q = {"type": "text", "text": "What color is grass? Answer in one sentence and cite."} call("cite_01_text_char_location", "POST", "/v1/messages", msg(HAIKU, [doc, q], max_tokens=100)) sr = {"type": "search_result", "source": "kb://colors/1", "title": "Color facts", "content": [{"type": "text", "text": "The sky is blue."}, {"type": "text", "text": "Grass is green."}], "citations": {"enabled": True}} call("cite_02_search_result_location", "POST", "/v1/messages", msg(HAIKU, [sr, q], max_tokens=100)) custom = {"type": "document", "source": {"type": "content", "content": [{"type": "text", "text": "The sky is blue."}, {"type": "text", "text": "Grass is green."}]}, "title": "Colors", "citations": {"enabled": True}} call("cite_03_content_block_location", "POST", "/v1/messages", msg(HAIKU, [custom, q], max_tokens=100)) pdf = tiny_pdf("The sky is blue. Grass is green.") (OUT / "tiny.pdf").write_bytes(pdf) pdoc = {"type": "document", "source": {"type": "base64", "media_type": "application/pdf", "data": base64.b64encode(pdf).decode()}, "title": "Colors PDF", "citations": {"enabled": True}} call("cite_04_pdf_page_location", "POST", "/v1/messages", msg(HAIKU, [pdoc, q], max_tokens=100)) call("cite_05_stream_citations_delta", "POST", "/v1/messages", msg(HAIKU, [doc, q], max_tokens=100, stream=True), stream=True) doc2 = dict(doc); doc2 = {**doc, "citations": {"enabled": False}} call("cite_06_mixed_enabled_error", "POST", "/v1/messages", msg(HAIKU, [doc, doc2, q], max_tokens=50)) # ---------------- (j) vision ---------------- if want("vision"): img = {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": tiny_png()}} call("vision_01_base64_1x1", "POST", "/v1/messages", msg(HAIKU, [img, {"type": "text", "text": "What color is this pixel? One word."}], max_tokens=10)) url_img = {"type": "image", "source": {"type": "url", "url": "https://upload.wikimedia.org/wikipedia/commons/thumb/4/47/PNG_transparency_demonstration_1.png/280px-PNG_transparency_demonstration_1.png"}} call("vision_02_url", "POST", "/v1/messages", msg(HAIKU, [url_img, {"type": "text", "text": "Describe this image in five words."}], max_tokens=20)) bad = {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": base64.b64encode(b"notapng").decode()}} call("vision_03_invalid_image_data", "POST", "/v1/messages", msg(HAIKU, [bad, {"type": "text", "text": "Describe."}], max_tokens=10)) call("vision_04_count_tokens_image", "POST", "/v1/messages/count_tokens", {"model": HAIKU, "messages": [{"role": "user", "content": [img, {"type": "text", "text": "Describe."}]}]}) txtdoc = {"type": "document", "source": {"type": "text", "media_type": "text/plain", "data": "Grass is green."}} call("vision_05_text_document_no_citations", "POST", "/v1/messages", msg(HAIKU, [txtdoc, {"type": "text", "text": "What color is grass? One word."}], max_tokens=10)) img_err = {**img, "transformations": {"oversized_image": "error"}} call("vision_06_transformations_field", "POST", "/v1/messages", msg(HAIKU, [img_err, {"type": "text", "text": "What color? One word."}], max_tokens=10)) # ---------------- (k) Files API ---------------- if want("files"): txt = b"The sky is blue. Grass is green. This is a 100-byte-ish plain text file for the API Atlas Files test." body, ct = multipart({}, "atlas.txt", txt, "text/plain") st, up, _ = call("files_01_upload_txt", "POST", "/v1/files", data=body, content_type=ct, model=HAIKU) fid = up.get("id") if isinstance(up, dict) else None call("files_02_list", "GET", "/v1/files?limit=2", model=HAIKU) call("files_03_list_legacy_beta_header", "GET", "/v1/files?limit=2", beta="files-api-2025-04-14", model=HAIKU) if fid: call("files_04_list_ids_filter", "GET", f"/v1/files?ids[]={fid}", model=HAIKU) call("files_05_retrieve", "GET", f"/v1/files/{fid}", model=HAIKU) call("files_06_content_download", "GET", f"/v1/files/{fid}/content", model=HAIKU) fdoc = {"type": "document", "source": {"type": "file", "file_id": fid}, "citations": {"enabled": True}} call("files_07_message_document_file_id", "POST", "/v1/messages", msg(HAIKU, [fdoc, {"type": "text", "text": "What color is grass? One sentence, cite."}], max_tokens=80)) call("files_08_message_document_file_id_beta_header", "POST", "/v1/messages", msg(HAIKU, [{"type": "document", "source": {"type": "file", "file_id": fid}}, {"type": "text", "text": "What color is grass? One word."}], max_tokens=10), beta="files-api-2025-04-14") pdf = tiny_pdf("The sky is blue. Grass is green.") body, ct = multipart({"expires_in_seconds": "3600"}, "atlas.pdf", pdf, "application/pdf") st, up2, _ = call("files_09_upload_pdf_expires", "POST", "/v1/files", data=body, content_type=ct, model=HAIKU) pid = up2.get("id") if isinstance(up2, dict) else None if pid: call("files_10_message_pdf_file_id", "POST", "/v1/messages", msg(HAIKU, [{"type": "document", "source": {"type": "file", "file_id": pid}}, {"type": "text", "text": "What color is grass? One word."}], max_tokens=10)) call("files_11_image_block_with_pdf_file_error", "POST", "/v1/messages", msg(HAIKU, [{"type": "image", "source": {"type": "file", "file_id": pid}}, {"type": "text", "text": "Describe."}], max_tokens=10)) png = base64.b64decode(tiny_png()) body, ct = multipart({}, "pixel.png", png, "image/png") st, up3, _ = call("files_12_upload_png", "POST", "/v1/files", data=body, content_type=ct, model=HAIKU) iid = up3.get("id") if isinstance(up3, dict) else None if iid: call("files_13_message_image_file_id", "POST", "/v1/messages", msg(HAIKU, [{"type": "image", "source": {"type": "file", "file_id": iid}}, {"type": "text", "text": "What color? One word."}], max_tokens=10)) body, ct = multipart({"expires_in_seconds": "60"}, "short.txt", b"x", "text/plain") call("files_14_upload_expires_too_short", "POST", "/v1/files", data=body, content_type=ct, model=HAIKU) for i, f in enumerate([fid, pid, iid]): if f: call(f"files_15_delete_{i}", "DELETE", f"/v1/files/{f}", model=HAIKU) if fid: call("files_16_retrieve_deleted", "GET", f"/v1/files/{fid}", model=HAIKU) # ---------------- extra probes ---------------- if want("extra"): url_img = {"type": "image", "source": {"type": "url", "url": "https://httpbin.org/image/png"}} call("vision_07_url_httpbin", "POST", "/v1/messages", msg(HAIKU, [url_img, {"type": "text", "text": "Describe this image in five words."}], max_tokens=20)) call("cache_11_message_level_cache_control", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 8, "messages": [{"role": "user", "content": "Reply with OK.", "cache_control": {"type": "ephemeral"}}]}) call("think_14_prefill_conflict", "POST", "/v1/messages", {"model": HAIKU, "max_tokens": 1200, "thinking": {"type": "enabled", "budget_tokens": 1024}, "messages": [{"role": "user", "content": "What is 2+2?"}, {"role": "assistant", "content": "The answer is"}]}) call("adaptive_06_display_updates_with_header", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", thinking={"type": "adaptive", "display": "updates"}, max_tokens=8), beta="thinking-display-updates-2026-08-18") call("adaptive_07_block_binding_beta", "POST", "/v1/messages", msg(SONNET5, "Reply with OK.", thinking={"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "drop_block"}}, max_tokens=8), beta="thinking-binding-controls-2026-08-01") prev = json.load(open(OUT / "compact_04_on_demand_summarize.json"))["body"] if isinstance(prev, dict) and prev.get("content"): blk = prev["content"][0] call("compact_05_continue_from_summary", "POST", "/v1/messages", {"model": SONNET5, "max_tokens": 20, "messages": [ {"role": "assistant", "content": [blk]}, {"role": "user", "content": "What is my name and favourite color? Answer in five words."}]}, beta="compact-2026-09-04") call("compact_06_block_misplaced", "POST", "/v1/messages", {"model": SONNET5, "max_tokens": 20, "messages": [ {"role": "user", "content": "My name is Ada and I like teal."}, {"role": "assistant", "content": [blk]}, {"role": "user", "content": "What is my name?"}]}, beta="compact-2026-09-04") call("cache_12_diagnostics_beta", "POST", "/v1/messages", msg(HAIKU, "Reply with OK.", max_tokens=8, diagnostics={"previous_message_id": None}), beta="cache-diagnosis-2026-04-07") b17, ct17 = multipart({}, "plain.txt", b"hello atlas", "application/octet-stream") st, up17, _ = call("files_17_upload_octet_stream_part", "POST", "/v1/files", data=b17, content_type=ct17, model=HAIKU) if isinstance(up17, dict) and up17.get("id"): call("files_18_delete_octet", "DELETE", f"/v1/files/{up17['id']}", model=HAIKU) tag = "-".join(sorted(only)) if only else "all" live.save_sanitized({"total_est_cost_usd": round(TOTAL[0], 5), "calls": SUMMARY}, OUT / f"_summary_{tag}.json") print(f"\nTOTAL est cost: ${TOTAL[0]:.4f} over {len(SUMMARY)} calls") if __name__ == "__main__": run(set(sys.argv[1:]) or None)