Model calls

Call a model with app.ai, request typed output, offer capabilities as tools, stream responses and work with images and audio. You will also learn how media generation and cost recording work. You need the SDK, a graph server and a provider reachable from the worker for live calls.

For chat examples, configure ATFGRAPH_MODEL_URL and ATFGRAPH_MODEL_NAME, or select ATFGRAPH_MODEL_PROVIDER=anthropic with ANTHROPIC_API_KEY and a model name. Media generation needs its provider key, such as OPENAI_API_KEY. Each section identifies the required configuration.

Configure a provider

An Agent reads provider settings when it is constructed, unless you pass model=. model_from_env in python/atfgraph/model.py performs that selection. A model request starts only when a handler calls app.ai.

Without model configuration, app.model is None. The first app.ai call raises ModelNotConfigured with "no model provider: pass model= to Agent or set ATFGRAPH_MODEL_URL and ATFGRAPH_MODEL_NAME (and optionally ATFGRAPH_MODEL_PROVIDER=anthropic)".

Provider classes are HTTPChatModel and AnthropicMessagesModel; test_model_from_env_selects_provider checks selection. Tests can assign a scripted HTTPChatModel(url, name, transport=...) to app.model. For routing by model name, use ModelRouter.

Per-provider recipes

The same node code works with these provider configurations. With reachable endpoints and credentials, each configuration enables the plan example in Traversals and skills to call a model.

# Databricks Model Serving: the workspace URL, not an App URL; the endpoint name is the model.
# Two ways to authenticate, both supported:
export ATFGRAPH_MODEL_URL=https://adb-1234567890.12.azuredatabricks.net/serving-endpoints
export ATFGRAPH_MODEL_NAME=databricks-claude-sonnet-5
#  (a) a token you hold: `databricks tokens create`, or a service-principal OAuth token, with CAN_QUERY
export ATFGRAPH_MODEL_API_KEY=dapi-your-token
#  (b) no token at all: leave ATFGRAPH_MODEL_API_KEY unset and the call is signed by the workspace identity
#      through databricks-sdk unified auth -- inside a Databricks App that is the App's own service principal
#      (grant it CAN_QUERY on the endpoint), on a laptop an explicit ATFGRAPH_DATABRICKS_PROFILE. Needs the
#      [databricks] extra. ATFGRAPH_MODEL_AUTH=databricks forces (b) even when a key is set; =key forbids it.

# OpenRouter: reports usage.cost, so recorded calls carry a price
export ATFGRAPH_MODEL_URL=https://openrouter.ai/api/v1
export ATFGRAPH_MODEL_NAME=anthropic/claude-sonnet-4
export ATFGRAPH_MODEL_API_KEY=sk-or-your-key

# Anthropic directly
export ATFGRAPH_MODEL_PROVIDER=anthropic
export ATFGRAPH_MODEL_NAME=claude-sonnet-4-20250514
export ANTHROPIC_API_KEY=sk-ant-your-key

# A local OpenAI-compatible server (Ollama, vLLM): no key
export ATFGRAPH_MODEL_URL=http://127.0.0.1:11434/v1
export ATFGRAPH_MODEL_NAME=llama3.1

The App preset at examples/databricks/presets/databricks-model-serving.yml uses the Databricks configuration with a secret-scope token binding. openrouter.yml provides the corresponding OpenRouter configuration.

Where the key lives, by how you run the worker

Model requests execute in the worker, which resolves provider credentials from its environment. Declare required variable names in the node manifest so installation and startup can check them. This is separate from the graph server's API credential. Deployment tooling may stage and bind secrets, but that does not make the graph server the model-call executor.

List required environment names in agentnode.yaml. This example is from the model-tools node:

user_environment:
  required:
    - name: ATFGRAPH_MODEL_URL
      description: OpenAI-compatible chat base URL
    - name: ATFGRAPH_MODEL_NAME
      description: Model or serving-endpoint name
  optional:
    - name: ATFGRAPH_MODEL_PROVIDER
      default: openai-compatible
    - name: ATFGRAPH_MODEL_API_KEY
      type: secret
      description: Provider key when the endpoint needs one

The manifest declares names and types, without secret values. A worker with no configured provider can still serve non-model skills. Calling app.ai then raises ModelNotConfigured and identifies the missing configuration.

Several providers in one node

Use ModelRouter when a node needs different providers or model strengths for different operations. Construct HTTPChatModel(base_url, model, *, api_key_env=...) or AnthropicMessagesModel(model, *, api_key_env=...) with the environment variable holding each key. The provider reads its value from the environment. See Resilience controls for routing rules.

from __future__ import annotations

