• Home
• airflow.providers.anthropic
• airflow.providers.anthropic.hooks
• airflow.providers.anthropic.hooks.anthropic

↑↓ Navigate ⏎ Select Esc Close

airflow.providers.anthropic.hooks.anthropic

Attributes

logger
DEFAULT_MODEL
FIRST_PARTY_PLATFORMS
AnthropicClient
MAX_CONSECUTIVE_POLL_FAILURES
OUTCOME_FAILURE_RESULTS
BUDGET_REACHED
BudgetSpec
TRIGGER_EVENT_STATUSES

Classes

BatchStatus	Top-level processing_status of an Anthropic Message Batch.
SessionStatus	Status of a Managed Agents session.
SessionPollResult	Verdict from one poll of a session; see AnthropicHook.poll_session_completion().
AnthropicHook	Use the Anthropic SDK to interact with the Claude API.

Functions

build_budget(budget)	Normalize a session budget into the API's max_list_cost payload.
evaluate_session_state(session, *, expect_outcome)	Judge a polled session from its object fields alone.
validate_execute_complete_event([event])	Validate the event a deferred task resumes with, returning it if well-formed.
evaluate_batch_counts(*, batch_id, canceled, errored, ...)	Apply the success/skip/fail policy for a terminal batch's request counts.

Module Contents

airflow.providers.anthropic.hooks.anthropic.logger

airflow.providers.anthropic.hooks.anthropic.DEFAULT_MODEL = 'claude-opus-4-8'

airflow.providers.anthropic.hooks.anthropic.FIRST_PARTY_PLATFORMS

airflow.providers.anthropic.hooks.anthropic.AnthropicClient

airflow.providers.anthropic.hooks.anthropic.MAX_CONSECUTIVE_POLL_FAILURES = 5

class airflow.providers.anthropic.hooks.anthropic.BatchStatus

Bases: str, enum.Enum

Top-level processing_status of an Anthropic Message Batch.

IN_PROGRESS = 'in_progress'

CANCELING = 'canceling'

ENDED = 'ended'

classmethod is_in_progress(status)

Return True while the batch has not reached the terminal ended status.

This is broader than the in_progress value: a canceling batch is also non-terminal (cancellation is in flight but the batch has not ended yet), so it returns True too. Read the name as “not yet terminal”, not “equals the in_progress status”.

class airflow.providers.anthropic.hooks.anthropic.SessionStatus

Bases: str, enum.Enum

Status of a Managed Agents session.

RESCHEDULING = 'rescheduling'

RUNNING = 'running'

IDLE = 'idle'

TERMINATED = 'terminated'

classmethod is_terminal(status)

Return True once the session has stopped working.

idle means the agent finished its turn (done, for an autonomous run); terminated is an unrecoverable failure. Both stop the wait.

airflow.providers.anthropic.hooks.anthropic.OUTCOME_FAILURE_RESULTS

airflow.providers.anthropic.hooks.anthropic.BUDGET_REACHED = 'budget_reached'

airflow.providers.anthropic.hooks.anthropic.BudgetSpec

airflow.providers.anthropic.hooks.anthropic.build_budget(budget)

Normalize a session budget into the API’s max_list_cost payload.

A scalar is read as US dollars (25, 25.0 and "25.00" all mean $25.00). The API wants minor units as an integer decimal string, so the conversion runs through Decimal (never binary float) and rejects an amount finer than a cent rather than silently rounding money. A mapping is deep-copied and otherwise returned unchanged, so a raw payload the provider has not caught up with stays usable without a provider release.

Warning

The ceiling is a stop trigger, not a cap: it is checked between model requests, so a request already in flight can carry the session well past it.

class airflow.providers.anthropic.hooks.anthropic.SessionPollResult

Bases: NamedTuple

Verdict from one poll of a session; see AnthropicHook.poll_session_completion().

Named rather than a bare tuple because error_message and stop_reason are both str | None, so transposing them at a call site would still type-check.

done: bool

error_message: str | None

stop_reason: str | None

airflow.providers.anthropic.hooks.anthropic.evaluate_session_state(session, *, expect_outcome)

Judge a polled session from its object fields alone.

Returns (done, error_message, needs_event_check). done=False means keep polling. needs_event_check=True means the session is idle on a message run and the object can’t say why — the caller must inspect the event log (see AnthropicHook.poll_session_completion()).

The status field can’t distinguish a genuine end_turn from requires_action or retries_exhausted, nor a just-created idle. For an outcome run the true verdict is in outcome_evaluations (judged here, which also defeats the start race).

airflow.providers.anthropic.hooks.anthropic.TRIGGER_EVENT_STATUSES

airflow.providers.anthropic.hooks.anthropic.validate_execute_complete_event(event=None)

