How do I run agents in parallel and cancel all of them if one fails?

You want to do several things at once and stop paying for the rest the moment one of them goes wrong.

When you'd want this

Fan out a Planner's sub-questions across N Researchers. Race two implementations of the same skill and keep the first. Kick off a batch of independent enrichment tasks. In each case you want two properties the raw asyncio.gather doesn't give you:

• Sibling cancellation on failure. If one Researcher blows up, the others should stop; you don't want to keep paying to finish work you're about to throw away.
• A shared budget and cancel token. Cost accrues on the parent's Budget; an external cancel (operator hit stop) propagates to every child at every safe point.

run_agents(...) is the primitive. It builds one ctx.child() per pair (depth+1, sharing budget / services / cancel by reference) and calls gather_bounded(...) under the parent's semaphore.

Assumes ANTHROPIC_API_KEY in the environment

The successful children run on providers.claude(...). To demonstrate sibling cancellation without hoping the model misbehaves, the failing child uses a tiny FailingCognition that raises immediately — cognitions are pure Protocol impls, so a ~5-line one is a legitimate way to script a failure inside a real wire-up. Swap providers.claude for providers.openai (and set OPENAI_API_KEY) if that's what you have.

To run it with no key at all, replace the providers.claude(...) call with FakeLLM("a short answer") from agentkit.core.testing. Every other line is identical — that substitution is how this snippet is verified.

Working code

"""Requires ANTHROPIC_API_KEY in the environment."""

import asyncio
import os
from collections.abc import AsyncIterator

from agentkit.agents import Agent
from agentkit.core.execution.concurrency import CancellationToken, run_agents
from agentkit.core.contracts.types import Scope, StreamEvent
from agentkit.models.providers import frontiers as providers
from agentkit.core.runtime import Invoker, RunContext, Services


class FailingCognition:
    """A five-line cognition that raises immediately.

    `Cognition` is a runtime-checkable Protocol — any object with a
    `name: str` and a `drive(...)` async generator satisfies it, so
    scripting a scripted-failure agent doesn't need a test double."""

    name = "failing"

    async def drive(self, agent, task, ctx, context) -> AsyncIterator[StreamEvent]:
        raise RuntimeError("scripted failure")
        yield  # unreachable — satisfies the async-generator signature


def build_services() -> Services:
    llm = providers.claude(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        model="claude-sonnet-4-6",
    )
    return Services(invoker=Invoker(llm=llm))


async def demo_sibling_cancel() -> None:
    """One failing agent cancels the rest — TaskGroup semantics."""
    ctx = RunContext(correlation_id="run-1", scope=Scope(), services=build_services())
    a = Agent(name="a", model="claude-sonnet-4-6", prompt="Brief the topic in one line.")
    b = Agent(name="b", model="claude-sonnet-4-6", prompt="unused", cognition=FailingCognition())
    try:
        await run_agents([(a, "topic-a"), (b, "topic-b")], ctx)
    except* RuntimeError as eg:
        for exc in eg.exceptions:
            print(f"[sibling-cancel] caught: {exc}")


async def demo_best_effort() -> None:
    """`best_effort=True` isolates failures into Failure objects."""
    ctx = RunContext(correlation_id="run-2", scope=Scope(), services=build_services())
    a = Agent(name="a", model="claude-sonnet-4-6", prompt="Brief the topic in one line.")
    b = Agent(name="b", model="claude-sonnet-4-6", prompt="unused", cognition=FailingCognition())
    results = await run_agents([(a, "topic-a"), (b, "topic-b")], ctx, best_effort=True)
    for r in results:
        print(f"[best-effort] {type(r).__name__}: {r}")


async def demo_external_cancel() -> None:
    """External signal trips the token; the agent unwinds cooperatively."""
    from agentkit.core.execution.concurrency import Cancelled

    ctx = RunContext(
        correlation_id="run-3",
        scope=Scope(),
        services=build_services(),
        cancel=CancellationToken(),
    )

    async def cancel_after(delay: float) -> None:
        await asyncio.sleep(delay)
        assert ctx.cancel is not None
        ctx.cancel.cancel()

    async def long_running() -> None:
        for _ in range(20):
            await asyncio.sleep(0.02)
            ctx.check_cancelled()

    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(cancel_after(0.05))
            tg.create_task(long_running())
    except* Cancelled as eg:
        print(f"[external-cancel] observed: {[type(e).__name__ for e in eg.exceptions]}")


async def main() -> None:
    await demo_sibling_cancel()
    await demo_best_effort()
    await demo_external_cancel()


if __name__ == "__main__":
    asyncio.run(main())

How it works

There are only two things to decide, and they are about what should happen when one of the children fails.

Should the others keep going? If you are fanning out five parts of one answer, no — a missing part makes the answer wrong, so stop everything and report. If you are enriching a thousand database rows, yes — one bad row should not discard nine hundred good ones.

That is the whole difference between the two modes below. The first cancels the siblings on the first failure; the second isolates each failure into its own slot and lets the rest finish.

run_agents(pairs, ctx) — the default mode. Builds per-child contexts, runs each agent.run(task, child_ctx) inside an asyncio.TaskGroup bounded by ctx.semaphore(). TaskGroup semantics give sibling-cancellation for free: the first failure raises, outstanding tasks are cancelled, and the raised exceptions are grouped into an ExceptionGroup you catch with except*.

run_agents(pairs, ctx, best_effort=True) — for batches where one failure must NOT sink the others. Each slot in the returned list is either an AgentResult or a Failure (from agentkit.core.contracts.errors). Failure carries the exception on .cause, the classified category on .category, and a source string naming the slot — so a caller can retry, route around, or escalate uniformly instead of guessing what a raw exception meant.

CancellationToken is cooperative — a well-behaved async loop checks ctx.check_cancelled() at every safe point (loop top, between steps, between tool calls). Cancelling the token doesn't kill running coroutines; it makes the next check raise Cancelled. run_agents shares the token by reference across all children, so cancelling the parent cancels the subtree.

What bites people

• ExceptionGroup isn't a plain exception. Use except* (Python 3.11+, agentkit requires 3.12) or iterate eg.exceptions inside a plain except ExceptionGroup.
• Cooperative cancel means points matter. A synchronous CPU-bound loop inside a tool never checks the token. Yield an await asyncio.sleep(0) or check ctx.check_cancelled() on your own iteration boundaries. Blocking I/O belongs on a ProcessPoolExecutor or a threadpool so the event loop stays reactive.
• Budget.max_concurrency bounds fan-out, not run_agents. The semaphore lives on Budget and defaults to 8; if you fan out 100 children they'll still run 8 at a time. Increase Budget(max_concurrency=...) if you need more.
• best_effort=True still charges the shared budget. A Failure-in-a-slot doesn't refund the cost of the LLM calls that raced to error.

Related

• Concepts · Agents — where run_agents fits alongside Workflow and the coordinator cognitions.
• Cap spend with Budget and Quota — the ceiling that stops a runaway fan-out.
• agentkit.core.execution.concurrency module docstring — the full concurrency surface (gather_bounded, gather_best_effort, run_sync).