Resilience & client controls

Configure retries and deadlines, respond to cancellation, and control outbound load with rate limiting, caching and model routing. Runtime retry and timeout examples need a graph server. The process-local controls can run in an ordinary event loop.

What the server enforces and what the client does

Controls in python/atfgraph/controls.py apply within one process, above the SDK's bounded HTTP connections. They do not coordinate limits across workers. The graph server remains responsible for its admission rules.

Retries: RetryableError and max_attempts

Raise RetryableError, or another HandlerError with retryable=True, when another attempt may succeed. The server retries only while the capability's max_attempts allows it. Ordinary exceptions fail the run without retry. Inspect current_context().attempt for the active attempt number. The resilience example uses:

from pydantic import BaseModel

from atfgraph import Agent, RetryableError, current_context
from atfgraph.model import ModelError

app = Agent(node_id="disruption", version="1.0.0", capacity=4)


class RebookingNote(BaseModel):
    summary: str
    rebook_within_hours: int


@app.skill(max_attempts=3)
async def rebooking_note(flight: str, model: str = "fixture-flaky") -> dict:
    attempt = current_context().attempt
    try:
        note = await app.ai(
            system="You are a disruption desk. Answer with the requested fields only.",
            user=f"Flight {flight} was cancelled. Summarise the rebooking action and the hours we have to act.",
            schema=RebookingNote,
            model=model,
        )
    except ModelError as exc:
        raise RetryableError(f"provider failed on attempt {attempt}: {exc}") from exc
    return {"flight": flight, "attempt": attempt, "summary": note.summary, "rebook_within_hours": note.rebook_within_hours}

This handler needs a model provider. Its fixture service offers fixture-flaky, which returns 503 twice before succeeding, and fixture-slow, which stalls. The service is in examples/python/features/resilience/fixture_model.py.

python -m examples.python.features.resilience.fixture_model --port 18899
export ATFGRAPH_MODEL_URL=http://127.0.0.1:18899 ATFGRAPH_MODEL_NAME=fixture-fast
python -m atfgraph.worker --module examples.python.features.resilience.agent:app
python -m examples.python.features.resilience.call

Real-daemon feature tests check success on attempt 3, a three-attempt budget, two retryable requeued events and exactly three provider calls. With more configured failures than the budget permits, the run ends after attempt 3; no fourth attempt is made. The budget applies to the capability execution. Without model URL and name settings, the call fails with ModelNotConfigured.

Capability budgets: timeout_seconds

A capability's timeout_seconds, default 300, limits each active handler attempt. The worker uses asyncio.wait_for and reports non-retryable handler_timeout when it expires. If the worker is absent, the server's deadline can mark the run timed_out, producing TraversalTimedOut for the caller. The same example declares:

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="disruption", version="1.0.0")


class RebookingNote(BaseModel):
    summary: str
    rebook_within_hours: int


@app.skill(timeout_seconds=2)
async def stalled_note(flight: str) -> dict:
    note = await app.ai(user=f"Flight {flight} was cancelled.", schema=RebookingNote, model="fixture-slow")
    return {"flight": flight, "summary": note.summary}

A test uses a 30-second stalled provider and checks termination within 20 seconds as either worker handler_timeout or server timed_out. A timeout cannot interrupt a Python thread: synchronous work may continue after the worker stops waiting, but its result is discarded.

Guardrails: refuse bad input and bad output

Guardrails are named checks around handler execution. Input checks inspect the admitted payload before the handler starts; output checks inspect its encoded result before completion. A refusal fails the step with non-retryable guardrail_violation, identifying the rule and reason. The failure appears on the path and Console Runs page rather than silently changing the data.

import json
import re

from atfgraph import Agent, allowed_values, deny_patterns, max_chars, require_keys

app = Agent(node_id="support")


@app.guardrail("no_card_numbers", phase="output", description="refuse card-like digit runs")
def no_card_numbers(value, ctx):
    # Return None to pass, or a message (or list of messages) to refuse.
    if re.search(r"\b\d{13,19}\b", json.dumps(value)):
        return "output contains a card-number-like sequence"