Validate the event a deferred task resumes with, returning it if well-formed.

The event crosses the triggerer/worker boundary through the metadata DB, so a resuming task can receive None or a status its handlers do not recognize (version skew, a custom trigger). Both must fail loudly: the execute_complete handlers raise on timeout/error and treat everything else as success, so an unrecognized status would otherwise silently succeed.

airflow.providers.anthropic.hooks.anthropic.evaluate_batch_counts(*, batch_id, canceled, errored, expired, succeeded, fail_on_partial_error)

Apply the success/skip/fail policy for a terminal batch’s request counts.

Lives in the hook module so both AnthropicBatchOperator and AnthropicBatchSensor share it without an operator/sensor cross-import. Raises AirflowSkipException for a fully-cancelled batch, AnthropicBatchJobError when fail_on_partial_error and any request failed, otherwise returns (logging a warning for partial failures).

class airflow.providers.anthropic.hooks.anthropic.AnthropicHook(conn_id=default_conn_name, *args, **kwargs)

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

Use the Anthropic SDK to interact with the Claude API.

The connection’s password is used as the API key and host as an optional base URL (for gateways/proxies). The extra field selects the platform client and passes platform-specific configuration:

• platform: one of anthropic (default), bedrock, vertex, aws, foundry.

• model: default model id used when an operator/hook call omits model (lets you change the model without editing Dags); falls back to DEFAULT_MODEL.

• aws_region: region for the bedrock and aws platforms.

• project_id / region: project and region for the vertex platform.

• resource: Azure resource name for the foundry platform.

• anthropic_client_kwargs: extra keyword arguments forwarded to the client constructor (e.g. timeout, max_retries, default_headers).

• workload_identity: configure Workload Identity Federation (keyless OIDC auth) with identity_token_file, federation_rule_id, organization_id, service_account_id and optional workspace_id / scope.

When the anthropic platform has no API Key and no workload_identity block, the client is built with no static credential so the SDK resolves them from the environment — supporting env-driven Workload Identity Federation and ant profiles.

See also

https://docs.claude.com/en/api/client-sdks

Parameters:

conn_id (str) – Anthropic connection id.

conn_name_attr = 'conn_id'

default_conn_name = 'anthropic_default'

conn_type = 'anthropic'

hook_name = 'Anthropic'

conn_id = 'anthropic_default'

property platform: str

Return the configured platform (defaults to anthropic).

property default_model: str

Default model id — connection extra['model'] if set, else DEFAULT_MODEL.

property conn: AnthropicClient

Return the Anthropic client for the configured platform.

get_conn()

Build and return the Anthropic client for the configured platform.

test_connection()

Test the Anthropic connection.

create_message(messages, model=None, max_tokens=1024, system=None, **kwargs)

Create a single message response (one-shot messages.create).

Parameters:

• messages (list[dict[str, Any]]) – The conversation so far, as a list of message dicts.

• model (str | None) – Model ID to use. Defaults to default_model (the connection’s extra['model'] or DEFAULT_MODEL).

• max_tokens (int) – Maximum number of tokens to generate.

• system (str | None) – Optional system prompt.

count_tokens(messages, model=None, system=None, **kwargs)

Return the number of input tokens the given request would consume.

create_batch(requests, model=None)

Submit a Message Batch.

Parameters:

• requests (list[dict[str, Any]]) – A list of {"custom_id": str, "params": {...}} dicts, where params is a messages.create payload (model, max_tokens, messages, …). A request that omits model inherits model below, or the connection’s default_model (extra['model']) when that is unset too.

• model (str | None) – Default model id for requests that do not set their own. Falls back to the connection’s default_model.

get_batch(batch_id)

Retrieve a Message Batch by ID.

cancel_batch(batch_id)

Request cancellation of a Message Batch.

list_batches(**kwargs)

Return a (paginated) list of Message Batches.

stream_batch_results(batch_id)

Return a streaming iterator of per-request results, keyed by custom_id.

Results stream from the API and arrive in arbitrary order — key them by result.custom_id, never by position. Results are available for 29 days after the batch is created. The result set can be very large: iterate and persist to object storage; do not materialize it into XCom.

wait_for_batch(batch_id, wait_seconds=3, timeout=24 * 60 * 60)

Poll a batch synchronously until it reaches the terminal ended status.

Parameters:

• batch_id (str) – The batch to wait for.

• wait_seconds (float) – Seconds to sleep between polls.

• timeout (float) – Maximum seconds to wait before raising AnthropicBatchTimeout.

Returns:

The terminal MessageBatch.

Return type:

anthropic.types.messages.MessageBatch

create_agent(name, model=None, **kwargs)

Create a (reusable, versioned) Managed Agents agent. One-time setup.

