airflow.providers.common.ai.tools.claude_agent_sdk

Give Airflow tools to an agent loop run by the Claude Agent SDK.

Note

Experimental; see Stable and experimental features.

Classes

AirflowTools

Airflow's tools for an agent loop run by the Claude Agent SDK.

Module Contents

class airflow.providers.common.ai.tools.claude_agent_sdk.AirflowTools(*sources, server_name='airflow')[source]

Airflow’s tools for an agent loop run by the Claude Agent SDK.

Pass toolsets that implement ToolProvider, such as SQLToolset and HookToolset, or individual AirflowTool objects. They are served through an in-process MCP server; add it and allowed_tools to ClaudeAgentOptions, or build them with build_options(), and drive the SDK’s query() with run():

from claude_agent_sdk import query

from airflow.providers.common.ai.tools.claude_agent_sdk import AirflowTools
from airflow.providers.common.ai.toolsets.sql import SQLToolset

tools = AirflowTools(SQLToolset(db_conn_id="warehouse", allowed_tables=["orders"]))
options = tools.build_options(model="claude-sonnet-5", max_turns=10)
result = await tools.run(query(prompt="...", options=options))

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 an is_error tool result. A failure the model cannot fix, such as a hook raising, makes run() raise ToolCallError, so the task fails and Airflow retries it. The SDK’s own tool dispatch catches every exception a handler raises and turns it into an is_error result the model reads, so drive the query with run() rather than iterating it yourself.

An instance runs one query at a time: build a new AirflowTools per task, or call run() again once the previous run has finished. A tool call that arrives after a run has ended is refused until the next run() starts. A call that lands after the next run has started cannot be told apart and is served by that run.

Parameters:
server_name = 'airflow'[source]
allowed_tools: list[str][source]
server: claude_agent_sdk.McpSdkServerConfig[source]
build_options(**kwargs)[source]

Build ClaudeAgentOptions wired to this instance’s tools, with safe defaults.

ClaudeAgentOptions() on its own loads every settings source on the worker (~/.claude, project settings, a project .mcp.json) and leaves the CLI’s built-in tools on, including a worker-host Bash. This method instead defaults tools to [] (no built-in tools), setting_sources to [] (no settings files) and strict_mcp_config to True (only the MCP servers passed here), and verbatim_prompts to True. It adds server and allowed_tools to whatever mcp_servers and allowed_tools the caller passes. A keyword this method does not default, such as model or env, is passed through unchanged.

verbatim_prompts stops the CLI from expanding @path file mentions in a prompt; otherwise anyone who controls the prompt could send a file the worker user can read to the model, without passing the secret masker. It also turns off slash command dispatch and the CLI’s turn-start attachment pass.

When can_use_tool is passed, the Airflow tools are not added to allowed_tools, because an allowed tool never reaches the callback, and permission_mode defaults to "default". The callback then decides on every Airflow tool call; the names are still in allowed_tools if you want to allow some by hand.

Everything it builds can also be assembled by hand; see server and allowed_tools.

Raises:

ValueError – when mcp_servers already has a server named server_name.

async run(messages)[source]

Drive messages, the stream query() or ClaudeSDKClient.receive_response() returns.

ClaudeSDKClient.receive_messages() is not supported: it never ends.

messages is closed when this returns or raises. For query() that ends the CLI session, but only once the SDK finishes tearing down, so a tool call can still arrive after this returns or raises; it is answered with an is_error “Not run” result. A ClaudeSDKClient stays connected until you disconnect it, for example by leaving its async with block.

Raises:

  • RuntimeError – when this instance is already running a query, when messages ends without a ResultMessage, or, for ClaudeSDKClient.receive_response(), when the result message has is_error set (for example max_turns reached or an API failure). For a streamed prompt, any result message with is_error set raises, even when a later turn succeeds.

  • ResultError – for query() when the result message has is_error set. The CLI exits non-zero right after such a result, so the SDK raises claude_agent_sdk.ResultError (a ProcessError) from the stream before this method checks the result.

  • ToolCallError – when a tool failed in a way the model cannot fix.

Was this entry helpful?