Anthropic Operators¶

The Anthropic provider runs the Claude Message Batches API from Airflow. Message Batches process many messages.create requests asynchronously at 50% of standard cost; most complete within an hour, with a 24-hour SLA — a good fit for Airflow’s deferrable execution model.

Note

For interactive, single-call or agentic LLM workloads, prefer the vendor-agnostic apache-airflow-providers-common-ai provider with model="anthropic:claude-opus-4-8". This provider focuses on the batch/async surface and direct SDK access that the agent abstraction does not model.

AnthropicBatchOperator¶

AnthropicBatchOperator submits a Message Batch and waits for it to reach the terminal ended status. In deferrable mode it releases the worker slot while an AnthropicBatchTrigger polls for completion.

The operator returns the batch ID only. Pull the per-request results with stream_batch_results() and persist them to object storage — results can be very large and must not be pushed to XCom. Results are retained for 29 days after the batch is created.

Parameters¶

requests — a list of {"custom_id": str, "params": {...}} dicts, where params is a messages.create payload (model, max_tokens, messages, …).
conn_id — the Anthropic connection ID (default anthropic_default).
deferrable — run in deferrable mode (defaults to the operators.default_deferrable config).
poll_interval — seconds between status checks, in both the synchronous and deferrable paths.
timeout — seconds to wait for a terminal status; defaults to 24 hours (the batch SLA).
wait_for_completion — if False, return the batch ID immediately after submission.
fail_on_partial_error — if True, fail the task when any request errored or expired. Defaults to False (succeed and log a warning so successful results are not discarded).

Warning

A task retry re-submits a new batch. Prefer retries=0 on this task. The submitted batch_id is pushed to XCom under key batch_id immediately after submission, so a crashed run never loses track of an in-flight batch.

Example¶

tests/system/anthropic/example_anthropic_batch.py[source]

from airflow.providers.anthropic.operators.batch import AnthropicBatchOperator

run_batch = AnthropicBatchOperator(
    task_id="run_batch",
    conn_id=ANTHROPIC_CONN_ID,
    requests=requests,
    deferrable=True,
)

AnthropicBatchSensor¶

AnthropicBatchSensor waits for an already-submitted batch (by batch_id) to reach a terminal status. Pair it with AnthropicBatchOperator(wait_for_completion=False) for a fire-and-forget submit followed by a re-entrant await — because the sensor only polls an existing batch, retrying it never re-submits, which sidesteps the “retry creates a new batch” hazard of a waiting submit task.

It applies the same terminal-status policy as the operator (skip on full cancellation, fail_on_partial_error to fail on errored/expired requests) and supports deferrable mode via the shared trigger.

from airflow.providers.anthropic.operators.batch import AnthropicBatchOperator
from airflow.providers.anthropic.sensors.batch import AnthropicBatchSensor

submit = AnthropicBatchOperator(
    task_id="submit",
    requests=requests,
    wait_for_completion=False,  # fire-and-forget; recommend retries=0
)
wait = AnthropicBatchSensor(
    task_id="wait",
    batch_id="{{ ti.xcom_pull(task_ids='submit') }}",
    deferrable=True,
)
submit >> wait

AnthropicAgentSessionOperator¶

AnthropicAgentSessionOperator runs a Managed Agents session: Anthropic runs the agent loop server-side while the worker drives a session and waits for it to finish. Unlike the common.ai provider (a local pydantic-ai loop), the loop and its tool-execution sandbox run on Anthropic’s infrastructure; the worker only orchestrates.

Agents and environments are created once (via create_agent() / create_environment(), the ant CLI, or the Console) and referenced by ID on every run — the operator never creates an agent per task. Configure the agent for autonomous operation (no client-side custom tools or always_ask permission) so the session reaches idle (turn complete) rather than blocking on input the operator cannot supply.

Provide exactly one of message (a single user turn) or outcome (a user.define_outcome rubric the agent iterates against until satisfied). The operator returns the session ID only; pull artifacts the agent wrote to /mnt/session/outputs/ afterwards via the Files API (scope_id=<session_id>).

Parameters¶

agent_id / environment_id — IDs of a pre-created agent and environment.
message — a single user message to start the session (mutually exclusive with outcome).
outcome — a user.define_outcome payload (description + required rubric, optional max_iterations); mutually exclusive with message.
conn_id — the Anthropic connection ID (default anthropic_default).
deferrable — run in deferrable mode (defaults to operators.default_deferrable).
poll_interval — seconds between session status checks.
timeout — seconds to wait for a terminal status; defaults to 24 hours.
vault_ids — vault IDs providing MCP/credential access to the session.
session_resources — files, GitHub repos, or memory stores to mount (forwarded to sessions.create as resources; renamed to avoid the reserved BaseOperator.resources).
session_kwargs — extra keyword arguments forwarded to sessions.create.

Note

Completion is detected accurately for both modes. A message run inspects the terminal session.status_idle event’s stop_reason (correlated against the kickoff event): end_turn succeeds; requires_action and retries_exhausted raise an error. An outcome run is judged from the outcome_evaluations verdict. The agent must still be configured for autonomous operation (no client-side custom tools / always_ask).

tests/system/anthropic/example_anthropic_agent.py[source]

from airflow.providers.anthropic.operators.agent import AnthropicAgentSessionOperator

run_agent = AnthropicAgentSessionOperator(
    task_id="run_agent",
    conn_id=ANTHROPIC_CONN_ID,
    agent_id=setup["agent_id"],
    environment_id=setup["environment_id"],
    message="Summarize the latest stable Apache Airflow release in two sentences.",
    deferrable=True,
)