Claude 950-Agent Swarm: Build a Funnel in Python

On September 24, Anthropic published a write-up that spread fast: about 950 Claude agents ran for 21 hours, burned roughly 210 million tokens, and turned up a previously undescribed enzyme system in bacteriophage genomes. Anthropic calls it array-associated reverse transcriptases (ART): a reverse transcriptase, a partner gene, and evenly spaced DNA repeats that look a bit like a CRISPR array. Its function is still unknown, and the result is a preprint that hasn't been peer reviewed.

The biology is the headline. The part you can steal is the shape of the workload. Anthropic's own numbers describe a funnel: 200,000+ reverse transcriptases pulled from public databases, about 3,500 candidate systems, 20 worth a human-readable report, and 1 that went to the lab. Most of the agents did cheap, wide, disposable work. Only a handful of expensive judgments happened at the end.

In this guide you'll build that funnel in Python with the Anthropic SDK: a cheap triage stage on Haiku 4.5, a structured deep read on Sonnet 5.5, and an adversarial report on Opus 5.5. It has a hard dollar cap, bounded concurrency, and a JSONL checkpoint so a crash at hour 18 doesn't cost you hours 1 through 17. Anthropic hasn't published its harness, so this is my reconstruction of the pattern, not their code. We'll test it offline with a mock client, then point it at the real API.

What you need

• Python 3.10+, pip install anthropic pydantic, and an ANTHROPIC_API_KEY (only for the real run; the mock needs no key).
• Some comfort with asyncio. We use gather and a semaphore, nothing exotic.
• A pile of items to screen. The worked example uses 400 synthetic research-note abstracts with 5 planted needles, so you can check that the funnel actually finds them.

Why a funnel beats N identical agents

The tempting design is to give every item to your best model. It works, and it's about 10x more expensive than it needs to be, because roughly 99% of candidates in a search like this are boring. A funnel spends model quality where the signal is.

Stage	Model	Job	Rough price per MTok (in/out)
1 Triage	claude-haiku-4-5-20251001	Score every item 0-10, one sentence why	$1 / $5
2 Deep read	claude-sonnet-5-5	Extract the claim, evidence for and against	$2 / $10
3 Report	claude-opus-5-5	Adversarial one-pager per finalist	$4 / $20

Prices and model IDs are from the models overview as of September 29, 2026. Check them before you budget, they move.

Two design rules carry most of the value. First, every stage writes its results to disk as it goes, so any stage can resume. Second, a failed agent is logged and skipped, never fatal. With hundreds of parallel calls, some will fail, and the run has to treat that as normal.

Step 1: Structured outputs for every agent

Free-text scores are a parsing nightmare at this scale. Claude's structured outputs feature constrains the reply to a JSON schema through output_config.format. Define the shapes with Pydantic and hand the schema over. Constraints like min and max aren't something to rely on inside the grammar, so we strip them from the schema and let Pydantic enforce them when we validate the reply.

PRICES = {  # USD per million tokens (input, output), from the models overview page
    "claude-haiku-4-5-20251001": (1, 5),
    "claude-sonnet-5-5": (2, 10),
    "claude-opus-5-5": (4, 20),
}

class Triage(BaseModel):
    score: int = Field(ge=0, le=10)
    reason: str

class DeepRead(BaseModel):
    score: int = Field(ge=0, le=10)
    novel_claim: str
    evidence_for: list[str]
    evidence_against: list[str]

def schema_for(model_cls):
    s = model_cls.model_json_schema()
    s["additionalProperties"] = False
    # drop min/max constraints the structured-output grammar may not support;
    # Pydantic re-validates them on our side anyway.
    for p in s["properties"].values():
        p.pop("minimum", None); p.pop("maximum", None)
    return {"type": "json_schema", "schema": s}

Note additionalProperties: False at the top level, which the API expects for object schemas.

Step 2: A budget that actually stops the swarm

The failure mode that hurts is not a bug, it's a loop that keeps spending. Every response carries usage.input_tokens and usage.output_tokens, so we can convert to dollars after each call and refuse to start new calls once the cap is hit.

