Calling other nodes & fan-out

Call other capabilities, run child work concurrently, record progress, handle cancellation and configure retries. These examples use the SDK and a graph server. nested_concurrency needs no additional provider; the resilience example also needs its fixture model service, as noted below.

traverse, call and submit

Choose one of these Agent methods to submit work (python/atfgraph/agent.py):

A target without a dot uses the current node's label: square resolves to fanout.square. To call another node, include its label, such as await app.traverse("pricing.quote", fare="Y"). The server checks access policies before admitting the call; see Identity and access. Select a version with submit(..., version="1.0.0").

Inside a handler, submissions become children of its current step. The SDK sends the parent step ID, lease token and operation key from ExecutionContext. The server records the relationship, inherits the root and session, and wakes the parent's worker to claim child work promptly. Outside a handler, the same methods use app.client to create ordinary root traversers.

The module-level traverse function uses the Agent in the current ExecutionContext and has the same behavior.

How a cross-node call travels

Every cross-node call goes through the graph server. Workers need no inbound ports or direct connections to one another. A call proceeds as follows:

1. A handler on node A calls await app.traverse("pricing.quote", fare="Y"). The label selects the registered pricing node.
2. The SDK submits a child with the parent step ID, lease token and operation key. The server checks access and rate policies, records the relationship and queues the task for pricing.
3. An outbound worker for node B claims the task under a lease, runs quote and returns its result to the server.
4. Node A's handler receives the result. Parent and child share the root and session, keeping state and usage associated with the same run.

The server authenticates requests, enforces policy, leases tasks and records calls. It retries eligible RetryableError failures and requeues expired leases. Nodes coordinate through that common authority.

Both skills and traversals are addressed as label.capability. Skills conventionally perform one operation; traversals can call more capabilities or models, creating branches in the path. The caller uses the same method for a local-node or remote-node capability.

Fan out with asyncio.gather

This example from examples/python/nested_concurrency/agent.py creates one child step per app.traverse call. asyncio.gather runs each batch concurrently and returns results in input order.

from __future__ import annotations

import asyncio

from atfgraph import Agent

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


@app.skill()
async def square(n: int) -> dict:
    await asyncio.sleep(0.05)
    return {"n": n, "square": n * n}


@app.traversal()
async def fan_out(count: int = 6, batch: int = 3) -> dict:
    results = []
    for start in range(0, count, batch):
        chunk = list(range(start, min(start + batch, count)))
        squares = await asyncio.gather(*[app.traverse("fanout.square", n=n) for n in chunk])
        results.extend(s["square"] for s in squares)
        await app.progress({"completed": len(results), "of": count})
    return {"squares": results, "sum": sum(results)}

The example tests use count=6, batch=3, check squares [0, 1, 4, 9, 16, 25] and sum 55, and expect two progress events. handle.path() includes every square as a child of fan_out.

A DAG across nodes

Write dependencies with ordinary Python control flow. Here, three supplier searches run concurrently; ranking waits for their results, and holding a fare waits for ranking. The server records the execution graph as these calls occur.

                 ┌── fares.search ──┐
trip.plan ───────┼── hotels.search ─┼──► trip.rank ──► fares.hold
                 └── weather.today ─┘

Each supplier node runs a worker. The caller uses its label rather than a network address:

# fares_node.py
from atfgraph import Agent

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


@app.skill()
async def search(origin: str, dest: str, date: str) -> dict:
    return {"offers": [{"offer_id": "F1", "total": 412.0}, {"offer_id": "F2", "total": 388.0}]}


@app.skill()
async def hold(offer_id: str) -> dict:
    return {"offer_id": offer_id, "pnr": "ABC123", "expires_in_minutes": 30}


if __name__ == "__main__":
    app.run()

# hotels_node.py
from atfgraph import Agent

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


@app.skill()
async def search(city: str, checkin: str, nights: int) -> dict:
    return {"properties": [{"property_token": "H9", "name": "Quay West", "total_rate": 640.0}]}


if __name__ == "__main__":
    app.run()

# weather_node.py
from atfgraph import Agent

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


@app.skill()
async def today(city: str) -> dict:
    return {"city": city, "summary": "22°C, light rain in the afternoon"}


if __name__ == "__main__":
    app.run()

The coordinating node combines them:

