airflow.providers.common.ai.tools.anthropic

Give Airflow tools to an agent loop run by the Anthropic SDK’s tool runner.

Note

Experimental; see Stable and experimental features.

Classes

AirflowTools

Airflow's tools for the tool runner of a synchronous Anthropic client.

AsyncAirflowTools

Airflow's tools for the tool runner of an AsyncAnthropic client.

Module Contents

class airflow.providers.common.ai.tools.anthropic.AirflowTools(*sources)[source]

Airflow’s tools for the tool runner of a synchronous Anthropic client.

Pass toolsets that implement ToolProvider, such as SQLToolset and HookToolset, or individual AirflowTool objects. Give tools to client.beta.messages.tool_runner and drive the runner with run():

from anthropic import Anthropic

from airflow.providers.common.ai.tools.anthropic import AirflowTools
from airflow.providers.common.ai.toolsets import SQLToolset

client = Anthropic()
tools = AirflowTools(SQLToolset("warehouse", allowed_tables=["orders"]))
runner = client.beta.messages.tool_runner(
    model="claude-opus-5-5",
    max_tokens=16000,
    tools=tools.tools,
    messages=[...],
)
message = tools.run(runner)

Each tool keeps the source tool’s name, description and argument schema, and every result passes through Airflow’s secret masker first. A failure the model can correct, such as a query naming a missing column, reaches it as a tool_result with is_error: true, counted against the tool’s retry limit once per model turn. A tool that is still failing once the limit is used up, or a failure the model cannot fix, such as a hook raising, makes run() raise ToolCallError before the next model request, so the task fails and Airflow retries it. The runner on its own hands every exception back to the model, so drive it with run() rather than until_done(). A runner that reaches its max_iterations while the model is still calling tools makes run() raise ToolRunnerMaxIterationsError, since its last message holds no answer. A runner created with stream=True is not supported.

For an AsyncAnthropic client, use AsyncAirflowTools.

Parameters:

sources (airflow.providers.common.ai.tools.ToolProvider | airflow.providers.common.ai.tools.AirflowTool) – Toolsets and tools to add, in order.

tools: list[anthropic.lib.tools.BetaFunctionTool[collections.abc.Callable[..., str]]][source]
run(runner)[source]

Drive runner to its final message.

Raises:
class airflow.providers.common.ai.tools.anthropic.AsyncAirflowTools(*sources)[source]

Airflow’s tools for the tool runner of an AsyncAnthropic client.

The asynchronous counterpart of AirflowTools, with the same behaviour:

tools = AsyncAirflowTools(SQLToolset("warehouse", allowed_tables=["orders"]))
runner = AsyncAnthropic().beta.messages.tool_runner(
    model="claude-opus-5-5",
    max_tokens=16000,
    tools=tools.tools,
    messages=[...],
)
message = await tools.run(runner)
Parameters:

sources (airflow.providers.common.ai.tools.ToolProvider | airflow.providers.common.ai.tools.AirflowTool) – Toolsets and tools to add, in order.

tools: list[anthropic.lib.tools.BetaAsyncFunctionTool[collections.abc.Callable[..., collections.abc.Coroutine[Any, Any, str]]]][source]
async run(runner)[source]

Drive runner to its final message.

Raises:

Was this entry helpful?