@dataclass
class Budget:
    cap_usd: float
    spent: float = 0.0
    calls: int = 0
    def charge(self, model, usage):
        i, o = PRICES[model]
        self.spent += (usage.input_tokens * i + usage.output_tokens * o) / 1e6
        self.calls += 1
    @property
    def exhausted(self): return self.spent >= self.cap_usd

class BudgetExceeded(Exception): pass

Because the check happens before each call, in-flight requests can overshoot the cap slightly. With 25 concurrent Haiku calls that's cents. If your calls are large, lower the concurrency or reserve headroom.

Step 3: One function that every agent calls

async def ask(client, budget, sem, model, system, user, out_cls=None, max_tokens=800):
    if budget.exhausted:
        raise BudgetExceeded(f"spent ${budget.spent:.2f}")
    kwargs = dict(model=model, max_tokens=max_tokens, system=system,
                  messages=[{"role": "user", "content": user}])
    if out_cls:
        kwargs["output_config"] = {"format": schema_for(out_cls)}
    async with sem:                       # cap concurrent agents
        resp = await client.messages.create(**kwargs)
    budget.charge(model, resp.usage)
    text = next(b.text for b in resp.content if b.type == "text")
    return out_cls.model_validate_json(text) if out_cls else text

The semaphore is your concurrency ceiling. It bounds simultaneous requests, which keeps you under rate limits and stops one stage from starving the next. Retries for 429s are handled by the SDK: AsyncAnthropic(max_retries=5) honors the retry-after header, so you don't write your own backoff for the common case.

Step 4: Checkpointed stages

This is the piece that makes a 21-hour run survivable. Each stage appends one JSON line per finished item. On restart it reads the file and only runs what's missing.

def load_done(path):
    if not path.exists(): return {}
    return {json.loads(l)["id"]: json.loads(l) for l in path.read_text().splitlines() if l}

async def run_stage(name, items, worker, ckpt_dir):
    """Run worker(item) for every item; skip ids already in the JSONL checkpoint."""
    path = Path(ckpt_dir) / f"{name}.jsonl"
    done = load_done(path)
    todo = [x for x in items if x["id"] not in done]
    print(f"[{name}] {len(done)} cached, {len(todo)} to run")
    async def one(item):
        try:
            res = await worker(item)
        except BudgetExceeded:
            return None
        except Exception as e:            # one bad agent must not sink the swarm
            print(f"[{name}] {item['id']} failed: {type(e).__name__}: {e}", file=sys.stderr)
            return None
        rec = {"id": item["id"], **res}
        with path.open("a") as f: f.write(json.dumps(rec) + "\n")
        return rec
    new = [r for r in await asyncio.gather(*(one(x) for x in todo)) if r]
    done.update({r["id"]: r for r in new})
    return done

The except Exception inside one() is deliberate. In the first offline run below, the simulator injects a 2% failure rate; those items are simply absent from the checkpoint, and the next run picks them up.

Step 5: The three stages and the worked example

Now the actual prompts and workers. The worked example asks: which of these research-note abstracts contains a surprising, testable claim nobody has reproduced? Swap the prompts and the dataset for your own problem, such as security advisories, support tickets, contracts, or papers.

# ---------- worked example: screen abstracts for unreplicated, surprising claims ----------
TRIAGE_SYS = ("You screen research-note abstracts. Score 0-10 how likely the note contains a "
              "surprising, testable claim that nobody has reproduced. Be stingy: most notes are 0-3.")
DEEP_SYS = ("You are a skeptical reviewer. Extract the single novel claim, then list evidence for "
            "and against it using only the text provided. Never invent citations.")
REPORT_SYS = ("Write a one-page adversarial report: the claim, the cheapest experiment that could "
              "falsify it, and what would change your mind. Under 250 words.")

async def triage(client, budget, sem, item):
    r = await ask(client, budget, sem, "claude-haiku-4-5-20251001", TRIAGE_SYS, item["text"], Triage, 200)
    return r.model_dump()

async def deep(client, budget, sem, item, texts):
    r = await ask(client, budget, sem, "claude-sonnet-5-5", DEEP_SYS, texts[item["id"]], DeepRead, 700)
    return r.model_dump()

async def report(client, budget, sem, item, texts):
    t = await ask(client, budget, sem, "claude-opus-5-5", REPORT_SYS,
                  f"{texts[item['id']]}\n\nReviewer notes: {json.dumps(item)}", None, 600)
    return {"report": t}

