Usage, logs & tracing

Use the shared usage store, logs and run metadata to understand what a node did. You can read these records through the SDK or HTTP, send signed events to another service, and export traces to MLflow or OpenTelemetry. Cost totals include only known prices; node log shipping requires an administrator to enable it.

For the app.ai example, configure ATFGRAPH_MODEL_URL and ATFGRAPH_MODEL_NAME. Install ./python[mlflow] for MLflow or ./python[otel] for OpenTelemetry. Usage accounting itself needs neither tracing extra.

What is recorded without asking

Agent model and media helpers record CostEntry objects for the current traverser, including completed realtime turns. Token counts come from provider usage. Prices can come from provider billing fields or an optional configured PriceTable (Agent(prices=...) or ATFGRAPH_MODEL_PRICES); table prices are labelled estimates. An unknown price stays None, distinct from a reported zero. Local harness calls need explicit usage recording as described in the harness chapter; do not assume their usage appears automatically.

Inside a worker task, app.record_usage sends an entry to three destinations:

A reporting failure does not fail the handler. The tracker retains the entry, while usage-event and server reporting can fail independently. The SDK logs reporting failures and exposes counters through app.usage_reporting().

Return the cost of a traverser

This example, adapted from examples/python/features/multimodal/agent.py, needs a chat provider that accepts image inputs.

from typing import List, Optional

from pydantic import BaseModel

from atfgraph import Agent, Image

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


class Findings(BaseModel):
    summary: str
    objects: List[str]
    readable_text: Optional[str] = None


@app.traversal()
async def describe(image_url: str, question: str = "What is in this image?") -> dict:
    findings = await app.ai(question, Image.from_url(image_url), schema=Findings)
    return {"findings": findings.model_dump(), "cost": app.execution_cost()}

app.execution_cost() returns call counts, input/output/total tokens, known cost, unpriced-call count and a breakdown by model. total_cost_usd sums known prices and is None when none are known; it can therefore be a partial total. Check calls_without_price before treating it as the full cost. Table estimates carry cost_source="table", and the Console marks them estimated. Use module-level execution_cost(ctx) for an explicit context, or use_tracker() to track code outside a worker task. See model prices.

Record usage yourself

If your handler calls a provider directly, record its usage explicitly. usage_from_response is in atfgraph.costs; this example comes from tests/test_mlflow_tracing.py.

from atfgraph import Agent
from atfgraph.costs import usage_from_response

app = Agent(node_id="mlf-node", version="1.0.0")


@app.traversal()
async def run(n: int) -> dict:
    await app.record_usage(
        usage_from_response({"prompt_tokens": 7, "completion_tokens": 5}, model="ollama/llama3", provider="ollama")
    )
    return {"value": n * 2}

You can construct a CostEntry directly. For example, the integration suite records CostEntry(model="openrouter/qwen", provider="openrouter", input_tokens=10, output_tokens=5, total_tokens=15, cost_usd=0.0012, cost_source="provider"). The shipper assigns <step>-<attempt>-<n> call IDs. Resending the same call is deduplicated; a new attempt that makes another provider call records another charge.

Interpret Console spend

Console distinguishes the following accounting states:

Decision diagnostics recorded with source="decision", such as a coded fallback after an exception, remain in the ledger and outcome counts. They do not increment model-call, token, spend or unpriced-call totals. An exception does not prove that the provider spent zero tokens. Successful HTTP responses are accounted for before structured-output validation: a schema repair records both calls, and a rejected response retains its reported usage even if the handler fails. Calls made directly with a third-party client need explicit usage reporting.

The run Debug summary shows spend alongside model calls and tokens. The agent Test panel uses server totals across the whole run, including all usage pages and child steps. Aggregate spend can include rate estimates; it is not an invoice. Evaluation summaries lack call counts, so missing cost appears as —; open the run to inspect its usage.

Read usage from a client

A caller with permission to read the traverser can also read its usage progress events. This example is adapted from examples/python/features/multimodal/call.py.

import asyncio
import os

from atfgraph import Client, UsageEvent


async def show(client: Client, target: str, payload: dict) -> None:
    handle = await client.submit(target, payload)
    print(target, await handle.result(timeout=300))
    async for event in client.events(handle.id):
        usage = UsageEvent.from_event(event)
        if usage is not None:
            print(f"  usage {usage.operation:<14} {usage.model:<28} tokens={usage.total_tokens} cost_usd={usage.cost_usd}")


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        await show(client, "media.describe", {"image_url": os.environ["DEMO_IMAGE_URL"]})


