Reliability Engineering Basics

8 examples to get you started with agent reliability - 5 basic and 3 intermediate.

You will wrap a flaky tool with retries and backoff, classify failures, budget attempts, short-circuit on non-retryable errors, and sketch a tiny circuit-open flag.

Prerequisites

Python 3.11+ recommended
Comfort with calling a function that can raise or time out
No paid APIs required for these sketches

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Call a Flaky Tool Without Protection

Start from the failure you will fix: a tool that sometimes raises.

import random
 
 
def flaky_fetch_order(order_id: int) -> dict:
    if random.random() < 0.7:
        raise ConnectionError("upstream_unavailable")
    return {"order_id": order_id, "status": "shipped"}
 
 
try:
    print(flaky_fetch_order(42))
except ConnectionError as e:
    print(f"failed: {e}")

Real tools fail for network, rate limits, and dependency outages.
Bare calls force the agent loop to crash or invent a result.
Reliability starts at the tool boundary, not in the prompt.

Related: Why Agents Fail Differently Than Traditional Services

2. Retry a Fixed Number of Times

Add a simple retry loop for clearly transient errors.

from typing import Callable, TypeVar
 
T = TypeVar("T")
 
 
def retry(fn: Callable[[], T], attempts: int = 3) -> T:
    last: Exception | None = None
    for i in range(1, attempts + 1):
        try:
            return fn()
        except ConnectionError as e:
            last = e
            print(f"attempt {i} failed: {e}")
    raise RuntimeError(f"exhausted {attempts} attempts") from last
 
 
print(retry(lambda: flaky_fetch_order(42)))

Cap attempts in the host; do not rely on the model alone.
Only catch errors you believe are transient.
Always keep a final raise or structured failure for the loop.

3. Add Exponential Backoff and Jitter

Immediate retries amplify outages. Sleep between attempts.

import time
import random
from typing import Callable, TypeVar
 
T = TypeVar("T")
 
 
def retry_with_backoff(
    fn: Callable[[], T],
    attempts: int = 4,
    base_s: float = 0.05,
    max_s: float = 1.0,
) -> T:
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except ConnectionError as e:
            last = e
            if i == attempts - 1:
                break
            sleep = min(max_s, base_s * (2**i))
            sleep = sleep * (0.5 + random.random())  # full jitter-ish
            print(f"backoff {sleep:.3f}s after: {e}")
            time.sleep(sleep)
    raise RuntimeError("retry budget exhausted") from last
 
 
print(retry_with_backoff(lambda: flaky_fetch_order(7)))

Exponential growth reduces thundering herds.
Jitter spreads clients so they do not align.
Production values are larger; keep demos fast.

4. Classify Retryable vs Fatal Errors

Not every exception deserves another attempt.

class FatalToolError(Exception):
    """Do not retry - fix inputs or config instead."""
 
 
def fetch_order(order_id: int | None) -> dict:
    if order_id is None:
        raise FatalToolError("order_id is required")
    return flaky_fetch_order(order_id)
 
 
def retry_classified(fn: Callable[[], T], attempts: int = 3) -> T:
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except FatalToolError:
            raise
        except ConnectionError as e:
            last = e
            print(f"retryable failure {i + 1}: {e}")
    raise RuntimeError("retry budget exhausted") from last
 
 
print(retry_classified(lambda: fetch_order(1)))
try:
    retry_classified(lambda: fetch_order(None))
except FatalToolError as e:
    print(f"fatal: {e}")

Auth misconfig, validation errors, and not-found are usually fatal mid-run.
Rate limits and timeouts are often retryable with limits.
Wrong classification creates either storms or dead ends.

5. Return a Structured Result Instead of Raising

Agent loops work better when tools return observations the model can read.

from dataclasses import dataclass
from typing import Any
 
 
@dataclass
class ToolResult:
    ok: bool
    data: Any = None
    error_type: str | None = None
    retryable: bool = False
    message: str | None = None
 
 