from pydantic import BaseModel

from atfgraph import Agent, AnthropicMessagesModel, HTTPChatModel, ModelRouter

router = ModelRouter()
router.register(
    "databricks",
    HTTPChatModel(
        "https://adb-1234567890.12.azuredatabricks.net/serving-endpoints",
        "claude-sonnet-4-endpoint",
        api_key_env="DATABRICKS_TOKEN",
    ),
    models=["claude-sonnet-4-endpoint"],
    default=True,
    default_model="claude-sonnet-4-endpoint",
)
router.register(
    "openrouter",
    HTTPChatModel("https://openrouter.ai/api/v1", "openai/gpt-4o-mini", api_key_env="OPENROUTER_API_KEY"),
    prefixes=["openai/", "google/"],
)
router.register(
    "anthropic",
    AnthropicMessagesModel("claude-opus-4-1", api_key_env="ANTHROPIC_API_KEY"),
    models=["claude-opus-4-1"],
)

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


class Route(BaseModel):
    needs_expert: bool
    reason: str


@app.traversal()
async def triage(message: str) -> dict:
    verdict = await app.ai(user=message, model="openai/gpt-4o-mini", schema=Route)  # OpenRouter, by prefix
    if verdict.needs_expert:
        return {"reply": await app.ai(user=message, model="claude-opus-4-1")}        # Anthropic, by name
    return {"reply": await app.ai(user=message)}                                      # default route: Databricks

Usage records keep the selected provider and model, so totals can show the split. Not verified: live access to the three endpoints in this example. test_python_controls.py checks routing with a fixture provider.

Structured output on Databricks serving endpoints

For an OpenAI-compatible endpoint, app.ai(schema=…) tries these structured-output forms in order and remembers the accepted form for the rest of the process:

1. Strict JSON Schema on Databricks: The SDK inlines nested models, represents optional values as nullable types, removes title and default, and requires all properties. Schemas still using pattern, allOf, oneOf or recursive references are sent without strict mode. ATFGRAPH_MODEL_STRICT=true requests strict mode on any host; false disables it. Recorded live behavior showed databricks-claude-sonnet-5 accepting strict output and returning previously omitted fields.
2. Plain JSON Schema: If strict mode returns 400 or 422, retry response_format: json_schema without strict mode.
3. Forced tool call: Send the schema as function parameters and parse the returned arguments.
4. Schema in the prompt: Omit response_format and tools, describe the schema in a system instruction, and validate the first JSON object in the response, tolerating prose or fences. Recorded Claude Fable endpoints (databricks-claude-fable-5, -5-1) accepted this form while rejecting structured output and forced tool use with "Structured output is not supported for this model…" and "tool_choice: type tool and any are not supported".

ATFGRAPH_MODEL_STRUCTURED=json_schema starts with the full sequence. Set tools or prompt to start at the corresponding form. A final 400/422, or another HTTP error, raises ModelError with scrubbed endpoint .status and .detail. For example, Sonnet's rejected temperature parameter appears in that error.

An invalid structured reply gets one repair attempt. The SDK sends back the reply and validation error, then validates the new response. A second invalid result raises ModelError; successful repair sets ModelResponse.repaired. Set ATFGRAPH_MODEL_REPAIR=0 to fail on the first invalid result.

An application's fixture fallback can hide a provider failure behind a successful run. Co-consult marks such results with usage operation: fallback, a model call failed: ModelError message or a degraded metadata flag. Check these indicators when assessing a run.

One call, typed output

app.ai("text") and app.ai(user="text") are equivalent. Without schema, the result is text. With a Pydantic model or JSON Schema dict, it requests structured output and returns the parsed value. system, model, temperature and max_tokens pass through, while context is appended to the user message as JSON. This example comes from the image-studio node:

from __future__ import annotations

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="image-studio", version="1.0.0")


class DetailedPrompt(BaseModel):
    prompt: str
    style_notes: str


@app.traversal()
async def create_prompt(topic: str) -> dict:
    answer = await app.ai(
        user=topic,
        system="Write a vivid one or two sentence image prompt with composition, colour and lighting.",
        schema=DetailedPrompt,
    )
    return answer.model_dump()

Use response_format="json" for a parsed dict without a schema, or return_response=True for the full ModelResponse, including text, parsed output, model, usage, tool calls, finish reason and provider. A system-only request without user content raises ValueError. Unsupported keywords, such as the earlier runtime's memory_scope=, raise TypeError. The positional-input test checks these forms.

Tools: the model calls your capabilities

Offer this node's capabilities with tools="self". Each selected tool runs through app.traverse and becomes a child step. The default loop limits are eight rounds and 25 tool calls. This example is from features/model_tools:

