Airflow Summit 2026 is coming August 31 - September 2 in Austin, TX. Register now to secure your spot!

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, vault_ids=None, session_resources=None, session_kwargs=None, **kwargs)[source]

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) and retries_exhausted 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.

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.

  • session_kwargs (dict[str, Any] | None) – Extra keyword arguments forwarded to sessions.create.

template_fields: collections.abc.Sequence[str] = ('agent_id', 'environment_id', 'message', 'outcome')[source]
agent_id[source]
environment_id[source]
message = None[source]
outcome = None[source]
conn_id = 'anthropic_default'[source]
deferrable[source]
poll_interval = 30[source]
timeout = 86400[source]
vault_ids = None[source]
session_resources = None[source]
session_kwargs[source]
session_id: str | None = None[source]
property hook: airflow.providers.anthropic.hooks.anthropic.AnthropicHook[source]

Return an instance of the AnthropicHook.

execute(context)[source]

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)[source]
on_kill()[source]

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.

Was this entry helpful?