@app.traversal(
    guardrails=[
        "no_card_numbers",                                  # registered by name above
        require_keys("has_answer", ["answer", "sources"]),  # output must carry these keys
        max_chars("short_reply", 4000, fields=["answer"]),  # bound one field's length
        allowed_values("lang", "lang", ["en", "fr"]),       # an input guardrail
    ]
)
async def answer(question: str, lang: str = "en") -> dict:
    ...

Built-ins include deny_patterns, max_chars, require_keys, json_schema and allowed_values. Choose phase input, output or both. Pattern failures name the pattern rather than echoing matched text. A custom synchronous or asynchronous check can return None, a boolean, a reason string or a list of reasons; register it with @app.guardrail, Guardrail(name, check, phase) or app.add_guardrail. Local calls apply the same checks, so a unit test can expect GuardrailViolation without a server:

import pytest
from atfgraph import GuardrailViolation

async def test_card_numbers_are_refused():
    with pytest.raises(GuardrailViolation) as refused:
        await app.call_local("answer", question="pay with 4111111111111111")
    assert refused.value.rule == "no_card_numbers" and refused.value.phase == "output"

Guardrails execute in the worker where handler output exists. Server-side access policies decide whether a caller may submit the requested target and input. Worker guardrails decide which data the handler accepts or returns. A violation is a failed, non-retryable attempt.

Cooperative cancellation

Handle.cancel() or atfg cancel marks the server run for cancellation. The worker observes it during lease renewal and cancels the asynchronous handler. Check app.cancel_requested() between slices of long-running loops and report useful progress before cancellation. The resilience example does this:

import asyncio

from atfgraph import Agent, ConflictError

app = Agent(node_id="disruption", version="1.0.0")


@app.traversal(timeout_seconds=600)
async def watch(flight: str, seconds: float = 30.0, slice_seconds: float = 0.1) -> dict:
    ticks = max(1, int(seconds / slice_seconds))
    per_note = max(1, int(1.0 / slice_seconds))
    done = 0
    while done < ticks and not app.cancel_requested():
        await asyncio.sleep(slice_seconds)
        done += 1
        if done % per_note == 0:
            try:
                await app.progress({"flight": flight, "watched_seconds": round(done * slice_seconds, 1)})
            except ConflictError:
                break
    return {"flight": flight, "watched_seconds": round(done * slice_seconds, 1), "cancelled": app.cancel_requested()}

A progress request after cancellation may raise ConflictError, so the example catches it. Its test cancels after 1.2 seconds, expects TraversalCancelled, and checks that the first one-second progress update remains recorded with cancel_requested: True. No model provider is needed.

RateLimiter

RateLimiter(default=None, policies={...}) limits calls by string key within one event loop. A RateLimitPolicy combines admissions per per seconds, token-bucket burst, concurrent-call capacity and max_pending waiters. Specify at least a rate or concurrency limit. Keys ending in : match prefixes: node: applies to node:slow unless it has a more specific policy. Without a matching or default policy, the key is unlimited. The serverless example and control tests use:

import asyncio

from atfgraph import Agent, RateLimiter
from atfgraph.controls import RateLimitExceeded, RateLimitPolicy

app = Agent(node_id="quoter", version="1.0.0", capacity=2)
limiter = RateLimiter(policies={"node:quoter": RateLimitPolicy(rate=20, per=1.0, concurrency=2, max_pending=50)})


@app.traversal()
async def quote(legs: list[dict]) -> dict:
    async def one(leg: dict) -> dict:
        try:
            async with limiter.limit("node:quoter", timeout=5.0):
                return await app.traverse("quoter.price", **leg)
        except RateLimitExceeded as exc:
            return {"error": str(exc), "retry_after": exc.retry_after}

    priced = await asyncio.gather(*(one(leg) for leg in legs))
    return {"legs": priced, "admitted": limiter.stats("node:quoter").admitted}