from __future__ import annotations

from datetime import date
from typing import List

from pydantic import BaseModel

from atfgraph import Agent

app = Agent(node_id="planner", version="1.0.0", description="Structured trip planning with tool calls")

FIXTURE_RATES_TO_AUD = {"AUD": 1.0, "NZD": 0.92, "USD": 1.52, "EUR": 1.65, "JPY": 0.0101}


class TripPlan(BaseModel):
    destination: str
    nights: int
    budget_aud: float
    highlights: List[str]


@app.skill()
def nights(check_in: str, check_out: str) -> dict:
    """Nights between two ISO dates."""
    return {"nights": (date.fromisoformat(check_out) - date.fromisoformat(check_in)).days}


@app.skill()
def convert(amount: float, currency: str) -> dict:
    """Convert an amount to AUD with fixture rates."""
    rate = FIXTURE_RATES_TO_AUD.get(currency.upper())
    if rate is None:
        raise ValueError(f"unknown currency {currency!r}")
    return {"amount_aud": round(amount * rate, 2), "rate": rate}


@app.traversal()
async def plan(brief: str) -> TripPlan:
    answer = await app.ai(
        system="You plan trips. Use the tools for date arithmetic and currency conversion; never guess numbers.",
        user=brief,
        tools="self",
        schema=TripPlan,
    )
    return answer.parsed

Tool-enabled calls return ToolCallResponse with text, optional parsed output, usage and a trace. Each ToolCallRecord includes tool name, target, arguments, result, error, latency and round. The trace also reports round count and stopped_by_limit. Scripted-provider tests check these behaviors:

• Tools use names such as planner__nights, shaped as label__capability. The active capability is excluded from its own tools.
• A failing tool returns an error to the model without automatically failing the coordinating traversal.
• Each invocation creates a path step, including repeated calls to the same capability. The test records one planner.nights and two planner.convert steps.
• At the round limit, the SDK makes a final request without tools and asks the model to answer.

Use tools="discover" to offer capabilities from registered nodes, or a list combining label.capability targets and prepared tool schemas. capability_to_tool_schema(target, capability) builds a schema. Other string values raise ValueError. Tool loops cannot be combined with stream=True or modalities=; those combinations raise TypeError.

With a provider configured, run the example from the repository root:

python -m atfgraph.worker --module examples.python.features.model_tools.agent:app
python -m examples.python.features.model_tools.call

Streaming

stream=True returns ModelStream. Iterate stream.text() for text deltas, or iterate the stream for typed StreamEvent objects: text, tool_call, audio, transcript, usage and done. After completion, stream.response() gives the assembled response and usage is recorded for the current task. The multimodal example uses:

from __future__ import annotations

from typing import List

from atfgraph import Agent

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


@app.traversal()
async def narrate_stream(text: str) -> dict:
    stream = await app.ai(f"Rewrite this as a two-sentence travel teaser: {text}", stream=True)
    chunks: List[str] = []
    async for delta in stream.text():
        chunks.append(delta)
        await app.note(delta, tags=["delta"])
    return {"text": "".join(chunks), "usage": stream.response().usage, "cost": app.execution_cost()}

Forward deltas with app.note when callers should see them through client.events(traverser_id).

Images and audio as input

Combine positional text with Image, Audio, File, pathlib.Path or prepared content parts to build one user message. Image constructors accept a URL, file or bytes. Audio accepts a file or bytes plus format; Audio.from_url accepts only data: URLs. The multimodal example's describe traversal is:

from __future__ import annotations

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()}

Providers declare supported features such as text, vision, audio, files, tools, structured output, streaming and image generation. Unsupported media raises MediaCapabilityError before I/O. test_media_parity.py checks, for example, that Anthropic accepts converted images and documents but rejects audio input. Wrap raw bytes in a typed media object rather than passing them positionally.

ai_with_multimodal is an explicit form accepting modalities and audio_config. ai_with_audio requests spoken output from the chat provider with modalities=["text", "audio"], or routes to media generation when model names a TTS model.

Generating media

Media generation uses app.media, a MediaRouter created from the environment on first use. The model prefix selects its provider, and credentials are read at request time. A missing key raises MediaProviderNotConfigured with the required variable name (python/atfgraph/media_providers.py).

When model is omitted, image generation uses ATFGRAPH_IMAGE_MODEL or dall-e-3, speech uses ATFGRAPH_AUDIO_MODEL or tts-1, and transcription uses ATFGRAPH_TRANSCRIPTION_MODEL or whisper-1. Music and video require either model= or ATFGRAPH_MUSIC_MODEL/ATFGRAPH_VIDEO_MODEL; otherwise MediaCapabilityError asks you to pass a model.