asyncio.run(main())

For stored usage, use app.observability. Outside a node, create the same facade with Observability.for_client(client). The integration suite uses these calls:

from atfgraph import Client, Observability


async def report(url: str, traverser_id: str) -> None:
    async with Client(url) as client:
        obs = Observability.for_client(client)
        page = await obs.usage(traverser_id=traverser_id)
        for record in page.items:
            print(record.model, record.total_tokens, record.cost_usd, record.cost_known, record.capability)
        totals = await obs.usage_totals(group_by="model")
        print(totals.totals.calls, totals.totals.unpriced_calls)
        print({key: bucket.cost_complete for key, bucket in totals.by_key.items()})

By default, usage_totals covers the caller's runs. Group by run, model, provider, day, attempt_status, label, principal or none; since and until use epoch milliseconds. Administrators can select a principal or set all_principals=True. A bucket's cost_complete is true only when every call has a price.

Reporting counters

app.usage_reporting() returns enabled, recorded, deduplicated, failed, events_failed and last_error. Agent(report_usage=False) disables posts to /api/v2/usage; the local tracker and usage progress events still work unless separately disabled.

Node logs

To ship Python logs, an administrator first enables node logs for the namespace. The node then installs its logging handler:

import logging

from atfgraph import Agent

app = Agent(node_id="sdk-logs")
logger = logging.getLogger("myapp")


@app.traversal()
async def work(step: str) -> dict:
    logger.info("handling %s", step)
    await app.node_logs.flush_async()          # make these lines durable before the attempt ends
    return {"traverser_id": app.ctx.traverser_id}


async def opt_in() -> None:
    await app.observability.set_settings(node_logs_enabled=True)     # administrator, once per namespace


handler = app.enable_node_logs(logger, level=logging.INFO, max_pending=1000, batch_size=100, flush_interval=1.0)

The real-daemon integration suite checks that logs sent before namespace opt-in are rejected with node_logs_disabled and counted in handler.failed. Once enabled, the handler follows these rules:

• Only records emitted during a traversal are shipped. Logs outside a traversal stay local, as do shipping-library logs (atfgraph.observability, atfgraph.transport, httpx, httpcore) and any exclude= prefixes.
• The queue holds max_pending records. When full, it drops the oldest record and increments handler.dropped.
• The flusher runs with serve(). Sending requires the task's lease, so queued records can become unsendable after an attempt ends. Flush within a long handler with await app.node_logs.flush_async().
• await app.disable_node_logs() detaches the handler and flushes queued records. Repeating disable is harmless; enabling twice raises RuntimeError.

Query stored logs with await app.observability.query_logs(traverser_id=..., levels=["warn", "error"], contains="desk"). Levels are trace, debug, info, warn, error, fatal and log; Python warning and critical map to warn and fatal.

stream_logs(root_id=...) follows GET /api/v2/logs/events, resumes with Last-Event-ID, and stops after the terminal run's final batch. tail_logs provides a polling alternative. To send a single task log without installing a handler, call app.observability.log("message", "warn", step=1).

Run metadata

Store labels, display names and trace links in the root traverser's metadata document. The integration suite uses this pattern:

from atfgraph import Agent

app = Agent(node_id="sdk-meta")


@app.traversal()
async def tag() -> dict:
    await app.observability.set_run_metadata(set={"display_name": "Trip 21", "labels": ["demo"]})
    return {"root_id": app.ctx.root_id}


async def read_back(root_id: str) -> None:
    metadata = await app.observability.run_metadata(root_id)
    print(metadata.metadata, metadata.mlflow_trace_id, metadata.updated_at)

Inside a handler, the SDK takes the root ID from its context. Outside one, pass root_id. set_run_metadata(set=..., unset=[...]) merges changes into a document of at most 64 keys. Conventional keys include display_name, labels, links, mlflow_trace_id and mlflow_experiment_url.

The namespace webhook

A namespace administrator can configure one outbound webhook URL for observability events. This differs from inbound webhooks that start traversers. For example:

from atfgraph import Client, Observability