Triage prompts should be stingy on purpose. If you say 'score how interesting this is', a cheap model will hand out 6s to everything and your funnel never narrows. Telling it that most items are 0-3 fixes that better than any threshold tuning.

def make_dataset(n, needles=5, seed=7):
    rng = random.Random(seed)
    rows = []
    for i in range(n):
        rows.append({"id": f"n{i:04d}", "text": f"Note {i}: routine replication of a known effect, "
                     f"n={rng.randint(20,900)}, no deviations from prior work."})
    for k in rng.sample(range(n), needles):
        rows[k]["text"] = (f"Note {k}: repeating spacer sequences flank an unusual reverse "
                           "transcriptase in phage genomes; short RNAs are produced; function unknown.")
    return rows

async def main(args):
    if args.mock:
        from mock_client import MockAsyncClient
        client = MockAsyncClient()
    else:
        from anthropic import AsyncAnthropic
        client = AsyncAnthropic(max_retries=5)
    Path(args.ckpt).mkdir(exist_ok=True)
    rows = make_dataset(args.n)
    texts = {r["id"]: r["text"] for r in rows}
    budget = Budget(args.budget)
    sem = asyncio.Semaphore(args.concurrency)

    s1 = await run_stage("s1_triage", rows, lambda x: triage(client, budget, sem, x), args.ckpt)
    keep1 = [{"id": k, **v} for k, v in s1.items() if v["score"] >= args.t1]
    print(f"  funnel: {len(rows)} -> {len(keep1)}  (spent ${budget.spent:.4f})")

    s2 = await run_stage("s2_deep", keep1, lambda x: deep(client, budget, sem, x, texts), args.ckpt)
    keep2 = sorted(({"id": k, **v} for k, v in s2.items() if v["score"] >= args.t2),
                   key=lambda r: -r["score"])[:args.top]
    print(f"  funnel: {len(keep1)} -> {len(keep2)}  (spent ${budget.spent:.4f})")

    s3 = await run_stage("s3_report", keep2, lambda x: report(client, budget, sem, x, texts), args.ckpt)
    print(f"  finalists: {len(s3)}  calls={budget.calls}  total=${budget.spent:.4f}")
    for r in list(s3.values())[:1]:
        print("\n--- sample report ---\n" + r["report"])

That last main() wires the stages: filter by --t1, filter and cap by --t2 and --top, then report. Save the whole thing as funnel.py.

Step 6: Test it offline first

Don't let your first run of a swarm be against a live API and a real budget. A 20-line mock that mimics client.messages.create lets you test the funnel logic, the checkpointing, and the failure handling for free.

# mock_client.py
import json, asyncio, random
from types import SimpleNamespace as NS

