Multi-node systems

Connect agent nodes through recorded child calls, discover capabilities, select versions and serve several nodes in one process. Most sections need a running graph server. OfflineMesh is a separate in-memory option for handler tests.

A cross-node edge

Inside a handler, app.traverse("label.capability", ...) starts a child call and waits for its output. The server records that child on the parent's path. These examples come from the Databricks two-node intake and pricing modules:

from pydantic import BaseModel

from atfgraph import Agent

intake = Agent(node_id="intake", version="1.0.0", description="Extract a request, then walk the edge to pricing.quote")
pricing = Agent(node_id="pricing", version="1.0.0", description="Deterministic nightly-rate quote")

NIGHTLY_RATES = {"lisbon": 140, "porto": 120}


class TripRequest(BaseModel):
    destination: str
    nights: int
    travellers: int
    confident: bool


@pricing.skill()
def quote(destination: str, nights: int, travellers: int) -> dict:
    nightly = NIGHTLY_RATES.get(destination.strip().lower(), 130)
    return {"destination": destination, "currency": "EUR", "total": nightly * nights * travellers}


@intake.traversal(timeout_seconds=180)
async def triage(request: str) -> dict:
    parsed = await intake.ai(system="Extract destination, nights and travellers.", user=request, schema=TripRequest)
    if not parsed.confident:
        return {"status": "needs_review", "request": request}
    result = await intake.traverse("pricing.quote", destination=parsed.destination, nights=parsed.nights, travellers=parsed.travellers)
    return {"status": "quoted", **result}

Each node can run in its own worker process or share one process as shown below. intake.triage needs a model configured through ATFGRAPH_MODEL_URL and ATFGRAPH_MODEL_NAME; pricing.quote needs no provider. Migrated-operations tests use a fixture model and check successful steps for both intake.triage and pricing.quote.

python -m atfgraph.worker --module examples.databricks.two_node.pricing.agent:app
python -m atfgraph.worker --module examples.databricks.two_node.intake.agent:app
python -m examples.databricks.two_node.call "Two people, four nights in Lisbon"

Read both steps from the caller's path, as in examples/databricks/two_node/call.py:

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("intake.triage", {"request": "Two people, four nights in Lisbon"})
        print(await handle.result(timeout=180))
        for step in (await handle.path()).items:
            print(step.target, step.status, step.parent_step_id, len(step.attempts))


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

A parent releases its worker permit while awaiting a child, within the max_waiting bound. This permits nested work on a single-capacity worker. See Calling other nodes for limits, concurrency and cancellation.

Discover capabilities at runtime

Use app.discovery, or its app.discover(**filters) shorthand, to inspect the authorized registry. Filters are applied locally to that snapshot: label, version when a label is supplied, name, kind, case-insensitive text query, limit and offset. Each DiscoveredCapability retains its exact version and schemas. Calling .invoke() selects that version and becomes a child call when used inside a handler. The tool-discovery example uses:

from atfgraph import Agent

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


async def worker_targets(query: str = "", limit: int = 8) -> list[str]:
    capabilities = await app.discovery.capabilities(label="utility-worker", kind="skill", query=query, limit=limit)
    return [item.target for item in capabilities]


@app.traversal()
async def inspect_tools(query: str = "") -> dict:
    return {"targets": await worker_targets(query)}


@app.traversal()
async def weather_report(cities: str) -> dict:
    result = await app.ai(user=f"Report the example weather for {cities}", tools=await worker_targets("weather"), max_tool_rounds=3, max_tool_calls=5)
    return {"report": str(result)}

inspect_tools needs no model; weather_report offers discovered targets to app.ai and needs a provider. resolve("label.capability") returns a match or raises NotFoundError with capability_not_found. invoke(target, input, version=...) resolves and calls in one operation. Discovery requires invoke or admin. A key in another namespace sees an empty registry, as checked by the discovery-identity tests.

AgentMesh: a view restricted to labels

AgentMesh(client, labels) restricts its discovery view to named labels. Asking for another returns mesh_target_not_found. Calls still execute durably on the graph server. The mesh does not start workers or host nodes, and its local label restriction does not replace server authorization. Use it to constrain an orchestrator's discovery view.

Several nodes in one process

Run serve() concurrently for multiple Agent objects to host their workers in one process against the same server. The version-routing tests do this for fares 1.0.0 and 2.0.0:

import asyncio