model defaults to default_model (the connection’s extra['model'] or DEFAULT_MODEL). Pass a mapping instead of a bare id to set the model config, e.g. {"id": "claude-opus-5", "inference_geo": "us"}.

create_environment(name, config=None, **kwargs)

Create a (reusable) environment for agent sessions. One-time setup.

create_session(agent, environment_id, **kwargs)

Start a session against a pre-created agent + environment.

A budget keyword accepts an amount in USD as well as the raw API payload; see build_budget().

update_session(session_id, **kwargs)

Update a live session – raise or clear its budget, swap agent tools, retitle.

Only the keywords you pass are sent, which matters because the API distinguishes omitted (preserve) from None (clear). So update_session(sid) changes nothing, while update_session(sid, budget=None) removes the ceiling – the escape hatch for a session stopped by a model with no list price, which raising the budget cannot unblock.

budget accepts an amount in USD or the raw payload (see build_budget()). agent={"tools": [...]} is a full replacement of the tool list, not a merge, and needs the mid-conversation-tool-changes-2026-07-01 beta.

get_session(session_id)

Retrieve a session (carries its current status).

get_session_usage(session_id)

Return a JSON-serializable token/cost summary for a session.

Plain scalars and a nested list_cost mapping rather than SDK models, so the result survives XCom serialization and can be queried across runs. amount is kept as the API’s minor-unit string ("44" is $0.44) rather than converted to a float, so no rounding is applied to a cost figure.

Every field is optional server-side – list_cost is absent when usage includes a model with no list price – so missing values come back as None. That absence is why every billable dimension is reported and not just the token totals: it is exactly when a caller has to reconstruct cost from usage that the breakdown must be complete. Cache writes (cache_creation) are billed above base input, and server tool calls are billed per request.

static summarize_usage(session)

Flatten an already-retrieved session’s usage; see get_session_usage().

Split out because sessions.archive also returns the session, so a caller that is tearing a session down can report its usage without a second request.

Dumps the model rather than copying a fixed list of fields. The usage model sets extra="allow", so a billable dimension added by the API is kept on the object – an allowlist here would drop it silently, which is worst precisely when list_cost is None and a caller has to price the run from the breakdown. mode="json" keeps the result XCom-safe and leaves amount a minor-unit string.

send_event(session_id, event)

Send a single event (e.g. a user.message or user.define_outcome).

interrupt_session(session_id)

Send user.interrupt to pause a running session.

The API refuses to archive or delete a session while it is running, so this is the only way to release one that is not going to stop on its own – see archive_session().

archive_session(session_id, *, attempts=6, wait_seconds=5)

Archive a session (frees the server-side container). Best-effort teardown.

Returns the archived session, which carries its final usage – so a caller tearing a session down does not need a separate retrieve to report what it spent.

A running session cannot be archived (nor deleted): the API rejects both with a 400. Only then does this interrupt the session and retry, because a session that will not stop on its own otherwise accrues billable runtime with no way to release it. Any other failure is re-raised untouched, so a transient 5xx does not send user.interrupt to a session that was working fine.

Retrying costs up to attempts further calls with wait_seconds between them (about 25s at the defaults), which is longer than some callers have: a killed task’s on_kill is SIGKILLed a few seconds in, so it passes a much tighter budget.

poll_session_completion(session_id, *, expect_outcome=False, kickoff_event_id=None)

Return the SessionPollResult for one poll of a session.

Combines the session object (status / outcome verdict) with the event log (stop_reason of the latest idle) so a message run distinguishes genuine end_turn completion from requires_action / retries_exhausted / budget_reached.

stop_reason is the SDK’s own idle stop reason, or None when the verdict did not come from an idle event (a terminated session, or an outcome verdict). It exists so callers can pick an error class without matching on the message text; pass it to _create_session_error().

Note

A budget stop is classified on message runs only. An outcome run is judged from outcome_evaluations before the event log is consulted, so a budget stop there surfaces as whatever verdict the outcome recorded.

wait_for_session(session_id, expect_outcome=False, kickoff_event_id=None, poll_interval=30, timeout=24 * 60 * 60)

Poll a session synchronously until it completes.

Parameters:

• session_id (str) – The session to wait for.

• expect_outcome (bool) – Whether the session is running a user.define_outcome loop (completion judged from outcome_evaluations).

• kickoff_event_id (str | None) – ID of the kickoff event, used to correlate the terminal idle event on a message run (defeats the start race).

• poll_interval (float) – Seconds to sleep between polls.

• timeout (float) – Maximum seconds to wait before raising AnthropicAgentSessionTimeout.

Raises:

AnthropicSessionBudgetExceeded – If the session stopped against its budget.

Previous Next

Was this entry helpful?