Skills, traversals, typed input/output & routers

Define traversals and skills, derive input and output schemas from Python types, choose synchronous or asynchronous handlers, and group functions with AgentRouter. These examples need the SDK and a graph server, with no model provider configured.

Two capability kinds

Both decorators register callable capabilities. Their recorded kind explains their intended role, as described in Core concepts:

The worker binds input, runs the handler and reports completion in the same way for both kinds (python/atfgraph/worker.py). The SDK does not enforce the convention by blocking a skill's app.traverse call. Follow it in your application so paths remain understandable: traversals coordinate work; skills perform individual operations. Both appear as steps and are listed by atfg ls.

By default, the registered capability uses the function name. @app.skill("upper") overrides it with upper. Call it as label.capability, or use app.target("greet") to construct a target for the current node.

Wrap an external operation as a skill

A skill can call an HTTP API, query a database, invoke an MCP tool, read a file or compute a result. Here, deterministic means that it does not call a model or coordinate further graph calls; it may still perform I/O. In examples/python/co_consult, hotel_search wraps SerpAPI, flight_shop wraps Sabre and context_lookup wraps an MCP tool.

The layers are:

external API or MCP tool  →  @app.skill()  →  a capability on the graph  →  offered to the model as a tool

The model receives your skill's input names, description and result rather than directly controlling the underlying API. You supply validation, bounds and error handling; graph calls record each invocation as a step. This gives the model an application-defined tool interface.

import httpx

from atfgraph import Agent

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


@app.skill()
async def forecast(city: str, days: int = 3) -> dict:
    """Daily forecast for a city from the weather API; days 1-7."""
    async with httpx.AsyncClient(timeout=20) as http:
        response = await http.get("https://api.example.com/forecast", params={"q": city, "days": min(max(days, 1), 7)})
        response.raise_for_status()
        return {"city": city, "days": response.json()["daily"]}


@app.traversal()
async def advise(question: str) -> str:
    # tools="self" offers this node's capabilities; the model sees weather__forecast(city, days)
    return await app.ai(
        system="Answer travel weather questions. Use the forecast tool; never guess.",
        user=question,
        tools="self",
    )

atfg traverse weather.advise --in '{"question": "Do I need a jacket in Tokyo this weekend?"}'

The path contains advise and a weather.forecast child step, which wraps the API request. The model uses the returned forecast. If the skill raises, the tool loop returns an error to the model; your application can then handle the failure. The model-call chapter explains the loop.

Calling an existing MCP tool happens inside your skill. Publishing your own capabilities happens at the graph server: enable ATFGRAPH_MCP_ENABLED or ATFGRAPH_A2A_ENABLED and use app.discovery.publish(...) as described in Multi-node systems. If a model tool must coordinate another node, define it as a traversal rather than a skill; traversals can also be offered as tools.

Two more kinds: decision and realtime

Use @app.decision() for a structured judgement. Its model call can record confidence through confidence_field on app.ai, allowing usage summaries to count low-confidence results and fallbacks.

Use @app.realtime(name, provider=, transport=, model=None) for a realtime entrypoint. It can be discovered, tagged and published over HTTP. Normal traverser submission returns 400 realtime_entrypoint: the node or a browser through that node opens the session instead. See Hosting and examples/python/realtime_browser.

Tags reach the server

Tags declared on capabilities, routers and nodes are included in registration. Each capability can have up to 16 distinct tags, each 1–64 characters using [a-z0-9_.:-]. Invalid tags return 400 invalid_tags. Use tags for agent and capability filtering, tag-scoped access policies and discovery search through POST /api/v2/ard/search {"tags": [...]}. HTTP filters are GET /api/v2/agents?tag= and GET /api/v2/agents/{label}/capabilities?tag=.

Sync and async handlers

The first-node example contains both handler styles:

from atfgraph import Agent

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


@app.skill()
def shout(text: str) -> dict:
    """Synchronous: runs in a worker thread via asyncio.to_thread."""
    return {"text": text.upper()}


@app.traversal()
async def greet(name: str) -> dict:
    """Asynchronous: runs on the worker's event loop."""
    loud = await app.traverse("hello.shout", text=f"hello {name}")
    return {"greeting": f"Hello {name}", "shouted": loud["text"]}

An async def handler runs on the worker's event loop. A plain def runs in a thread through asyncio.to_thread. Lease renewal runs independently, so blocking synchronous work can retain its lease; test_heartbeats_keep_lease_during_blocking_sync_handler checks this. Keep two thread-related limits in mind:

