Error Handling¶

Handle agent failures gracefully. Catch structured errors, retry transient failures, and fall back to alternative agents.

The Code¶

import asyncio
import random

from pydantic import BaseModel

from openagentmesh import (
    AgentMesh,
    AgentSpec,
    HandlerError,
    InvalidInput,
    MeshError,
    NotFound,
)


class SummarizeInput(BaseModel):
    text: str


class SummarizeOutput(BaseModel):
    summary: str


async def call_with_retry(mesh: AgentMesh, agent: str, payload, retries: int = 3, base_delay: float = 0.1):
    for attempt in range(retries):
        try:
            return await mesh.call(agent, payload)
        except (NotFound, InvalidInput):
            # Caller faults: retrying with the same input won't help.
            raise
        except MeshError as e:
            if attempt == retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"  Attempt {attempt + 1} failed ({e.code}), retrying in {delay}s")
            await asyncio.sleep(delay)


async def call_with_fallback(mesh: AgentMesh, agents: list[str], payload):
    last_error: MeshError | None = None
    for agent in agents:
        try:
            return await mesh.call(agent, payload, timeout=5.0)
        except InvalidInput:
            # Bad payload: every fallback will reject it the same way.
            raise
        except (NotFound, HandlerError) as e:
            last_error = e
            print(f"  {agent} failed ({e.code}), trying next...")
            continue
    assert last_error is not None, "agents list must not be empty"
    raise last_error


async def main(mesh: AgentMesh) -> None:
    @mesh.agent(AgentSpec(name="nlp.summarizer", description="Summarizes text. May fail under load."))
    async def summarize(req: SummarizeInput) -> SummarizeOutput:
        if random.random() < 0.3:
            raise RuntimeError("LLM provider timeout")
        return SummarizeOutput(summary=f"Summary of: {req.text[:50]}")

    @mesh.agent(AgentSpec(name="nlp.basic-summarizer", description="Simple fallback summarizer."))
    async def basic_summarize(req: SummarizeInput) -> SummarizeOutput:
        return SummarizeOutput(summary=req.text[:80] + "...")

    # Pattern 1: Basic error handling
    try:
        result = await mesh.call("nlp.summarizer", SummarizeInput(text="Long document about AI agents"))
        print(f"  Success: {result['summary']}")
    except MeshError as e:
        print(f"  Error: [{e.code}] {e}")

    # Pattern 2: Retry with backoff
    result = await call_with_retry(mesh, "nlp.summarizer", SummarizeInput(text="Important document"))
    print(f"  Success: {result['summary']}")

    # Pattern 3: Fallback agent
    result = await call_with_fallback(
        mesh,
        agents=["nlp.summarizer", "nlp.basic-summarizer"],
        payload=SummarizeInput(text="Critical document that must be processed"),
    )
    print(f"  Success: {result['summary']}")

Run It¶

import asyncio
from openagentmesh import AgentMesh

async def run():
    async with AgentMesh.local() as mesh:
        await main(mesh)

asyncio.run(run())

How It Works¶

Key properties:

Structured errors everywhere. Every failure produces a MeshError subclass with code, agent, and request_id. No raw exceptions leak through the mesh.
Branch on type, not strings. NotFound and InvalidInput are caller faults — retrying won't help. HandlerError and MeshTimeout are potentially transient. Catching subclasses is more robust than matching e.code strings.
Fallback across agents. When multiple agents can handle the same task, try them in preference order. The catalog and contracts enable this pattern at runtime via mesh.discover().
Dead-letter stream. Every error is published to mesh.errors.{name} regardless of whether the caller handles it. Subscribe for monitoring, alerting, or audit logging.

Error Monitor¶

Subscribe to the dead-letter subject for real-time error observability:

import asyncio
import json
import nats

async def monitor():
    nc = await nats.connect("nats://localhost:4222")

    async def on_error(msg):
        error = json.loads(msg.data)
        print(f"[{error['agent']}] {error['code']}: {error['message']}")

    await nc.subscribe("mesh.errors.>", cb=on_error)

    print("Monitoring errors... (Ctrl+C to stop)")
    while True:
        await asyncio.sleep(1)

asyncio.run(monitor())