async def configure(url: str) -> None:
    async with Client(url) as client:
        obs = Observability.for_client(client)
        config = await obs.set_webhook(
            "https://hooks.example/atfgraph",
            secret_env="HOOK_SECRET",
            headers={"X-Team": "pricing"},
            event_types=["usage.recorded", "run_metadata.updated"],
        )
        assert config.secret_present
        status = await obs.webhook_status()
        print(status.pending, status.delivered, status.failed)
        delivery_id = await obs.test_webhook("ping")
        redriven = await obs.redrive_webhook(limit=100)
        print(delivery_id, redriven)
        await obs.delete_webhook()

secret_env names an environment variable on the server host; the SDK does not send its secret value. The Python helper currently accepts usage.recorded, node_logs.appended, run_metadata.updated and webhook.test. Each delivery carries an x-atfgraph-signature: sha256=... HMAC over its body. Verify that signature at the receiver; tests/test_mlflow_tracing.py includes a bridge example.

run.completed

The server also supports run.completed, emitted once when a root traverser reaches a terminal state. The current Python set_webhook convenience validator omits this event, so configure it through PUT /api/v2/observability/webhook using the HTTP contract rather than that helper.

The signed payload includes root_id, traverser_id, target, label, version, status, principal, session_id, started_at, finished_at, duration_ms, optional error, and output_bytes. It excludes the output; fetch that through GET /api/v2/traversers/{id}. Child steps do not emit this event. Use it for completion notifications, or keep Client.wait and the event stream when the caller needs the result directly.

MLflow tracing

MLflow tracing is optional. Importing atfgraph does not import MLflow or contact a tracking server. Enable it explicitly, as in python/atfgraph/mlflow.py and its tests:

import asyncio

from atfgraph import Agent
from atfgraph import mlflow as tracing

app = Agent(node_id="trace-node", version="1.0.0")


@tracing.trace(kind="tool")
async def score(nights: int) -> float:
    await asyncio.sleep(0.01)
    return nights * 1.5


@app.traversal()
async def plan(city: str, nights: int) -> dict:
    with tracing.span("lookup", kind="retriever", attributes={"atfgraph.city": city}) as live:
        if live is not None:
            live.set_attribute("rates", 3)
    return {"city": city, "score": await score(nights)}


if __name__ == "__main__":
    tracing.enable(app, tracking_uri="sqlite:///./data/mlflow.db")
    app.run()

If MLflow is missing, enable(app) warns once and returns False. Otherwise it wraps this Agent's handlers, child traversals, model calls, harness calls and usage recording. Each worker task gets a trace tagged with its root, node, capability and session. The root writes trace links into run metadata; failed metadata writes increment tracing.stats()["metadata_failed"] without failing the handler.

Inputs and outputs are excluded unless you enable capture_inputs or capture_outputs; use redact to filter captured values. enable_from_env(app) reads ATFGRAPH_MLFLOW_TRACING=true. Call tracing.flush() before process exit. See MLflow tracing for configuration, span types and the bridge for externally traced runs.

OpenTelemetry export

Use atfgraph.otel to export traces to an OTLP collector, including services such as Grafana Tempo, Honeycomb or Datadog. Install ./python[otel]. The integration is optional, does not load on import atfgraph, and does not install a global tracer provider.

from atfgraph import Agent
from atfgraph import otel

app = Agent(node_id="trace-node", version="1.0.0")


@app.traversal()
async def plan(city: str, nights: int) -> dict:
    with otel.span("lookup", attributes={"atfgraph.city": city}):
        return {"city": city, "nights": nights}


if __name__ == "__main__":
    otel.enable(app, endpoint="http://localhost:4318", service_name="atfgraph.trace-node")
    app.run()

Handlers create SERVER spans with node, capability, traverser, root, session, attempt and kind attributes. app.traverse creates a CLIENT span and propagates W3C traceparent to the child. Model spans include gen_ai.request.model; recorded usage adds token and cost attributes. The root writes otel_trace_id and otel_traceparent into run metadata for the Console.

Set ATFGRAPH_OTEL_TRACING=true, OTEL_EXPORTER_OTLP_ENDPOINT, optional OTEL_EXPORTER_OTLP_HEADERS, and OTEL_SERVICE_NAME (default atfgraph.<node_id>). Agent.serve calls enable_from_env, so installed nodes can use their package .env. ATFGRAPH_OTEL_CAPTURE_PAYLOADS=true includes inputs and outputs. otel.flush() drains the exporter; otel.stats() reports spans, failures and metadata writes.

Next

• ./06-model-calls.md for the calls that produce these usage entries.
• ./13-identity-credentials-and-access.md for the lease and role model behind worker writes.
• ./15-testing.md for running these journeys against a real daemon.