def call_with_retries(order_id: int) -> ToolResult:
    try:
        data = retry_with_backoff(lambda: fetch_order(order_id), attempts=3)
        return ToolResult(ok=True, data=data)
    except FatalToolError as e:
        return ToolResult(
            ok=False,
            error_type="invalid_args",
            retryable=False,
            message=str(e),
        )
    except Exception as e:
        return ToolResult(
            ok=False,
            error_type="upstream",
            retryable=False,
            message=f"gave up: {e}",
        )
 
 
print(call_with_retries(99))

ok: false is still a valid tool observation.
The model can adapt; the host already spent the retry budget.
Log the same structure for traces and SLOs later.

Related: Handling Tool Errors Gracefully Inside an Agent Loop

Intermediate Examples

Per-call retries are not enough when the model re-invokes the same tool every turn.

from dataclasses import dataclass, field
 
 
@dataclass
class RetryBudget:
    max_errors: int = 5
    errors: int = 0
 
    def allow(self) -> bool:
        return self.errors < self.max_errors
 
    def record(self) -> None:
        self.errors += 1
 
 
def call_with_shared_budget(order_id: int, budget: RetryBudget) -> ToolResult:
    if not budget.allow():
        return ToolResult(
            ok=False,
            error_type="retry_budget_exhausted",
            retryable=False,
            message="host refused further upstream attempts",
        )
    try:
        data = flaky_fetch_order(order_id)
        return ToolResult(ok=True, data=data)
    except ConnectionError as e:
        budget.record()
        return ToolResult(
            ok=False,
            error_type="upstream",
            retryable=budget.allow(),
            message=str(e),
        )
 
 
budget = RetryBudget(max_errors=2)
for _ in range(4):
    print(call_with_shared_budget(1, budget), "errors=", budget.errors)

Run-level budgets stop death-by-many-turns.
Surface retry_budget_exhausted so the agent can stop or escalate.
Combine with per-signature limits in production.

7. Soft Timeout Around a Slow Tool

Hang forever is worse than a clean timeout.

import concurrent.futures
 
 
def slow_tool() -> str:
    time.sleep(2.0)
    return "done"
 
 
def call_with_timeout(fn: Callable[[], T], timeout_s: float) -> ToolResult:
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        fut = pool.submit(fn)
        try:
            return ToolResult(ok=True, data=fut.result(timeout=timeout_s))
        except concurrent.futures.TimeoutError:
            return ToolResult(
                ok=False,
                error_type="timeout",
                retryable=True,
                message=f"tool exceeded {timeout_s}s",
            )
 
 
print(call_with_timeout(slow_tool, timeout_s=0.2))

Prefer native client deadlines when the SDK supports them.
Thread timeouts do not kill all work; treat this as a teaching sketch.
See Timeout Strategies for Long-Running Agent Steps for production patterns.

8. Trip a Tiny Circuit After Repeated Failures

When the dependency is down, fail fast instead of retrying every turn.

@dataclass
class Breaker:
    failure_threshold: int = 3
    failures: int = 0
    open: bool = False
 
    def before_call(self) -> ToolResult | None:
        if self.open:
            return ToolResult(
                ok=False,
                error_type="circuit_open",
                retryable=False,
                message="circuit open; skip upstream",
            )
        return None
 
    def on_success(self) -> None:
        self.failures = 0
        self.open = False
 
    def on_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.open = True
 
 
breaker = Breaker()
 
 
def protected_fetch(order_id: int) -> ToolResult:
    blocked = breaker.before_call()
    if blocked:
        return blocked
    try:
        data = flaky_fetch_order(order_id)
        breaker.on_success()
        return ToolResult(ok=True, data=data)
    except ConnectionError as e:
        breaker.on_failure()
        return ToolResult(ok=False, error_type="upstream", retryable=not breaker.open, message=str(e))
 
 
for i in range(6):
    # Force failures for the demo
    random.seed(0)
    print(i, protected_fetch(1), "open=", breaker.open)

Open circuits protect both the dependency and your token budget.
Full half-open recovery is covered in Circuit Breakers for Agent Tool Calls.
Pair with fallbacks so users still get a degraded path.

What You Learned

Retry only classified transient failures, with caps and backoff.
Prefer structured tool results over uncaught exceptions in the agent loop.
Share budgets across turns; one tool call is not the whole story.
Timeouts and circuit flags keep flaky dependencies from owning the run.