# trip_node.py
import asyncio

from atfgraph import Agent

app = Agent(node_id="trip", version="1.0.0", capacity=2)


@app.skill()
async def rank(fares: dict, hotels: dict, weather: dict) -> dict:
    best = min(fares["offers"], key=lambda offer: offer["total"])
    return {"fare": best, "hotel": hotels["properties"][0], "note": weather["summary"]}


@app.traversal()
async def plan(origin: str, dest: str, date: str, nights: int = 3) -> dict:
    # Level 1: three independent calls to three nodes, one child step each.
    fares, hotels, weather = await asyncio.gather(
        app.traverse("fares.search", origin=origin, dest=dest, date=date),
        app.traverse("hotels.search", city=dest, checkin=date, nights=nights),
        app.traverse("weather.today", city=dest),
    )
    await app.progress({"stage": "searched", "offers": len(fares["offers"])})

    # Level 2: depends on all three.
    choice = await app.traverse("trip.rank", fares=fares, hotels=hotels, weather=weather)

    # Level 3: depends on the ranking.
    held = await app.traverse("fares.hold", offer_id=choice["fare"]["offer_id"])
    return {"choice": choice, "hold": held, "requires_review": True}


if __name__ == "__main__":
    app.run()

Start each worker against the same server, then submit the root run:

python -m atfgraph.worker --module fares_node:app
python -m atfgraph.worker --module hotels_node:app
python -m atfgraph.worker --module weather_node:app
python -m atfgraph.worker --module trip_node:app --capacity 2
atfg traverse trip.plan --in '{"origin": "BNE", "dest": "SYD", "date": "2026-10-03"}'

The recorded execution has these properties:

• Each call inside plan carries its parent step and lease. The path shows the calls under plan, including concurrent searches; inspect them with atfg path <id>.
• Workers serving fares, hotels and weather execute the searches. The orchestrator does not need their addresses. Access policies determine whether the caller may invoke them.
• While waiting for children, plan releases its worker permit. This lets trip.rank run on the same worker even at --capacity 1, subject to the waiting bound below.
• A retryable failure in fares.hold retries that step within its max_attempts budget. Cancelling the root also cancels unfinished children.

The shapes a path can take

You do not need to declare a fixed DAG before execution. Each traversal decides which calls to make; the server records the graph they produce. Ordinary Python expresses these common patterns:

Step records, attempts and leases survive worker failure. Completed children remain recorded while unfinished work can be recovered. Permit release helps a small worker execute nested calls without its waiting parents occupying all capacity; the next section explains its configured bound.

When the DAG is not known in advance

In examples/python/research_path, a model proposes tasks and dependencies. The example rejects dependency cycles and groups tasks into topological levels using parallelizable_groups. Its scheduler is:

for level in plan["parallelizable_groups"]:
    batch = await asyncio.gather(*(
        app.call("research_path.execute_research_task",
                 task=by_id[key],
                 dependencies=[results[dep] for dep in by_id[key]["dependencies"]])
        for key in level))
    results.update({item["task_id"]: item for item in batch})

Tasks within a level execute concurrently, each receiving its dependencies' outputs. Use a discovered target instead of the fixed target to route tasks to different nodes.

Capacity and deadlock avoidance

A worker has capacity concurrent handler permits, default 8, configurable with --capacity. A parent waiting for a child on a one-permit worker would block that child if it kept the permit. waiting_on_child in worker.py handles this case:

1. On the first active child wait, the worker calls worker_suspend and releases the parent's permit. Several concurrent child waits in that handler cause only one release.
2. When the last child wait ends, it reacquires a permit and calls worker_resume, retrying 429 capacity_exhausted responses.
3. Released parents are limited by max_waiting, default capacity * 4, configurable with --max-waiting. Beyond that bound, parents keep their permits while waiting. test_bounded_waiting_keeps_permit_beyond_max_waiting covers this limit.

The example completes at --capacity 1, as checked by tests/python_test_examples.py and the real-daemon test_python_worker_nested_call_at_capacity_one:

.venv/bin/python -m atfgraph.worker --module examples.python.nested_concurrency.agent:app --capacity 1
.venv/bin/python -m examples.python.nested_concurrency.call

