• Home
• airflow.providers.anthropic
• airflow.providers.anthropic.operators
• airflow.providers.anthropic.operators.agent

↑↓ Navigate ⏎ Select Esc Close

airflow.providers.anthropic.operators.agent

Classes

AnthropicAgentSessionOperator

Run a Managed Agents session against a pre-created agent and environment.

Module Contents

class airflow.providers.anthropic.operators.agent.AnthropicAgentSessionOperator(*, agent_id, environment_id, message=None, outcome=None, conn_id=AnthropicHook.default_conn_name, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), poll_interval=30, timeout=24 * 60 * 60, budget=None, vault_ids=None, session_resources=None, session_kwargs=None, **kwargs)

Bases: airflow.providers.common.compat.sdk.BaseOperator

Run a Managed Agents session against a pre-created agent and environment.

Anthropic runs the agent loop server-side; the worker creates a session, sends the initial instruction, and waits for the session to reach a terminal status. In deferrable mode it releases the worker slot while a trigger polls the session status.

Provide exactly one of message (a single user turn) or outcome (a user.define_outcome rubric that the agent iterates against until satisfied).

Important

This operator is for autonomous agents — configure the agent without client-side custom tools or an always_ask permission policy.

Completion is detected accurately for both modes. A message run reads the terminal session.status_idle event’s stop_reason (correlated against the kickoff event to avoid a start-race false positive): end_turn succeeds, while requires_action (the agent is blocked on input), retries_exhausted and budget_reached raise an error rather than silently passing. An outcome run is judged from the session’s outcome_evaluations verdict (satisfied vs. failed/max_iterations_reached/interrupted).

Agents and environments are created once (see create_agent()), not per task run.

Outputs the agent writes to /mnt/session/outputs/ are retrieved afterwards via the Files API (scope_id=<session_id>); the operator returns the session ID only.

The session’s token counts and list cost are pushed to XCom under usage on both success and failure, so cost per Dag run can be queried. This matters because a budget is a stop trigger rather than a cap: the ceiling is checked between model requests, so it does not tell you what the session actually spent.

See also

For more information, take a look at the guide: AnthropicAgentSessionOperator

Parameters:

• agent_id (str) – ID of a pre-created agent.

• environment_id (str) – ID of a pre-created environment.

• message (str | None) – A single user message to start the session. Mutually exclusive with outcome.

• outcome (dict[str, Any] | None) – A user.define_outcome payload (description, rubric, optional max_iterations). Mutually exclusive with message.

• conn_id (str) – The Anthropic connection ID to use.

• deferrable (bool) – Run the operator in deferrable mode.

• poll_interval (float) – Seconds between session status checks (both paths).

• timeout (float) – Seconds to wait for a terminal status. Defaults to 24 hours. In deferrable mode the trigger enforces this and tears the session down on timeout; set execution_timeout only for a shorter hard cap (which preempts that graceful teardown).

• vault_ids (list[str] | None) – Vault IDs providing MCP/credential access to the session.

• session_resources (list[dict[str, Any]] | None) – Session resources (files, GitHub repos, memory stores). Named session_resources to avoid colliding with the reserved BaseOperator.resources; forwarded to sessions.create as resources.

• budget (airflow.providers.anthropic.hooks.anthropic.BudgetSpec | None) –

  Spend ceiling for the session. A number or numeric string is read as US dollars (budget=25 is $25.00); a mapping is passed through as the raw API payload. On a message run, a session that stops against it raises AnthropicSessionBudgetExceeded. On an outcome run completion is judged from outcome_evaluations before the idle event is consulted, so a budget stop surfaces as the outcome verdict and the generic AnthropicAgentSessionError instead.

  Warning

  The ceiling is a stop trigger, not a cap. It is checked between model requests, so a request already in flight runs to completion and the session can finish well above the limit. Airflow retries also each start a new session with a fresh budget, so prefer retries=0 when a budget is set.

• session_kwargs (dict[str, Any] | None) – Extra keyword arguments forwarded to sessions.create. Setting budget here as well as via the budget argument is rejected.

template_fields: collections.abc.Sequence[str] = ('agent_id', 'environment_id', 'message', 'outcome', 'budget')

agent_id

environment_id

message = None

outcome = None

conn_id = 'anthropic_default'

deferrable

poll_interval = 30

timeout = 86400

budget = None

vault_ids = None

session_resources = None

session_kwargs

session_id: str | None = None

property hook: airflow.providers.anthropic.hooks.anthropic.AnthropicHook

Return an instance of the AnthropicHook.

execute(context)

Derive when creating an operator.

The main method to execute the task. Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

execute_complete(context, event=None)

on_kill()

Archive the session if the (non-deferred) task is killed.

Only fires while the worker process is alive, i.e. the synchronous path (deferrable=False). On Airflow 3.3+ a killed deferred task is archived by the trigger’s on_kill. On older Airflow the session of a killed deferred task is not archived automatically; archive it manually via the hook.

The supervisor escalates to SIGKILL a few seconds after asking the task to stop, so the hook’s default retry budget would never finish here – the process dies inside the first sleep and the session is never released. This budget gives the interrupt a beat to land (the status change is not instant) while still fitting inside that window: two attempts, one second apart.

Previous Next

Was this entry helpful?