AI Agent Orchestration with Claude: Running Many Agents at Once Without Losing Control of Quality or Cost

01What orchestration means once agents run in parallel

A single Claude agent is a loop: the model reads its context, calls a tool or answers, and repeats until it finishes or something stops it. Orchestration starts when several loops run together and you must decide who starts each one, what it can see, how results merge and what ends the run. Concurrency changes four things.

Rate limits are shared. Limits apply per organization and model, so ten Sonnet 5.5 agents draw from one bucket: on the Start tier, 1,000 requests, 2,000,000 uncached input tokens and 400,000 output tokens per minute.

State is shared. Two agents editing one file, or two workers that both think they own the plan, create conflicts a single loop never meets. Failures must be isolated. When one worker times out, the others should continue, and a long run should resume from a checkpoint rather than restart; Anthropic's Research system works this way.

Cost multiplies. In Anthropic's data, agents use about four times the tokens of a chat and multi-agent systems about fifteen times. A Claude Opus 4 lead with Claude Sonnet 4 subagents beat a single Opus 4 agent by 90.2% on an internal research evaluation, and token usage alone explained 80% of performance variance on BrowseComp. Parallel agents buy quality with tokens, so the spending has to be deliberate.

The examples use the lineup current on 22 October 2026: Claude Fable 5.1 (claude-fable-5-1), Opus 5.5 (claude-opus-5-5), Sonnet 5.5 (claude-sonnet-5-5) and Haiku 5.5 (claude-haiku-5-5), all with 1M-token context and 128K maximum output.

02Core concepts

Orchestrator: whatever decides what work exists, who does it and when the run ends. It can be an agent, plain code, or code that asks an agent to plan. Anthropic separates workflows, where code fixes the path, from agents, where the model directs itself, and advises adding agentic control only when it measurably helps.

Worker or subagent: a separate agent instance with its own context. In the Claude Agent SDK, Claude starts subagents through the Agent tool; each begins fresh, keeps its tool calls to itself and returns only its final message.

Tools: functions the model can request. Client tools run in your process; server tools run on Anthropic's side and can cost extra (web search is $10 per 1,000 searches). The Model Context Protocol (MCP) is the open standard for exposing tools and data to compatible clients.

Handoffs: a subagent sees its own system prompt and the prompt string its parent passes, never the parent's history, so paths, errors and decisions must travel in that string.

Shared memory: anything agents read and write outside their context windows, such as files, a database or a task board. Anthropic's Research lead saves its plan to memory and lets subagents write results to a filesystem and pass back references.

Termination: every loop needs a stop besides the model deciding it is done. The Agent SDK offers max_turns and max_budget_usd, ending with error_max_turns or error_max_budget_usd. Managed Agents sessions take a dollar budget and go idle with budget_reached.

Every pattern below composes this loop: an orchestrator is a loop whose tools start other loops, and every cost problem traces back to the growing context box on the left.

03Five multi-agent architectures

These five patterns cover the multi-agent designs that appear most often in published systems; Anthropic's own names for them are noted where they differ.

1. Orchestrator-worker

A lead agent splits the task at run time, starts workers in parallel, reads their reports and decides whether more work is needed. Anthropic's Research feature works this way.

The handoff arrow matters most. Anthropic found that each worker needs an objective, an output format, tool guidance and clear boundaries, or workers duplicate each other.

2. Hierarchical

A director delegates to team leads, and each lead runs its own workers: orchestrator-worker applied twice.

3. Sequential pipeline

Agents run in a fixed order, each consuming the previous output, with programmatic gates between steps; Anthropic calls this prompt chaining. Many pipelines can run in parallel over different inputs.

4. Parallel fan-out and fan-in

A splitter divides the input, identical workers process the pieces in parallel and a reducer merges the results. Anthropic calls this parallelization, in two forms: sectioning and voting.

Fan-out is the pattern most likely to hit rate limits, so check the workload against your tier before you launch it.

5. Router

A classifier hands each request to a specialist with its own prompt, tools and model, for example small models for easy questions.

Three more patterns