Two other rules affect throughput and recovery. Root turns in the same session run sequentially, while their child calls may run concurrently (test_root_session_turns_are_serialized_children_bypass). If a lease expires, the server requeues work and rejects results from the old lease (test_lease_expiry_requeues_and_stale_result_is_discarded).

Progress events

Inside a handler, await app.progress(data) writes a durable progress event. await app.note("text", tags=["phase"]) writes a progress payload with note and tags. Both require the active execution context and task lease. A caller can read them through its Handle:

from __future__ import annotations

import asyncio
import os

from atfgraph import Client


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        handle = await client.submit("fanout.fan_out", {"count": 6, "batch": 3})
        async for event in handle.events():
            print(f"event {event.sequence:>3} {event.type:<10} {event.data}")
        print("result", await handle.result(timeout=120))
        path = await handle.path()
        print(f"path: {len(path.items)} steps under root {path.root_id}")


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

This comes from examples/python/nested_concurrency/call.py, without its cancellation section. handle.events() streams SSE events until the terminal event; handle.event_log() returns a snapshot list. Progress payloads appear in event.data["data"]. Model usage also appears as progress with a usage key; see Observability.

Cancellation

await handle.cancel() and atfg cancel cancel a traverser and its children (test_cancel_propagates_to_children). The worker notices on lease renewal, marks the context as cancelled and cancels the handler task. The nested-concurrency example checks cancellation in two ways:

from __future__ import annotations

import asyncio

from atfgraph import Agent

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


@app.skill(timeout_seconds=600)
async def slow_step(seconds: float) -> dict:
    """Sleeps in small slices so a server cancel interrupts it promptly."""
    waited = 0.0
    while waited < seconds:
        if app.cancel_requested():
            break
        await asyncio.sleep(0.1)
        waited += 0.1
    return {"waited": round(waited, 1)}


@app.traversal(timeout_seconds=600)
async def long_parent(seconds: float = 30) -> dict:
    child = await app.traverse("fanout.slow_step", seconds=seconds)
    return {"child": child}

Use app.cancel_requested() in cooperative loops. An asynchronous handler may also receive asyncio.CancelledError at an await. The caller's handle.result() raises TraversalCancelled. A progress write racing with cancellation can raise ConflictError; the resilience example catches it and stops. Synchronous threads cannot be interrupted, so they finish and their result is discarded.

Retries

An ordinary exception ends the attempt with its class as the error code and retryable: false, failing the run. To request another attempt, raise RetryableError or HandlerError(..., retryable=True) and configure max_attempts on the capability:

from atfgraph import Agent, RetryableError, current_context

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


@app.skill(max_attempts=3)
async def flaky_step(succeed_on_attempt: int = 2) -> dict:
    """Fails with a retryable error until the server has scheduled ``succeed_on_attempt`` attempts."""
    ctx = current_context()
    if ctx.attempt < succeed_on_attempt:
        raise RetryableError(f"attempt {ctx.attempt} failed on purpose")
    return {"attempt": ctx.attempt}

The example and feature tests check these retry behaviors:

• ctx.attempt starts at 1 and increases on each scheduled attempt. PathStep.attempts preserves each attempt and its error's retryable flag.
• Each retry writes a requeued event with its error. client.get(id).raw exposes attempt and max_attempts.
• Exhausting the budget fails the run with the last error code, available as TraversalFailed.code.
• Parent retries reuse stable child operation keys derived from target, input and call count. The server can therefore return an existing child rather than create it again (test_operation_keys_are_stable_across_parent_retries).

The resilience example catches a provider ModelError and raises RetryableError for a transient failure. A bad request expected to fail again remains an ordinary exception. Start its fixture provider with python -m examples.python.features.resilience.fixture_model --port 18899 and point ATFGRAPH_MODEL_URL and ATFGRAPH_MODEL_NAME to it. Without a configured provider, app.ai raises ModelNotConfigured.

Timeouts

Distinguish the handler, server and client deadlines:

The stalled-call test submits disruption.stalled_note with a two-second capability budget to a provider that stalls for 30 seconds. It accepts failed/handler_timeout or timed_out/timed_out, completing within 20 seconds. A nested call's WaitTimeout fails the parent like another exception unless your handler catches it. A client wait expiring does not itself cancel the child.

Next

• Share state using Sessions and memory.
• Configure resilience controls.
• Find other capabilities through Multi-node systems.