Agentic Orchestration Basics

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        allowed_tools=["file_edit"],
        subagents=[
            SubagentConfig(
                name="changelog-writer",
                description="Writes a CHANGELOG.md entry from a diff summary.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Use the changelog-writer subagent to add a CHANGELOG.md entry for this release.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Result: {message['result']}")
 
asyncio.run(main())

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Summarizes frontend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Summarizes backend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Use both summary subagents, then write one combined PR description.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

import asyncio
from claude_agent_sdk import query, AgentOptions
 
async def summarize_then_translate(text: str, language: str) -> str:
    summary_chunks = []
    async for message in query(prompt=f"Summarize in three bullets:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Translate this into {language}, keep it as three bullets:\n\n{summary}"
    async for message in query(prompt=prompt):
        if message.get("type") == "text":
            translated_chunks.append(message["text"])
    return "".join(translated_chunks)
 
print(asyncio.run(summarize_then_translate("...", "Spanish")))

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_ticket(ticket_text: str) -> None:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="bug-triager",
                description="Triages bug reports: reproduces, labels severity.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Scopes feature requests into a rough implementation plan.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Read this ticket, decide if it's a bug or a feature request, "
        f"and route it to the matching subagent:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("Login button does nothing on Safari 18."))

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def audit_packages(packages: list[str]) -> None:
    subagents = [
        SubagentConfig(
            name=f"audit-{pkg}",
            description=f"Checks the {pkg} package for outdated dependencies.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Run all audit subagents for {packages} and list findings per package."
 
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            print(f"[{message['subagent_name']}] {message['result']}")
 
asyncio.run(audit_packages(["billing", "auth", "search"]))

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Researches one topic; may not delegate further.",
    allowed_tools=["web_search"],
    # No `subagents` field on the worker itself: it has no delegation
    # capability of its own, so depth is capped at one level here.
)
 
options = AgentOptions(subagents=[worker])

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def run_with_retry(prompt: str, options: AgentOptions, attempts: int = 3):
    last_error = None
    for attempt in range(attempts):
        try:
            results = []
            async for message in query(prompt=prompt, options=options):
                if message.get("type") == "subagent_result":
                    results.append(message["result"])
            return results
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(2 ** attempt)  # backoff before retrying
    raise RuntimeError(f"Failed after {attempts} attempts") from last_error

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Small change: route straight to one reviewer, no fan-out needed.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Reviews a single file change.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Use the reviewer subagent on {changed_files[0]}."
    else:
        # Larger change: fan out one reviewer per file, in parallel.
        options = AgentOptions(
            subagents=[
                SubagentConfig(
                    name=f"reviewer-{i}",
                    description=f"Reviews {f}.",
                    allowed_tools=["file_edit"],
                )
                for i, f in enumerate(changed_files)
            ]
        )
        prompt = f"Use all reviewer subagents to review {changed_files} in parallel."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

import asyncio
import time
from claude_agent_sdk import query, AgentOptions
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_after: float = 60.0):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_after = reset_after
        self.opened_at: float | None = None
 
    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.monotonic() - self.opened_at > self.reset_after:
            self.opened_at = None  # half-open: allow one attempt through
            self.failures = 0
            return False
        return True
 
    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()
 
    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None
 
breaker = CircuitBreaker()
 
async def call_flaky_subagent(prompt: str, options: AgentOptions):
    if breaker.is_open():
        raise RuntimeError("Circuit open: skipping call, dependency looked unhealthy")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
def looks_reasonable(result: str) -> bool:
    return len(result.strip()) > 20 and "TODO" not in result
 
async def orchestrate_with_check(prompt: str, options: AgentOptions) -> str:
    final_result = ""
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            final_result = message["result"]
 
    if not looks_reasonable(final_result):
        raise ValueError("Orchestrator result failed the sanity check")
    return final_result