Next Steps

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

You will wrap a flaky tool with retries and backoff, classify failures, budget attempts, short-circuit on non-retryable errors, and sketch a tiny circuit-open flag.

Prerequisites

Python 3.11+ recommended
Comfort with calling a function that can raise or time out
No paid APIs required for these sketches

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Call a Flaky Tool Without Protection

Start from the failure you will fix: a tool that sometimes raises.

import random
 
 
def flaky_fetch_order(order_id: int) -> dict:
    if random.random() < 0.7:
        raise ConnectionError("upstream_unavailable")
    return {"order_id": order_id, "status": "shipped"}
 
 
try:
    print(flaky_fetch_order(42))
except ConnectionError as e:
    print(f"failed: {e}")

Real tools fail for network, rate limits, and dependency outages.
Bare calls force the agent loop to crash or invent a result.
Reliability starts at the tool boundary, not in the prompt.

Related: Why Agents Fail Differently Than Traditional Services

2. Retry a Fixed Number of Times

Add a simple retry loop for clearly transient errors.

from typing import Callable, TypeVar
 
T = TypeVar("T")
 
 
def retry(fn: Callable[[], T], attempts: int = 3) -> T:
    last: Exception | None = None
    for i in range(1, attempts + 1):
        try:
            return fn()
        except ConnectionError as e:
            last = e
            print(f"attempt {i} failed: {e}")
    raise RuntimeError(f"exhausted {attempts} attempts") from last
 
 
print(retry(lambda: flaky_fetch_order(42)))

Cap attempts in the host; do not rely on the model alone.
Only catch errors you believe are transient.
Always keep a final raise or structured failure for the loop.

3. Add Exponential Backoff and Jitter

Immediate retries amplify outages. Sleep between attempts.

import time
import random
from typing import Callable, TypeVar
 
T = TypeVar("T")
 
 
def retry_with_backoff(
    fn: Callable[[], T],
    attempts: int = 4,
    base_s: float = 0.05,
    max_s: float = 1.0,
) -> T:
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except ConnectionError as e:
            last = e
            if i == attempts - 1:
                break
            sleep = min(max_s, base_s * (2**i))
            sleep = sleep * (0.5 + random.random())  # full jitter-ish
            print(f"backoff {sleep:.3f}s after: {e}")
            time.sleep(sleep)
    raise RuntimeError("retry budget exhausted") from last
 
 
print(retry_with_backoff(lambda: flaky_fetch_order(7)))

Exponential growth reduces thundering herds.
Jitter spreads clients so they do not align.
Production values are larger; keep demos fast.

4. Classify Retryable vs Fatal Errors

Not every exception deserves another attempt.

class FatalToolError(Exception):
    """Do not retry - fix inputs or config instead."""
 
 
def fetch_order(order_id: int | None) -> dict:
    if order_id is None:
        raise FatalToolError("order_id is required")
    return flaky_fetch_order(order_id)
 
 
def retry_classified(fn: Callable[[], T], attempts: int = 3) -> T:
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except FatalToolError:
            raise
        except ConnectionError as e:
            last = e
            print(f"retryable failure {i + 1}: {e}")
    raise RuntimeError("retry budget exhausted") from last
 
 
print(retry_classified(lambda: fetch_order(1)))
try:
    retry_classified(lambda: fetch_order(None))
except FatalToolError as e:
    print(f"fatal: {e}")

Auth misconfig, validation errors, and not-found are usually fatal mid-run.
Rate limits and timeouts are often retryable with limits.
Wrong classification creates either storms or dead ends.

5. Return a Structured Result Instead of Raising

Agent loops work better when tools return observations the model can read.

from dataclasses import dataclass
from typing import Any
 
 
@dataclass
class ToolResult:
    ok: bool
    data: Any = None
    error_type: str | None = None
    retryable: bool = False
    message: str | None = None
 
 
def call_with_retries(order_id: int) -> ToolResult:
    try:
        data = retry_with_backoff(lambda: fetch_order(order_id), attempts=3)
        return ToolResult(ok=True, data=data)
    except FatalToolError as e:
        return ToolResult(
            ok=False,
            error_type="invalid_args",
            retryable=False,
            message=str(e),
        )
    except Exception as e:
        return ToolResult(
            ok=False,
            error_type="upstream",
            retryable=False,
            message=f"gave up: {e}",
        )
 
 