Generated items contain a URL or inline data. save_all(directory, prefix=) writes inline items and skips URL-only items; it does not download them. The multimodal example shows:

from __future__ import annotations

import os
from pathlib import Path
from typing import Optional

from atfgraph import Agent

app = Agent(node_id="media", version="1.0.0")
ARTIFACT_DIR = Path(os.environ.get("ATFGRAPH_MEDIA_DIR", "/tmp/atfgraph-media"))


@app.traversal()
async def poster(prompt: str, model: Optional[str] = None, size: str = "1024x1024") -> dict:
    result = await app.ai_generate_image(prompt, model=model, size=size)
    saved = [str(p) for p in result.save_all(ARTIFACT_DIR, prefix="poster")]
    return {"saved": saved, "urls": [i.url for i in result.images if not i.is_inline], "cost_usd": result.cost_usd, "cost": app.execution_cost()}

Its transcribe handler wraps base64 audio as Audio(input_audio={"data": ..., "format": "wav"}). ai_transcribe_audio also accepts bytes, a data: URL or an HTTP(S) URL. Media-provider tests use httpx.MockTransport. The multimodal example itself has no fixture mode and needs real provider keys to produce media.

What a call cost

Model and generated-media calls record a CostEntry for the current task. Fields identify model, provider, source (llm, media, realtime, harness), operation, token counts, USD cost and cost source. Operations include chat, image, audio, music, video, transcription and realtime. Without a configured rate table, a price exists only if the provider reports one, such as OpenRouter's usage.cost. Missing prices stay None, rather than 0.0; the SDK ships no built-in rates.

Estimate unpriced calls with your rate table. Databricks serving, Azure OpenAI and many self-hosted endpoints return token counts without prices. To estimate them, pass Agent(prices={...}) or set ATFGRAPH_MODEL_PRICES to inline JSON or a JSON-file path:

export ATFGRAPH_MODEL_PRICES='{"databricks-claude-sonnet-5": {"input_per_million": 2.00, "output_per_million": 10.00, "cache_read_per_million": 0.20}}'

A rate table fills a missing provider price when token counts are available. Such entries use cost_source="table" and retain the applied rates in extra["rate"]. A provider-reported price takes priority. The Console shows estimates with ≈ and an estimated label. MLflow and OpenTelemetry record the source, and usage totals include estimated calls as priced.

Rate keys match either the returned or requested model name, without case sensitivity; provider/model keys match the part after /. If those names differ, extra.requested_model preserves the request. For example, a Databricks Sonnet endpoint may return global.anthropic.claude-sonnet-5.

Cache reads and writes are charged once. Each uses its cache-specific rate when present, otherwise the input rate. The SDK detects Anthropic-shaped usage, where input_tokens excludes cache tokens, and records the convention in extra.cache_included_in_input. For manually constructed entries with that convention, set cache_included_in_input: false in the rate table. When the provider supplies no total, the SDK includes separate cache counts in total_tokens. Console token mix and cache read share use total input, so an OpenAI cached token is not counted twice.

Maintain these rates yourself. For Databricks foundation-model serving, convert published DBU-per-million-token rates using your workspace's USD-per-DBU serving rate. A historical 2026-09-23 example used Sonnet rates of 28.571 input, 142.857 output and 2.857 cache-read DBU per million tokens. At an assumed $0.07 per DBU, these produce $2.00, $10.00 and $0.20. These are dated planning assumptions, not current prices or a measured bill; check the pricing page and your contract before using them.

app.execution_cost() returns call count, input/output/total tokens, known USD cost, unpriced-call count and per-model totals. Its keys are total_calls, total_input_tokens, total_output_tokens, total_tokens, total_cost_usd, calls_without_price and by_model. Cost is None if no call has a price; a non-null total can still exclude unpriced calls. Usage also appears in the event log and, by default, the durable usage store. A caller can parse events with UsageEvent.from_event:

from __future__ import annotations

import asyncio
import os

from atfgraph import Client, UsageEvent


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        handle = await client.submit("media.narrate", {"text": "Welcome to Brisbane. Your gate is B12."})
        print(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}")


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

For a provider called outside the SDK, use app.record_usage(CostEntry(...)). Observability covers durable queries, grouped totals and reporting counters.

Not verified: the exact live usage fields and cost availability of every image and speech provider. Mock-transport tests check OpenRouter's usage.cost. Verify other providers by running the multimodal example with their real credentials.

Next

• Delegate a complete coding task with Harnesses.
• Route providers and control calls with Resilience controls.
• Query recorded usage in Observability.