Multi-Agent Pipeline Design — State Sharing and Error Propagation Between Claude Agent SDK Orchestrators and Subagents

Orchestrator
  ├── Task decomposition
  ├── Delegate to Subagent A ──▶ [Independent context] ──▶ Return result
  ├── Delegate to Subagent B ──▶ [Independent context] ──▶ Return result
  └── Synthesize results

import anthropic
import asyncio
import json
from typing import TypedDict
 
# Using AsyncAnthropic — does not block the event loop when running in parallel with asyncio.gather
async_client = anthropic.AsyncAnthropic()
 
# Recommended to start with a lower-cost model and swap as needed
MODEL = "claude-haiku-4-5-20251001"
 
class PipelineState(TypedDict):
    task: str
    plan: str | None
    output: str | None
    evaluation: str | None
    passed: bool
 
async def run_subagent(system_prompt: str, user_message: str) -> str:
    """Run a single subagent — returns only the result text"""
    response = await async_client.messages.create(
        model=MODEL,
        max_tokens=4096,
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}],
    )
    return response.content[0].text
 
async def planner_agent(task: str) -> str:
    return await run_subagent(
        system_prompt="You are a planner that designs task structure. Clearly describe the goal and a step-by-step approach.",
        user_message=f"Analyze the following task and write an execution plan:\n\n{task}",
    )
 
async def generator_agent(task: str, plan: str) -> str:
    return await run_subagent(
        system_prompt="You are an execution agent. Generate the deliverable according to the given plan.",
        user_message=f"Task: {task}\n\nPlan:\n{plan}\n\nWrite the deliverable according to the plan above.",
    )
 
async def evaluator_agent(task: str, output: str) -> tuple[bool, str]:
    result = await run_subagent(
        system_prompt=(
            "You are an independent quality verification agent. "
            "Determine whether the deliverable meets the task objectives, "
            "and respond only in the following JSON format:\n"
            '{"status": "pass" or "fail", "reason": "reason for judgment"}'
        ),
        user_message=f"Task: {task}\n\nDeliverable:\n{output}\n\nVerify the quality.",
    )
    try:
        data = json.loads(result)
        passed = data.get("status") == "pass"
    except json.JSONDecodeError:
        passed = False
    return passed, result
 
async def orchestrator(task: str) -> PipelineState:
    state: PipelineState = {
        "task": task,
        "plan": None,
        "output": None,
        "evaluation": None,
        "passed": False,
    }
 
    print("[Orchestrator] Running planner subagent...")
    state["plan"] = await planner_agent(task)
 
    print("[Orchestrator] Running generator subagent...")
    state["output"] = await generator_agent(task, state["plan"])
 
    print("[Orchestrator] Running evaluator subagent...")
    passed, evaluation = await evaluator_agent(task, state["output"])
    state["evaluation"] = evaluation
    state["passed"] = passed
 
    return state
 
async def main():
    result = await orchestrator(
        "Design and implement a simple to-do management CLI tool in Python."
    )
    print(f"\nFinal result — PASS: {result['passed']}")
 
asyncio.run(main())

import asyncio
import json
import time
from typing import Any
 
import anthropic
 
async_client = anthropic.AsyncAnthropic()
MODEL = "claude-haiku-4-5-20251001"
 
# In-memory KV store (replace with Redis or equivalent in production)
_store: dict[str, Any] = {}
 
def kv_set(key: str, value: Any) -> None:
    _store[key] = {"value": value, "timestamp": time.time()}
 
def kv_get(key: str) -> Any | None:
    entry = _store.get(key)
    return entry["value"] if entry else None
 
async def compliance_check_agent(run_id: str, domain: str, task: str) -> dict:
    """Per-domain compliance check subagent"""
    response = await async_client.messages.create(
        model=MODEL,
        max_tokens=512,
        system=(
            f"You are a {domain} compliance expert. "
            "Review the given transaction data and determine whether there are any issues. "
            'Respond only in JSON: {"status": "pass" or "fail", "issues": ["list of issues"]}'
        ),
        messages=[{"role": "user", "content": f"Transaction data:\n{task}"}],
    )
 
    try:
        parsed = json.loads(response.content[0].text)
    except json.JSONDecodeError:
        parsed = {"status": "error", "issues": ["Failed to parse response"]}
 
    result = {
        "domain": domain,
        "status": parsed.get("status", "error"),
        "issues": parsed.get("issues", []),
        "checked_at": time.time(),
    }
    kv_set(f"{run_id}:compliance:{domain}", result)
    print(f"  [{domain}] Check complete → saved to KV")
    return result
 
async def report_agent(run_id: str, domains: list[str]) -> str:
    """Collect results from KV and generate final report"""
    results = []
    for domain in domains:
        result = kv_get(f"{run_id}:compliance:{domain}")
        if result:
            results.append(result)
 
    passed = sum(1 for r in results if r["status"] == "pass")
    report = (
        f"Compliance check results: {passed}/{len(results)} passed\n"
        + "\n".join(f"- {r['domain']}: {r['status']}" for r in results)
    )
    return report
 
async def financial_compliance_orchestrator(transaction_data: str) -> str:
    run_id = f"run_{int(time.time())}"
    domains = ["AML", "KYC", "FATF"]
 
    print(f"[Orchestrator] Starting parallel compliance checks (run_id={run_id})")
 
    # Run independent checks in parallel
    await asyncio.gather(
        *[
            compliance_check_agent(run_id, domain, transaction_data)
            for domain in domains
        ]
    )
 
    print("[Orchestrator] Running report agent")
    return await report_agent(run_id, domains)
 
asyncio.run(financial_compliance_orchestrator("Sample transaction data"))

import asyncio
import time
from enum import Enum
 
class CircuitOpenError(RuntimeError):
    """Dedicated exception raised when the Circuit Breaker is in OPEN state"""
    pass
 
class CircuitState(Enum):
    CLOSED = "closed"        # Normal operation
    OPEN = "open"            # Tripped, requests blocked
    HALF_OPEN = "half_open"  # Probing for recovery
 
class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time: float | None = None
        self.state = CircuitState.CLOSED
 
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return False
        return time.time() - self.last_failure_time >= self.recovery_timeout
 
    async def call(self, coro):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                print("[Circuit Breaker] HALF_OPEN — attempting recovery")
            else:
                raise CircuitOpenError("Circuit Breaker OPEN: request blocked")
 
        try:
            result = await coro
            self.failure_count = 0
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                print("[Circuit Breaker] CLOSED — recovery complete")
            return result
 
        except CircuitOpenError:
            raise
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(
                    f"[Circuit Breaker] OPEN — {self.failure_count} failures, "
                    f"retry available after {self.recovery_timeout}s"
                )
            raise e
 
async def agent_with_retry(
    agent_fn,
    breaker: CircuitBreaker,
    max_retries: int = 3,
    base_delay: float = 1.0,
):
    """Exponential backoff retry + Circuit Breaker combination"""
    for attempt in range(max_retries):
        try:
            return await breaker.call(agent_fn())
        except CircuitOpenError:
            raise  # Do not retry on Circuit Breaker trip
        except Exception:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"  Retry {attempt + 1}/{max_retries}, waiting {delay:.1f}s")
            await asyncio.sleep(delay)