• A synchronous handler cannot use await. Use app.traverse_sync(target, **input) to schedule a child call on the event loop and wait in the thread. Calling this method from the loop thread raises RuntimeError. This is covered by test_traverse_sync_from_thread_handler.
• Cancellation cannot interrupt a running thread. An asynchronous handler can receive cancellation at an await; a synchronous handler finishes and its result is discarded. Prefer asynchronous handlers for long operations.

current_context() works in either style. The worker binds an ExecutionContext with contextvars, and asyncio.to_thread copies it into the thread (python/atfgraph/context.py).

Typed input

input_schema_for in python/atfgraph/schema.py builds an object schema from handler parameters. The worker's coerce_input uses the following binding rules:

Missing required inputs, unknown inputs without **kwargs, and Pydantic validation failures produce invalid_input with retryable: false. Normal submissions are also checked against the registered schema at admission, before the worker runs. Trigger-specific input handling is described in the emitter chapter.

Typed output

output_schema_for derives a schema from the return annotation. No annotation, or Any, yields {}, allowing any JSON value. The worker's encode_output converts Pydantic objects with model_dump(mode="json"), dataclasses with asdict, and enums to their values. The first-node example declares:

from pydantic import BaseModel


class Greeting(BaseModel):
    greeting: str
    shouted: str
    length: int

Return the declared type: the server validates completion output. tests/python_test_examples.py checks that hello requires greeting, shouted and length. test_registered_schema_rejects_bad_output checks that a string cannot successfully complete a capability registered to return an object.

Capability options

The decorators accept these options, defined by AgentRouter.add in python/atfgraph/agent.py:

This example comes from test_decorators_derive_capabilities_and_routers_merge:

from atfgraph import Agent

app = Agent(node_id="node", version="0.0.1")


@app.traversal(max_attempts=3, timeout_seconds=10)
def plan(brief: dict, nights: int = 3) -> dict:
    return {"ok": True}


caps = app.capabilities()
assert caps["plan"].max_attempts == 3 and caps["plan"].timeout_seconds == 10
assert caps["plan"].input_schema["required"] == ["brief"]

Routers

Use AgentRouter(prefix="", tags=None) to organize related handlers before including them in an Agent. Decorated functions remain directly callable. The co-consult example has a router per desk, such as desks/intake.py and desks/flights.py, all included by main.py:

from atfgraph import Agent, AgentRouter

NODE_ID = "co-consult"
app = Agent(node_id=NODE_ID, version="0.3.0")
router = AgentRouter(tags=["intake"])


@router.traversal()
async def extract_brief(request: str, today: str) -> dict:
    """Given the traveller's sentence, return the structured facts it contains."""
    return {"request": request, "today": today, "confident": True}


@router.traversal()
async def intake_brief(request: str) -> dict:
    extracted = await router.traverse(f"{NODE_ID}.extract_brief", request=request, today="2026-09-15")
    return {"status": "ready", "brief": extracted}


app.include_router(router)

Router behavior is defined in agent.py and checked by the worker tests:

• A prefix joins to a capability name with _: AgentRouter(prefix="desk") and lookup register desk_lookup, called as node.desk_lookup. Co-consult uses distinct names without prefixes.
• router.traverse, submit, ai and memory delegate to the including Agent. Before inclusion and without a handler context, they raise NoExecutionContext.
• Duplicate capability names raise ValueError with capability ... is registered twice. Reusing a router in a second agent, or including an Agent inside another Agent, also raises ValueError.
• Adding a capability after include_router also adds it to the agent.

Inspecting what will be registered

Inspect app.capabilities() for the {name: Capability} mapping. app.definition() returns the registration body: label, version, node tags and each capability's kind, schemas, retry policy, timeout, description and tags. Realtime capabilities also carry their binding. app.entry(name) includes the underlying function, tags and triggers. Changing a registered definition requires a new version; otherwise registration raises ConflictError.

For a unit test, app.call_local(capability, **input) invokes a local handler with normal input conversion and output encoding. It creates no traverser and records no path, so it tests handler logic rather than graph execution.

Retained module-level decorators

The retained module-level traversal and skill decorators register with the most recently created Agent, through Agent.get_current(), or with an explicitly supplied agent=. They support earlier application code. Prefer @app.traversal() and @app.skill() for new code; this guide uses those forms.

Next

• Coordinate work in Calling other nodes.
• Configure capability triggers in Emitters.
• Look up methods in the Python API reference.