class MockAsyncClient:
    def __init__(self, fail_rate=0.02):
        self.messages = NS(create=self._create)
        self.rng = random.Random(1); self.fail_rate = fail_rate
    async def _create(self, model, max_tokens, system, messages, output_config=None):
        await asyncio.sleep(0.001)
        if self.rng.random() < self.fail_rate:
            raise RuntimeError("simulated 529 overloaded")
        text_in = messages[0]["content"]
        hit = "reverse transcriptase" in text_in
        props = (output_config or {}).get("format", {}).get("schema", {}).get("properties", {})
        if "reason" in props:
            out = json.dumps({"score": 9 if hit else self.rng.choice([0,1,2,3,5]), "reason": "mock"})
        elif "novel_claim" in props:
            out = json.dumps({"score": 8 if hit else 3, "novel_claim": "spacer-flanked RT makes short RNAs",
                              "evidence_for": ["RNAs produced"], "evidence_against": ["function unknown"]})
        else:
            out = "CLAIM: spacer-flanked RT systems are a new class."
        n_in = len(text_in)//4 + len(system)//4
        return NS(content=[NS(type="text", text=out)],
                  usage=NS(input_tokens=n_in, output_tokens=len(out)//4))

Run it twice. The output below is from the mock, not from Claude, so the dollar figures are meaningless. What matters is the shape.

$ python funnel.py --mock
[s1_triage] 0 cached, 400 to run
[s1_triage] n0012 failed: RuntimeError: simulated 529 overloaded
... (13 failures total)
  funnel: 400 -> 89  (spent $0.0372)
[s2_deep] 0 cached, 89 to run
  funnel: 89 -> 4  (spent $0.0785)
[s3_report] 0 cached, 4 to run
  finalists: 4  calls=479  total=$0.0829

$ python funnel.py --mock          # second run, same checkpoint dir
[s1_triage] 387 cached, 13 to run
  funnel: 400 -> 91  (spent $0.0012)
[s2_deep] 88 cached, 3 to run
  funnel: 91 -> 5  (spent $0.0026)
[s3_report] 4 cached, 1 to run
  finalists: 5  calls=16  total=$0.0037

Look at the finalists: 4 on the first run, 5 on the second. There were 5 planted needles, and one of them hit a simulated 529 in stage 1 and was silently dropped. The rerun spent 16 calls, not 479, and recovered it. That's the behavior you want at 950 agents: partial failure costs you a retry, not a restart. It's also a warning. A funnel that skips failed items will under-report unless you look at the failure count, so print it.

Step 7: Point it at the real API

export ANTHROPIC_API_KEY=sk-ant-...
python funnel.py --n 400 --budget 2.00 --concurrency 20

Start with a small --n and a budget you'd shrug at. Read the JSONL files before scaling: cat ckpt/s1_triage.jsonl | head. If stage 1 is scoring more than about 30% of items above your threshold, your triage prompt is too generous, and stage 2 will eat the budget. Tighten the prompt, delete ckpt/s1_triage.jsonl, and rerun.

Pitfalls I'd check before trusting the results

• Checkpoint keys must be stable. The checkpoint is keyed on id. If your ids are generated per run (uuid4), resume silently reruns everything. Derive them from the content or the source row.
• Changing a prompt without clearing the checkpoint. Cached results were produced by the old prompt. Keep the prompt version in the file name, like s1_triage_v2.jsonl.
• Overshoot on the budget. The cap is checked before each call, so concurrent in-flight calls can push you past it. It's cents at Haiku sizes, more if stage 3 prompts are big.
• Silent failures. The worker swallows errors so one bad item doesn't sink the run. Count them and print the count at the end of each stage, or you'll never know your recall dropped.
• Triage recall is your ceiling. Stage 3 can't rescue an item stage 1 dropped. Audit by planting known positives (as the example does) and by sampling a random slice of the rejected pile for a stronger model to re-score.
• Rate limits are per model. A wide Haiku stage can be fine while the Opus stage hits 429s. The SDK retries with retry-after, but keep the semaphore lower for the expensive stage.
• Don't confuse ranking with truth. The model's score orders candidates. It doesn't validate them. Anthropic's pipeline ended with a wet lab, and its own write-up says the enzyme's function is still unknown. Your equivalent of the wet lab is a human check or an executable test.
• Structured output constraints. Not every JSON Schema keyword is supported in the grammar. Keep schemas plain, and validate ranges with Pydantic after the fact, as we did.

Quick reference

Knob	Where	What it does
--budget	Budget.cap_usd	Stops new calls once estimated spend hits the cap
--concurrency	asyncio.Semaphore	Max simultaneous requests
--t1 / --t2	threshold flags	Score cutoffs that shape the funnel width
--top	stage 2 to 3 cut	Max finalists sent to the expensive model
max_retries	AsyncAnthropic(...)	SDK retries 429/5xx, honors retry-after (default 2)
ckpt/*.jsonl	run_stage	One line per finished item; delete to force a rerun
output_config.format	ask()	JSON-schema constrained replies

Where to go next

• Add a fourth stage that runs code: have the survivors generate a check (a query, a test, a simulation) and execute it in a sandbox, so the last filter is evidence rather than opinion.
• Swap stage 1 for the Message Batches API if you don't need results immediately. Batch pricing is discounted, and a triage stage over tens of thousands of items is the ideal shape for it.
• Add prompt caching for the system prompts, since every call in a stage shares one.
• Read Anthropic's original write-up for the biology and its stated caveats, and compare their funnel numbers to yours.

---

Sources: Anthropic, Claude discovers a novel enzyme system (Sept 24, 2026); Anthropic API docs for models overview, structured outputs and rate limits (platform.claude.com/docs). The funnel code in this guide is original and was tested against a mock client, not the live API.