Waiters are admitted FIFO for each key. limit(key, timeout=...) raises RateLimitExceeded if the waiting queue is full or the wait expires. Cancelled waiters leave without retaining a permit. run(key, func, *args) holds a permit until the call ends; acquire and release provide manual control. stats(key) reports admissions, refusals, timeouts, cancelled waits and peak concurrency and pending counts.

ResultCache

ResultCache(ttl=300.0, max_size=1024) is an expiring least-recently-used cache within one event loop. Concurrent get_or_compute(key, factory) calls share one calculation for that key. Stored and returned values are deep copies. Factory errors reach all waiters without being cached. If the calculating caller is cancelled, a waiter can take over. The serverless example uses:

import asyncio

from atfgraph import Agent, ResultCache

app = Agent(node_id="quoter", version="1.0.0", capacity=2)
cache = ResultCache(ttl=60.0, max_size=256)


@app.traversal()
async def quote(legs: list[dict]) -> dict:
    async def one(leg: dict) -> dict:
        key = ("price", leg.get("origin"), leg.get("destination"), leg.get("nights", 1))

        async def compute() -> dict:
            return await app.traverse("quoter.price", **leg)

        return await cache.get_or_compute(key, compute)

    priced = await asyncio.gather(*(one(leg) for leg in legs))
    return {"legs": priced, "cache": {"hits": cache.stats.hits, "misses": cache.stats.misses}}

The controls tests submit three legs, including a duplicate, and check that only two child pricing steps execute. Repeating a cached leg records a cache hit without creating another child step. Start optional background expiry with start/stop or async with cache; it belongs to its starting loop. len() and keys() also sweep expired entries.

retry_rate_limited

retry_rate_limited(func, max_attempts=3, base_delay=0.2, max_delay=5.0, deadline=None) retries a zero-argument async function only for recognized rate-limit errors: API status 429, RateLimitExceeded, or ModelError text containing HTTP 429. It uses retry_after when supplied; otherwise it backs off exponentially up to max_delay. It stops before sleeping beyond the monotonic deadline. Other errors propagate immediately. The tests demonstrate:

import asyncio

from atfgraph import Client, retry_rate_limited


async def main() -> None:
    async with Client("http://127.0.0.1:18880") as client:
        async def once() -> dict:
            return await client.traverse("quoter.quote", {"legs": [{"origin": "BNE", "destination": "SYD"}]}, timeout=30)

        print(await retry_rate_limited(once, max_attempts=3))


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

ModelRouter

ModelRouter sends complete and stream calls only to providers you register. It neither reads environment settings nor creates providers. Resolution first checks an explicit <route>:<model>, then matching model names or prefixes, then the default route and its default model. An unresolved request raises ModelNotConfigured. The controls tests configure:

from atfgraph import Agent, ModelRouter, RateLimiter
from atfgraph.controls import RateLimitPolicy
from atfgraph.model import HTTPChatModel

limiter = RateLimiter(policies={"model:fast": RateLimitPolicy(concurrency=2)})
router = ModelRouter(limiter=limiter, admission_timeout=10.0)
router.register("fast", HTTPChatModel("http://127.0.0.1:18899", "fixture-fast"), default_model="fixture-fast", prefixes=["fixture-"])

app = Agent(node_id="routed", version="1.0.0", model=router)


@app.skill()
async def answer(question: str) -> dict:
    text = await app.ai(question)
    bigger = await app.ai(question, model="fast:fixture-fast")
    return {"default": str(text), "explicit": str(bigger), "calls": dict(router.calls)}

With a limiter, each route uses its limit_key, default model:<route>. Streams hold the permit until completed or closed. resolve(model, needs=["vision"]) excludes providers lacking that capability. The shown provider URL must point to the resilience fixture or another reachable OpenAI-compatible endpoint.

Next

• Review worker capacity in Calling other nodes.
• Configure providers in Model calls.
• Combine controls with a Serverless host.