from atfgraph import Client
from examples.python.features.version_routing.agent import v1, v2


async def main() -> None:
    stop = asyncio.Event()
    async with Client("http://127.0.0.1:18880") as client:
        v1.bind_client(client)
        v2.bind_client(client)
        workers = [asyncio.create_task(v1.serve(stop=stop)), asyncio.create_task(v2.serve(stop=stop))]
        try:
            print(await client.traverse("fares.quote", {"route": "BNE-SYD", "adults": 2}, timeout=30))
        finally:
            stop.set()
            await asyncio.gather(*workers, return_exceptions=True)


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

Each agent registers its definition and enrolls its worker separately. Calls between them still pass through the server and appear on the path.

Version routing

A registered (label, version) has an immutable definition. Workers claim tasks only for their enrolled version. The version-routing example defines:

from atfgraph import Agent

v1 = Agent(node_id="fares", version="1.0.0", description="Fare quotes, bags included", capacity=2)
v2 = Agent(node_id="fares", version="2.0.0", description="Fare quotes with unbundled baggage", capacity=2)


@v1.traversal()
async def quote(route: str, adults: int = 1) -> dict:
    return {"route": route.upper(), "total": 149.0 * max(1, adults), "served_by": "1.0.0"}


@v2.traversal(name="quote")
async def quote_v2(route: str, adults: int = 1, bags: int = 0) -> dict:
    return {"route": route.upper(), "total": 149.0 * max(1, adults) + 35.0 * bags, "served_by": "2.0.0"}


@v2.skill()
async def seat_map(route: str) -> dict:
    return {"route": route.upper(), "available": ["3A", "3B"]}

The real-daemon feature tests check these routes:

Version selection is available on Client.submit, Client.traverse through submit kwargs, Agent.submit, Discover.invoke and CLI --node-version. Agent.traverse does not reserve a version option: for a pinned child, use await (await app.submit(target, payload, version="1.0.0")).result() or app.discovery.invoke(target, payload, version="1.0.0"). Inspect the newest schema with client.capabilities("fares"), or select one with client.capabilities("fares", "1.0.0").

Publications

Administrators can publish a versioned descriptor with app.discovery.publish(label, version=..., protocol="http" | "mcp" | "a2a", published=..., display_name=..., description=...). Search using publications(query=...), or remove only the descriptor with remove_publication(id). Publication does not start a worker or enable a protocol.

The discovery-identity tests check that readers cannot publish and that a disabled protocol returns protocol_disabled. Read policy with app.discovery.settings(); change deployment policy in the server YAML and restart. See Publishing.

OfflineMesh for tests without a server

Import OfflineMesh from atfgraph.offline, rather than the package root, for multi-node tests without a server. It mounts agents as ASGI apps under /{node_id} and implements selected /api/v2 operations in memory, including submissions, paths, events, memory, checkpoints, progress and usage. Member calls execute locally as child calls; mesh.discovery is an AgentMesh over that registry. The offline-mesh example uses:

import asyncio

from atfgraph.offline import OfflineMesh
from examples.python.features.offline_mesh.agent import concierge, fares


async def main() -> None:
    async with OfflineMesh([concierge, fares]) as mesh:
        result = await mesh.invoke("concierge.plan", {"origin": "BNE", "cities": ["SYD", "MEL"], "nights": 2}, session_id="trip-1")
        recalled = await mesh.invoke("concierge.recall", {}, session_id="trip-1")
        handle = await mesh.submit("concierge.plan", {"origin": "PER", "cities": ["ADL"]})
        await handle.result(timeout=10)
        path = await handle.path()
        print(result["total"], recalled, [(s.target, s.status) for s in path.items])
        print(mesh.usage(handle.id), [c.target for c in await mesh.discovery.capabilities()])


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

python -m examples.python.features.offline_mesh.run

These agent objects can also serve unchanged against a real graph server. Offline execution has no persistence and cannot establish production recovery or permission behavior. Unsupported routes, including worker leases, approvals, knowledge, publications and retention, return 501 offline_unsupported; inspect mesh.graph.unsupported to see attempted calls. With optional uvicorn, mesh.serve and serve_async expose its ASGI app over TCP. Use real-daemon tests for graph runtime behavior.

Next

• Handle concurrency and cancellation in Calling other nodes.
• Add retries and budgets with Resilience controls.
• Define cross-node permissions in Identity and access.