Peer-to-peer handoff passes control between agents without a lead, which cuts latency but complicates tracing and budgets. Blackboard systems share a workspace; Tutorial Advanced 1 uses one. Critic loops pair a generator with an evaluator until it accepts (Anthropic's evaluator-optimizer). Cap the rounds: each one pays for both agents' context again.

Side-by-side comparison

Green marks the favourable end of each column, red the unfavourable end. Relative cost assumes the same total work and model mix.

04Six tutorials

Each tutorial runs several agents at once, reads the key from ANTHROPIC_API_KEY, records token usage and stops at a dollar cap. Cost estimates use list prices checked on 22 October 2026.

# basic_parallel.py: three Claude calls at once, then one merge call.
# Requires: pip install anthropic   (Python 3.10+). Key read from ANTHROPIC_API_KEY.
import asyncio
import sys

from anthropic import AsyncAnthropic

MODEL = "claude-sonnet-5-5"
# USD per million tokens, from the Claude pricing page, checked 22 Oct 2026.
PRICE = {"in": 2.00, "out": 10.00, "cache_write": 2.50, "cache_read": 0.10}
SESSION_CAP_USD = 0.25          # hard stop for this whole run
MAX_PARALLEL = 3                # stay well under your RPM limit

client = AsyncAnthropic()       # retries 429 and 5xx twice by default


class BudgetExceeded(RuntimeError):
    pass


class Ledger:
    """Running cost total shared by every concurrent call."""

    def __init__(self, cap_usd: float):
        self.cap = cap_usd
        self.spent = 0.0
        self._lock = asyncio.Lock()

    def check(self) -> None:
        if self.spent >= self.cap:
            raise BudgetExceeded(f"cap ${self.cap:.2f} reached (${self.spent:.4f} spent)")

    async def charge(self, usage) -> float:
        cost = (
            usage.input_tokens * PRICE["in"]
            + usage.output_tokens * PRICE["out"]
            + (usage.cache_creation_input_tokens or 0) * PRICE["cache_write"]
            + (usage.cache_read_input_tokens or 0) * PRICE["cache_read"]
        ) / 1_000_000
        async with self._lock:
            self.spent += cost
        return cost


ledger = Ledger(SESSION_CAP_USD)
gate = asyncio.Semaphore(MAX_PARALLEL)

ROLES = {
    "summary": "Summarise the document in five bullet points for an engineering manager.",
    "risks": "List the technical and delivery risks the document implies, most severe first.",
    "questions": "Write the five questions a sceptical reviewer would ask the author.",
}


async def run_agent(role: str, instruction: str, document: str) -> tuple[str, str]:
    async with gate:
        ledger.check()                          # refuse to start if the cap is gone
        msg = await client.messages.create(
            model=MODEL,
            max_tokens=800,
            system=f"You are the {role} agent in a review team. {instruction}",
            messages=[{"role": "user", "content": document}],
        )
        cost = await ledger.charge(msg.usage)
        print(f"[{role}] in={msg.usage.input_tokens} out={msg.usage.output_tokens} "
              f"cost=${cost:.4f} total=${ledger.spent:.4f}")
        text = "".join(b.text for b in msg.content if b.type == "text")
        return role, text


async def main(path: str) -> None:
    document = open(path, encoding="utf-8").read()
    results = await asyncio.gather(
        *(run_agent(r, i, document) for r, i in ROLES.items()),
        return_exceptions=True,
    )
    sections = []
    for r in results:
        if isinstance(r, Exception):
            print(f"agent failed: {r!r}")      # one failure does not sink the others
            continue
        role, text = r
        sections.append(f"## {role}\n{text}")

    ledger.check()
    merged = await client.messages.create(
        model=MODEL,
        max_tokens=1000,
        system="Merge the review sections into one brief. Remove repetition. Keep every distinct risk.",
        messages=[{"role": "user", "content": "\n\n".join(sections)}],
    )
    await ledger.charge(merged.usage)
    print("".join(b.text for b in merged.content if b.type == "text"))
    print(f"\nRun total: ${ledger.spent:.4f} of ${SESSION_CAP_USD:.2f} cap")


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

# parallel_subagents.py: one prompt, three read-only subagents running at once.
# Requires: pip install claude-agent-sdk   Run it from the root of the repository to review.
import asyncio

from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, ResultMessage, query

READ_ONLY = ["Read", "Grep", "Glob"]          # no Edit, Write or Bash for any worker

AGENTS = {
    "api-mapper": AgentDefinition(
        description="Lists every public HTTP endpoint with its handler file.",
        prompt="Find each route definition. Report path, method and handler file as a table. "
               "Do not judge code quality.",
        tools=READ_ONLY,
        model="haiku",                         # mechanical search: smallest model
    ),
    "test-auditor": AgentDefinition(
        description="Finds public functions that have no test covering them.",
        prompt="Compare public functions in src/ with tests/. Report untested functions "
               "with file and line. Maximum 25 items.",
        tools=READ_ONLY,
        model="sonnet",
    ),
    "secrets-scanner": AgentDefinition(
        description="Looks for hard-coded credentials and unsafe config defaults.",
        prompt="Search for API keys, tokens, passwords and debug flags left on. "
               "Report file, line and why it matters. Never print a full secret.",
        tools=READ_ONLY,
        model="sonnet",
    ),
}

PROMPT = (
    "Run the api-mapper, test-auditor and secrets-scanner agents in parallel and wait for "
    "all three. Then write one report with a section per agent and a short list of the "
    "three most urgent fixes. Do not explore the code yourself."
)


async def main() -> None:
    options = ClaudeAgentOptions(
        allowed_tools=READ_ONLY + ["Agent"],   # Agent lets Claude start subagents
        agents=AGENTS,
        env={
            "CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "1",   # workers cannot spawn workers
            "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "3",
        },
        max_budget_usd=1.50,                   # whole run, subagents included
        max_turns=20,
    )
    try:
        async for message in query(prompt=PROMPT, options=options):
            if isinstance(message, ResultMessage):
                print(message.result if message.subtype == "success" else f"Stopped: {message.subtype}")
                print(f"Estimated cost: ${message.total_cost_usd or 0:.4f}")
                for model, usage in (message.model_usage or {}).items():
                    print(f"  {model}: {usage}")
    except Exception as error:                 # query() raises after an error result
        print(f"Run ended with an error: {error}")


asyncio.run(main())

---
name: secrets-scanner
description: Looks for hard-coded credentials and unsafe config defaults. Use for security sweeps.
tools: Read, Grep, Glob
model: sonnet
---

Search the repository for API keys, tokens, passwords and debug flags left on.
Report file, line and why it matters. Never print a full secret value.

# orchestrator.py: plan with Opus, run Agent SDK workers in parallel, validate, synthesise.
# Requires: pip install anthropic claude-agent-sdk
import asyncio
import json
import sys

from anthropic import AsyncAnthropic
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query

LEAD_MODEL, WORKER_MODEL = "claude-opus-5-5", "claude-sonnet-5-5"
OPUS = {"in": 4.00, "out": 20.00}          # USD per million tokens, checked 22 Oct 2026
RUN_CAP_USD = 4.00                          # orchestrator + all workers
WORKER_CAP_USD = 0.60                       # each worker stops itself here
MAX_WORKERS, MAX_PARALLEL = 5, 3

client = AsyncAnthropic()
spent = 0.0


def charge_lead(usage) -> None:
    global spent
    spent += (usage.input_tokens * OPUS["in"] + usage.output_tokens * OPUS["out"]) / 1e6
    if spent >= RUN_CAP_USD:
        raise RuntimeError(f"run cap hit at ${spent:.3f}")


async def plan(question: str) -> list[dict]:
    msg = await client.messages.create(
        model=LEAD_MODEL,
        max_tokens=1500,
        system=(
            "You plan research for parallel workers. Return only JSON: a list of at most "
            f"{MAX_WORKERS} objects with keys 'id', 'objective', 'sources', 'out_of_scope'. "
            "Subtasks must not overlap. Simple questions need one subtask."
        ),
        messages=[{"role": "user", "content": question}],
    )
    charge_lead(msg.usage)
    text = "".join(b.text for b in msg.content if b.type == "text")
    tasks = json.loads(text[text.find("["): text.rfind("]") + 1])
    required = {"id", "objective", "sources", "out_of_scope"}
    tasks = [t for t in tasks if required <= t.keys()][:MAX_WORKERS]
    if not tasks:
        raise ValueError("planner returned no valid subtasks")
    return tasks


async def run_worker(task: dict, gate: asyncio.Semaphore) -> dict:
    global spent
    prompt = (
        f"Objective: {task['objective']}\nPreferred sources: {task['sources']}\n"
        f"Out of scope (another worker covers it): {task['out_of_scope']}\n"
        "Finish with a section headed FINDINGS: listing claims, each with a source URL."
    )
    options = ClaudeAgentOptions(
        model=WORKER_MODEL,
        allowed_tools=["WebSearch", "WebFetch"],
        max_budget_usd=WORKER_CAP_USD,
        max_turns=15,
    )
    async with gate:
        if spent >= RUN_CAP_USD:
            return {"id": task["id"], "ok": False, "reason": "run cap reached before start"}
        result = None
        try:
            async for m in query(prompt=prompt, options=options):
                if isinstance(m, ResultMessage):
                    result = m
        except Exception as error:              # error results raise after being yielded
            if result is None:
                return {"id": task["id"], "ok": False, "reason": repr(error)}
        spent += result.total_cost_usd or 0     # SDK estimate; billing is authoritative
        ok = result.subtype == "success" and "FINDINGS:" in (result.result or "")
        return {"id": task["id"], "ok": ok, "reason": result.subtype,
                "cost": result.total_cost_usd, "text": result.result or ""}


async def main(question: str) -> None:
    tasks = await plan(question)
    print(f"Plan: {[t['id'] for t in tasks]}  (spent ${spent:.3f})")
    gate = asyncio.Semaphore(MAX_PARALLEL)
    reports = await asyncio.gather(*(run_worker(t, gate) for t in tasks))
    for r in reports:
        print(f"  {r['id']}: ok={r['ok']} reason={r['reason']} cost=${r.get('cost') or 0:.3f}")

    good = [r for r in reports if r["ok"]]
    if not good:
        sys.exit("No worker produced valid findings; nothing to synthesise.")
    findings = "\n\n".join(f"### {r['id']}\n{r['text']}" for r in good)
    final = await client.messages.create(
        model=LEAD_MODEL,
        max_tokens=2500,
        system="Synthesise the worker findings into one answer. Keep every source URL. "
               "Flag claims that only one worker supports.",
        messages=[{"role": "user", "content": f"Question: {question}\n\n{findings}"}],
    )
    charge_lead(final.usage)
    print("".join(b.text for b in final.content if b.type == "text"))
    print(f"\nRun total (estimate): ${spent:.3f} of ${RUN_CAP_USD:.2f}")


if __name__ == "__main__":
    asyncio.run(main(" ".join(sys.argv[1:])))

# batch_fanout.py: classify thousands of documents with the Message Batches API,
# sharing one cached rubric, then compare the bill with synchronous calls.
# Requires: pip install anthropic
import json
import pathlib
import sys
import time

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

MODEL = "claude-haiku-5-5"
# Haiku 5.5 list prices for prompts up to 100,000 tokens, USD per million (22 Oct 2026).
P = {"in": 0.10, "out": 0.50, "write_1h": 0.20, "read": 0.01}
BATCH_DISCOUNT = 0.5             # batch pricing is 50% of standard, caching stacks with it
JOB_CAP_USD = 5.00
CHARS_PER_TOKEN = 3              # deliberately pessimistic pre-flight estimate

client = anthropic.Anthropic()
rubric = pathlib.Path("rubric.md").read_text(encoding="utf-8")   # long, shared, cacheable
docs = sorted(pathlib.Path(sys.argv[1]).glob("*.txt"))

SYSTEM = [
    {"type": "text", "text": "Classify the support ticket using the rubric. Reply with JSON."},
    # 1-hour TTL because batch work can take longer than the default 5 minutes
    {"type": "text", "text": rubric, "cache_control": {"type": "ephemeral", "ttl": "1h"}},
]

# Pre-flight: refuse to submit a job whose worst case is over the cap.
est_in = sum(len(d.read_text(encoding="utf-8")) + len(rubric) for d in docs) / CHARS_PER_TOKEN
est_out = 300 * len(docs)
worst = (est_in * P["write_1h"] + est_out * P["out"]) / 1e6 * BATCH_DISCOUNT
print(f"{len(docs)} docs, worst-case estimate ${worst:.2f}")
if worst > JOB_CAP_USD:
    sys.exit(f"Refusing to submit: estimate exceeds ${JOB_CAP_USD:.2f}")

requests = [
    Request(
        custom_id=f"doc-{i:06d}",                       # results return in any order
        params=MessageCreateParamsNonStreaming(
            model=MODEL,
            max_tokens=300,
            system=SYSTEM,
            messages=[{"role": "user", "content": d.read_text(encoding="utf-8")}],
        ),
    )
    for i, d in enumerate(docs)
]
batch = client.messages.batches.create(requests=requests)   # up to 100,000 requests
print("Submitted", batch.id)

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    print("processing:", batch.request_counts)
    time.sleep(60)

tok = {"in": 0, "out": 0, "write": 0, "read": 0}
labels, failed = {}, []
for result in client.messages.batches.results(batch.id):
    if result.result.type != "succeeded":
        failed.append(result.custom_id)               # errored/expired requests are not billed
        continue
    msg = result.result.message
    u = msg.usage
    tok["in"] += u.input_tokens
    tok["out"] += u.output_tokens
    tok["write"] += u.cache_creation_input_tokens or 0
    tok["read"] += u.cache_read_input_tokens or 0
    labels[result.custom_id] = "".join(b.text for b in msg.content if b.type == "text")

list_cost = (tok["in"] * P["in"] + tok["out"] * P["out"]
             + tok["write"] * P["write_1h"] + tok["read"] * P["read"]) / 1e6
batch_cost = list_cost * BATCH_DISCOUNT
sync_no_cache = ((tok["in"] + tok["write"] + tok["read"]) * P["in"] + tok["out"] * P["out"]) / 1e6
hit_rate = tok["read"] / max(1, tok["read"] + tok["write"])

pathlib.Path("labels.json").write_text(json.dumps(labels, indent=1))
print(json.dumps(tok), f"cache hit rate {hit_rate:.0%}")
print(f"Batch + cache: ${batch_cost:.2f}   Synchronous, no cache: ${sync_no_cache:.2f}")
print(f"Failed or expired: {len(failed)} (resubmit these ids)")

# hierarchy.py: a Python director runs two team leads in parallel. Each lead is an Agent SDK
# session with its own subagents, its own budget and access to a shared findings board
# served as an in-process MCP server. Every event is logged as one JSON line.
# Requires: pip install claude-agent-sdk
import asyncio
import json
import time
from typing import Any

from claude_agent_sdk import (AgentDefinition, ClaudeAgentOptions, ResultMessage,
                              create_sdk_mcp_server, query, tool)

GLOBAL_CAP_USD = 6.00
LEAD_CAP_USD = {"security": 3.00, "docs": 2.00}
LOG = open("run.jsonl", "a", encoding="utf-8")


def log(**event: Any) -> None:
    event["ts"] = round(time.time(), 3)
    LOG.write(json.dumps(event) + "\n")
    LOG.flush()


# ---- shared state: the blackboard ------------------------------------------------
BOARD: list[dict] = []
BOARD_LOCK = asyncio.Lock()
SPENT = {"total": 0.0}


@tool("post_finding", "Record one finding on the shared board. Call once per finding.",
      {"author": str, "severity": str, "summary": str, "location": str})
async def post_finding(args: dict[str, Any]) -> dict[str, Any]:
    if SPENT["total"] >= GLOBAL_CAP_USD:          # the board refuses work once money is gone
        return {"content": [{"type": "text", "text": "Budget exhausted: stop and report."}],
                "is_error": True}
    async with BOARD_LOCK:
        BOARD.append(args)
        n = len(BOARD)
    log(event="finding", **args)
    return {"content": [{"type": "text", "text": f"Recorded as finding #{n}."}]}


@tool("read_board", "Read every finding posted so far by any team.", {})
async def read_board(args: dict[str, Any]) -> dict[str, Any]:
    async with BOARD_LOCK:
        snapshot = json.dumps(BOARD, indent=1)
    return {"content": [{"type": "text", "text": snapshot or "[]"}]}


board = create_sdk_mcp_server(name="board", version="1.0.0", tools=[post_finding, read_board])
BOARD_TOOLS = ["mcp__board__post_finding", "mcp__board__read_board"]
READ = ["Read", "Grep", "Glob"]

TEAMS = {
    "security": {
        "brief": "Audit authentication, input validation and secrets handling in src/.",
        "workers": {
            "authn-reviewer": AgentDefinition(
                description="Reviews login, session and token code.",
                prompt="Review authentication code. Post each issue with post_finding.",
                tools=READ + BOARD_TOOLS, model="sonnet", maxTurns=12),
            "input-reviewer": AgentDefinition(
                description="Reviews request parsing and validation.",
                prompt="Find unvalidated input paths. Post each issue with post_finding.",
                tools=READ + BOARD_TOOLS, model="sonnet", maxTurns=12),
        },
    },
    "docs": {
        "brief": "Check that docs/ matches the public API in src/.",
        "workers": {
            "api-diff": AgentDefinition(
                description="Compares documented endpoints with code.",
                prompt="List endpoints that are undocumented or documented wrongly. "
                       "Post each with post_finding.",
                tools=READ + BOARD_TOOLS, model="haiku", maxTurns=10),
        },
    },
}


async def run_lead(team: str) -> dict:
    spec = TEAMS[team]
    options = ClaudeAgentOptions(
        model="claude-sonnet-5-5",
        mcp_servers={"board": board},
        agents=spec["workers"],
        allowed_tools=READ + BOARD_TOOLS + ["Agent"],
        env={"CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "1",
             "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "3"},
        max_budget_usd=LEAD_CAP_USD[team],      # halts this lead and its subagents
        max_turns=25,
    )
    prompt = (f"You lead the {team} team. {spec['brief']} Delegate to your agents in "
              f"parallel, wait for them, then read_board and write a team summary.")
    log(event="lead_start", team=team, cap=LEAD_CAP_USD[team])
    result = None
    try:
        async for m in query(prompt=prompt, options=options):
            if isinstance(m, ResultMessage):
                result = m
    except Exception as error:
        log(event="lead_error", team=team, error=repr(error))
    cost = (result.total_cost_usd or 0) if result else 0
    SPENT["total"] += cost
    log(event="lead_end", team=team, subtype=result.subtype if result else "crash",
        cost=cost, model_usage=result.model_usage if result else None)
    return {"team": team, "summary": result.result if result and result.subtype == "success" else ""}


async def director() -> None:
    teams = await asyncio.gather(*(run_lead(t) for t in TEAMS))
    log(event="teams_done", spent=SPENT["total"], findings=len(BOARD))
    if SPENT["total"] >= GLOBAL_CAP_USD:
        print("Global cap reached; skipping synthesis.")
        return
    summaries = "\n\n".join(f"## {t['team']}\n{t['summary']}" for t in teams)
    async for m in query(
        prompt=f"Merge these team summaries into one release-risk report:\n\n{summaries}",
        options=ClaudeAgentOptions(model="claude-opus-5-5", allowed_tools=[],
                                   max_budget_usd=GLOBAL_CAP_USD - SPENT["total"], max_turns=2),
    ):
        if isinstance(m, ResultMessage):
            SPENT["total"] += m.total_cost_usd or 0
            print(m.result)
    log(event="run_end", spent=SPENT["total"])
    print(f"Estimated spend ${SPENT['total']:.2f} of ${GLOBAL_CAP_USD:.2f}")


asyncio.run(director())

---
name: Dependency Upgrader
model: claude-sonnet-5-5
tools:
  - type: agent_toolset_20260401
---

You upgrade one dependency in the repository you are given. Run the test suite before
and after. If tests fail after the upgrade, revert and explain why. Never push to a remote.

# yaml-language-server: $schema=https://platform.claude.com/schemas/ant/beta/environment.json
name: upgrade-env
config:
  type: cloud
  networking:
    type: limited
    allow_package_managers: true

# fleet.py: run several Claude Managed Agents sessions at once with a per-session dollar
# budget, a global reservation-based cap, retries on rate limits and a kill switch.
# Requires: pip install anthropic. AGENT_ID and ENVIRONMENT_ID come from `ant apply`.
import os
import random
import signal
import threading
import time
from concurrent.futures import ThreadPoolExecutor

import anthropic

client = anthropic.Anthropic()
AGENT_ID, ENV_ID = os.environ["AGENT_ID"], os.environ["ENVIRONMENT_ID"]

SESSION_CAP_CENTS = 150          # $1.50 per session: tokens + web search + $0.08/running hour
GLOBAL_CAP_CENTS = 900           # $9.00 for the whole fleet
MAX_CONCURRENT = 4               # create endpoints allow 300 requests/minute per org
KILL = threading.Event()         # set by Ctrl+C, a file flag or your own alerting
lock = threading.Lock()
ledger = {"reserved": 0, "spent": 0}

signal.signal(signal.SIGINT, lambda *_: KILL.set())


def reserve(cents: int) -> bool:
    """Reserve a session's whole budget before starting it, so the fleet can never
    commit more than GLOBAL_CAP_CENTS even if every session spends its maximum."""
    with lock:
        if KILL.is_set() or ledger["reserved"] + cents > GLOBAL_CAP_CENTS:
            return False
        ledger["reserved"] += cents
        return True


def settle(reserved: int, actual: int) -> None:
    with lock:
        ledger["reserved"] -= reserved - actual   # release the unused part
        ledger["spent"] += actual


def with_retry(fn, attempts: int = 5):
    for i in range(attempts):
        try:
            return fn()
        except anthropic.RateLimitError:          # SDK already retried twice
            time.sleep(min(60, 2 ** i) + random.random())
    raise RuntimeError("rate limited after retries")


def run_task(task: str) -> dict:
    if not reserve(SESSION_CAP_CENTS):
        return {"task": task, "status": "skipped: cap or kill switch"}
    session = with_retry(lambda: client.beta.sessions.create(
        agent=AGENT_ID,
        environment_id=ENV_ID,
        title=task[:60],
        budget={"type": "limit",
                "max_list_cost": {"amount": str(SESSION_CAP_CENTS), "currency": "USD"}},
    ))
    cost_cents, stop, interrupted = SESSION_CAP_CENTS, "unknown", False  # assume the worst
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(session.id, events=[
            {"type": "user.message", "content": [{"type": "text", "text": task}]}])
        for event in stream:
            if KILL.is_set() and not interrupted:  # stop work; keep reading until idle
                client.beta.sessions.events.send(session.id, events=[{"type": "user.interrupt"}])
                interrupted = True
            if event.type == "session.usage":     # fires just before the session goes idle
                cost_cents = int(event.usage.list_cost.amount)   # whole cents as a string
            elif event.type == "session.status_idle":
                stop = event.stop_reason.type     # end_turn, requires_action or budget_reached
                break
            elif event.type == "session.error":
                stop = "error"
                break
    settle(SESSION_CAP_CENTS, cost_cents)
    return {"task": task, "session": session.id, "stop": stop, "cents": cost_cents}


TASKS = [f"Upgrade {pkg} to its latest minor version in repo ./service"
         for pkg in ["requests", "pydantic", "sqlalchemy", "httpx", "rich", "click"]]

with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as pool:
    for r in pool.map(run_task, TASKS):
        print(r)
print(f"Fleet spent ${ledger['spent'] / 100:.2f} of ${GLOBAL_CAP_CENTS / 100:.2f}")

05Budgeting and controlling costs

Why spend multiplies

Each agent carries its own context and re-sends all of it as input every turn, so five workers that load the same 8,000-token prompt pay for it five times unless the prefix is cached. Output costs five times input on every current model. Claude 4.7 and later models also produce about 30% more tokens for the same text, and declaring tools adds hidden prompt tokens (286 or more per request on the 5.5 models). Budget from measured usage, not character counts.

Explicit limits at every level

Set a cap at each of six levels, nest the lower ones inside the higher ones, and decide in advance what happens when each one trips. The figures are starting points, not recommendations.

The daily cap sits above monthly ÷ working days so one busy day does not stop the team; the weekly cap catches a sustained run of them. The annual cap sits below twelve monthly caps to force a review before year end. The small-team monthly figure stays under $1,000, the Build tier's monthly spend cap; Start caps at $500 and Scale at $200,000.

The inner caps react within one model response, while cloud billing alerts can take most of a day. That gap is why the inner caps must live in code.

Enforcing limits in code

Each response's usage object splits input into input_tokens, cache_creation_input_tokens and cache_read_input_tokens. Price each one, add it to a ledger shared by all agents, and check the ledger before every call. Persist daily and weekly totals in a database keyed by date. In the Agent SDK, max_budget_usd does this per query, subagents included. The Rust dispatcher below applies the same idea to the raw Messages API.

// src/main.rs: concurrent Claude dispatcher with a hard spend cap, written against the
// raw Messages API. Cargo deps: tokio = { version = "1", features = ["full"] },
// reqwest = { version = "0.12", features = ["json"] }, serde_json = "1", anyhow = "1"
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;

const MODEL: &str = "claude-haiku-5-5";
// Micro-dollars per token = USD per million tokens (Haiku 5.5, prompts up to 100k tokens).
const IN_PRICE: f64 = 0.10;
const OUT_PRICE: f64 = 0.50;
const CAP_MICRO_USD: u64 = 2_000_000; // $2.00 for the whole run
const MAX_PARALLEL: usize = 8;

struct Budget { spent: AtomicU64 } // micro-dollars, shared by every task

impl Budget {
    fn exhausted(&self) -> bool { self.spent.load(Ordering::Relaxed) >= CAP_MICRO_USD }
    fn charge(&self, input: u64, output: u64) -> u64 {
        let cost = (input as f64 * IN_PRICE + output as f64 * OUT_PRICE).ceil() as u64;
        self.spent.fetch_add(cost, Ordering::Relaxed) + cost
    }
}

async fn call(http: &reqwest::Client, key: &str, prompt: &str) -> anyhow::Result<(String, u64, u64)> {
    let body = serde_json::json!({
        "model": MODEL, "max_tokens": 400,
        "messages": [{ "role": "user", "content": prompt }]
    });
    for attempt in 0..5u32 {
        let resp = http.post("https://api.anthropic.com/v1/messages")
            .bearer_auth(key)
            .header("anthropic-version", "2026-10-21")
            .json(&body).send().await?;
        if resp.status().as_u16() == 429 || resp.status().is_server_error() {
            // Honour retry-after when present; spend-cap 429s carry none and will keep failing.
            let wait = resp.headers().get("retry-after")
                .and_then(|v| v.to_str().ok()).and_then(|s| s.parse::().ok())
                .unwrap_or(1u64 << attempt);
            tokio::time::sleep(Duration::from_secs(wait.min(60))).await;
            continue;
        }
        let v: serde_json::Value = resp.error_for_status()?.json().await?;
        let text = v["content"].as_array().into_iter().flatten()
            .filter(|b| b["type"] == "text")          // skip thinking blocks
            .filter_map(|b| b["text"].as_str()).collect::>().join("");
        let input = v["usage"]["input_tokens"].as_u64().unwrap_or(0);
        let output = v["usage"]["output_tokens"].as_u64().unwrap_or(0);
        return Ok((text, input, output));
    }
    anyhow::bail!("gave up after retries")
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let key = std::env::var("ANTHROPIC_API_KEY")?;
    let http = reqwest::Client::builder().timeout(Duration::from_secs(120)).build()?;
    let budget = Arc::new(Budget { spent: AtomicU64::new(0) });
    let gate = Arc::new(Semaphore::new(MAX_PARALLEL));
    let tickets: Vec = std::fs::read_to_string("tickets.txt")?
        .lines().map(str::to_owned).collect();

    let mut handles = Vec::new();
    for (i, t) in tickets.into_iter().enumerate() {
        let (http, key, budget, gate) = (http.clone(), key.clone(), budget.clone(), gate.clone());
        handles.push(tokio::spawn(async move {
            let _permit = gate.acquire_owned().await.ok()?;
            if budget.exhausted() { return None; } // stop starting new work at the cap
            let prompt = format!("Label this ticket as bug, feature or question:\n{t}");
            match call(&http, &key, &prompt).await {
                Ok((label, inp, out)) => {
                    let total = budget.charge(inp, out);
                    println!("#{i} {label} (in {inp}, out {out}, run ${:.4})", total as f64 / 1e6);
                    Some(label)
                }
                Err(e) => { eprintln!("#{i} failed: {e}"); None }
            }
        }));
    }
    let done = futures_count(handles).await;
    println!("{done} labelled, spent ${:.4}", budget.spent.load(Ordering::Relaxed) as f64 / 1e6);
    Ok(())
}

async fn futures_count(handles: Vec>>) -> usize {
    let mut n = 0;
    for h in handles { if let Ok(Some(_)) = h.await { n += 1; } }
    n
}

It sends the current anthropic-version, 2026-10-21, which also removes temperature, top_p and top_k. A 429 caused by a spend cap has no retry-after header and carries the code enforced_spend_limit_reached; treat it as a stop, not a retry.

Enforcing limits on the platform

Platform limits catch bugs in your own code. Set an organization spend limit below your tier cap on the Console Billing page; requests over it return HTTP 400. Run agents in their own workspace with its own spend and rate limits (the default workspace cannot take a spend limit). Reconcile with the Usage and Cost API, which reports usage in buckets as small as one minute, usually within five minutes of a request. AWS Budgets can alert and run budget actions but refreshes only up to three times a day, and Google Cloud budgets alert at 50%, 90% and 100% by default without stopping spend.

Cloud agents and their cost

Bedrock and Google Cloud invoice you directly at their own prices, published on the Bedrock pricing page and Google Cloud's pricing page. For Claude 4.5 models and later, regional and multi-region endpoints cost 10% more than global ones. Bedrock batch inference is 50% below on-demand, and Bedrock cut Sonnet 5.5 cache-read prices by 50% from 7 October 2026.

Claude Platform on AWS and Microsoft Foundry are Anthropic-operated and bill standard API rates through the cloud marketplace in $0.01 consumption units; US-only inference costs 1.1×.

Managed Agents charge standard token rates plus $0.08 per running session-hour, with no batch discount. Outside Managed Agents, the code execution tool includes 1,550 free hours a month, then costs $0.05 per container-hour.

Self-hosted on Amazon Bedrock AgentCore Runtime, compute is billed apart from tokens: $0.1276 per vCPU-hour and $0.0169 per GB-hour on v2, per second, with no CPU charge during I/O wait. Web search there costs $7 per 1,000 queries, logs bill at CloudWatch rates and data transfer at EC2 rates.

Cutting the bill

Caching: reads cost 0.1× base input on most models, 0.05× on Opus and Sonnet 5.5 and 0.025× on Fable 5.1, and do not count toward input rate limits. Start parallel agents after the first response begins so they can hit the cache. Batching: 50% off. Routing: Haiku 5.5 input costs one twentieth of Sonnet 5.5's. Trimming: a fetched documentation page runs about 25,000 tokens, so cap fetches with max_content_tokens. Capping: max_turns, lower effort and automatic compaction all cut tokens per run.

Worked example: a five-agent research run

One Opus 5.5 lead runs 6 turns with a 12,000-token cached prefix, 4,000 new input and 1,500 output tokens per turn. Four Sonnet 5.5 workers run 10 turns each with an 8,000-token cached prefix, 6,000 new input and 800 output tokens per turn, and 5 web searches.

Uncached, the same run costs $2.20, so caching saves 34%. At 20 runs a day that is $29.14 a day, $145.68 a week, $640.99 a month (22 working days) and $7,691.90 a year. That fits the small-team caps except one: twenty sessions at the $5 session cap could reach $100 against a $45 daily cap. Lower the session cap to $2.25, or accept that the daily cap stops work after nine worst-case runs.

06Observability, testing and failure handling

Log every turn: agent, parent, model, the four usage fields, stop reason, tool, latency and cost. Subagent messages carry a parent_tool_use_id that gives you the tree, and the Python SDK emits an OpenTelemetry span per API call once a tracer provider is registered. Anthropic credits full production tracing for diagnosing why its Research agents missed information.

Once those logs exist, paste a sample below to see which agent and which model the money went to.

Retries and timeouts: the Python SDK retries connection errors, 408, 409, 429 and 5xx twice with backoff and times out after ten minutes by default. Fleets need their own backoff on top, guided by retry-after when present.

Circuit breakers: after repeated failures, stop calling a tool for a while and return an is_error result saying so; Claude can then wait, switch tools or report the gap. Anthropic found that telling an agent a tool is failing "works surprisingly well".

Approval gates: anything that sends, merges, pays or deletes should pass a person first. The Agent SDK's default permission mode routes unapproved calls to your canUseTool callback, dontAsk denies them outright for headless agents, and bypassPermissions belongs only in isolated containers.

Testing: Anthropic began with about 20 real queries, because early prompt changes moved success rates enough for a small set to detect, and graded outputs with one LLM judge scoring accuracy, citations, completeness, source quality and tool efficiency, backed by human testers. It judges state-changing agents on their end state and ships changes with rainbow deployments so runs in flight keep their old version.

07Security notes

Least privilege: give each agent the smallest tool list its job needs. Omitted tools are absent from the session; a bare name in disallowed_tools removes a tool, and a scoped rule such as Bash(rm *) blocks only matching calls.

Secrets: load keys from environment variables or a secret manager, never from prompts, and run fleets in their own workspace with their own key; Anthropic recommends workspaces for separating environments and spend.

Retention: batch data is stored up to 29 days unless deleted. Managed Agents stores session history and sandbox state server-side and is not currently eligible for Zero Data Retention or HIPAA BAA coverage; sessions and files can be deleted through the API.

08Choosing an architecture

Start simple. Anthropic's advice is that one well-prompted call with retrieval is often enough and that complexity belongs only where it measurably helps. If you do need several agents, answer the five questions below.

09References

All links were opened and read on 22 October 2026.

1. Pricing, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
2. Models overview, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
3. Rate limits, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
4. Batch processing, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
5. Prompt caching, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
6. Subagents in the SDK, Claude Code Docs, Anthropic. Accessed 22 Oct 2026.
7. How the agent loop works, Claude Code Docs, Anthropic. Accessed 22 Oct 2026.
8. Track cost and usage, Claude Code Docs, Anthropic. Accessed 22 Oct 2026.
9. Give Claude custom tools, Claude Code Docs, Anthropic. Accessed 22 Oct 2026.
10. Create custom subagents, Claude Code Docs, Anthropic. Accessed 22 Oct 2026.
11. Claude Managed Agents overview, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
12. Get started with Claude Managed Agents, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
13. Session budgets, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
14. Session event stream, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
15. Inspect sessions and track usage, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
16. API overview, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
17. Versions, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
18. Python SDK, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
19. Usage and Cost API, Claude Platform Docs, Anthropic. Accessed 22 Oct 2026.
20. How we built our multi-agent research system, Engineering at Anthropic, J. Hadfield et al., 13 June 2025. Accessed 22 Oct 2026.
21. Building effective agents, Engineering at Anthropic, Erik S. and Barry Zhang, 19 December 2024. Accessed 22 Oct 2026.
22. What is the Model Context Protocol (MCP)?, modelcontextprotocol.io. Accessed 22 Oct 2026.
23. Amazon Bedrock pricing, Amazon Web Services. Accessed 22 Oct 2026.
24. Amazon Bedrock AgentCore pricing, Amazon Web Services. Accessed 22 Oct 2026.
25. Managing your costs with AWS Budgets, AWS Documentation. Accessed 22 Oct 2026.
26. Create, edit, or delete budgets and budget alerts, Google Cloud Documentation. Accessed 22 Oct 2026.
27. anthropic-sdk-python, Anthropic on GitHub. Accessed 22 Oct 2026.
28. Generative AI on Vertex AI pricing, Google Cloud. Accessed 22 Oct 2026.