print(call_with_retries(99))

ok: false is still a valid tool observation.
The model can adapt; the host already spent the retry budget.
Log the same structure for traces and SLOs later.

Related: Handling Tool Errors Gracefully Inside an Agent Loop

Intermediate Examples

Per-call retries are not enough when the model re-invokes the same tool every turn.

from dataclasses import dataclass, field
 
 
@dataclass
class RetryBudget:
    max_errors: int = 5
    errors: int = 0
 
    def allow(self) -> bool:
        return self.errors < self.max_errors
 
    def record(self) -> None:
        self.errors += 1
 
 
def call_with_shared_budget(order_id: int, budget: RetryBudget) -> ToolResult:
    if not budget.allow():
        return ToolResult(
            ok=False,
            error_type="retry_budget_exhausted",
            retryable=False,
            message="host refused further upstream attempts",
        )
    try:
        data = flaky_fetch_order(order_id)
        return ToolResult(ok=True, data=data)
    except ConnectionError as e:
        budget.record()
        return ToolResult(
            ok=False,
            error_type="upstream",
            retryable=budget.allow(),
            message=str(e),
        )
 
 
budget = RetryBudget(max_errors=2)
for _ in range(4):
    print(call_with_shared_budget(1, budget), "errors=", budget.errors)

Run-level budgets stop death-by-many-turns.
Surface retry_budget_exhausted so the agent can stop or escalate.
Combine with per-signature limits in production.

7. Soft Timeout Around a Slow Tool

Hang forever is worse than a clean timeout.

import concurrent.futures
 
 
def slow_tool() -> str:
    time.sleep(2.0)
    return "done"
 
 
def call_with_timeout(fn: Callable[[], T], timeout_s: float) -> ToolResult:
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        fut = pool.submit(fn)
        try:
            return ToolResult(ok=True, data=fut.result(timeout=timeout_s))
        except concurrent.futures.TimeoutError:
            return ToolResult(
                ok=False,
                error_type="timeout",
                retryable=True,
                message=f"tool exceeded {timeout_s}s",
            )
 
 
print(call_with_timeout(slow_tool, timeout_s=0.2))

Prefer native client deadlines when the SDK supports them.
Thread timeouts do not kill all work; treat this as a teaching sketch.
See Timeout Strategies for Long-Running Agent Steps for production patterns.

8. Trip a Tiny Circuit After Repeated Failures

When the dependency is down, fail fast instead of retrying every turn.

@dataclass
class Breaker:
    failure_threshold: int = 3
    failures: int = 0
    open: bool = False
 
    def before_call(self) -> ToolResult | None:
        if self.open:
            return ToolResult(
                ok=False,
                error_type="circuit_open",
                retryable=False,
                message="circuit open; skip upstream",
            )
        return None
 
    def on_success(self) -> None:
        self.failures = 0
        self.open = False
 
    def on_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.open = True
 
 
breaker = Breaker()
 
 
def protected_fetch(order_id: int) -> ToolResult:
    blocked = breaker.before_call()
    if blocked:
        return blocked
    try:
        data = flaky_fetch_order(order_id)
        breaker.on_success()
        return ToolResult(ok=True, data=data)
    except ConnectionError as e:
        breaker.on_failure()
        return ToolResult(ok=False, error_type="upstream", retryable=not breaker.open, message=str(e))
 
 
for i in range(6):
    # Force failures for the demo
    random.seed(0)
    print(i, protected_fetch(1), "open=", breaker.open)

Open circuits protect both the dependency and your token budget.
Full half-open recovery is covered in Circuit Breakers for Agent Tool Calls.
Pair with fallbacks so users still get a degraded path.

What You Learned

Retry only classified transient failures, with caps and backoff.
Prefer structured tool results over uncaught exceptions in the agent loop.
Share budgets across turns; one tool call is not the whole story.
Timeouts and circuit flags keep flaky dependencies from owning the run.

Next Steps

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Reliability Engineering Basics

Search Documentation

Reliability Engineering Basics