You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kaxil (via GitHub)" <gi...@apache.org> on 2026/09/30 15:11:53 UTC
[PR] Add Anthropic tool runner adapter for common.ai toolsets [airflow]
kaxil opened a new pull request, #73962:
URL: https://github.com/apache/airflow/pull/73962
Agents whose loop is the Anthropic SDK's `client.beta.messages.tool_runner` can now use common.ai's connection-backed toolsets (`SQLToolset`, `HookToolset`, `ObjectStorageToolset`, sandboxes) without being rebuilt as an `AgentOperator` or a Pydantic AI agent:
```python
tools = AirflowTools(SQLToolset(db_conn_id="warehouse"))
runner = client.beta.messages.tool_runner(model=..., max_tokens=16000, tools=tools.tools, messages=[...])
message = tools.run(runner)
```
`AirflowTools` (for `Anthropic`) and `AsyncAirflowTools` (for `AsyncAnthropic`) sit next to the Strands and ADK adapters from #73898 and are built on the same framework-neutral `AirflowTool` interface. They make the same promises:
- Every result goes through the secret masker.
- A failure the model can correct reaches it as a `tool_result` with `is_error: true`, counted against the tool's retry limit once per model turn.
- Any other failure fails the task.
- Calls are counted under `common_ai.tool_calls` with `framework=anthropic`.
The adapters are experimental. There's a guide under Frameworks and a row on the stability page.
Checked end to end against Claude through a gateway connection with four runs:
- the example Dag;
- a registered connection password inside a query result, which reached the model as `***`;
- a query on a table outside `allowed_tables`, which came back as `is_error` and which the model recovered from;
- a hook that raises, which failed the task after exactly one model request.
## Design rationale
**Why `tools.run(runner)` and not the runner's own `until_done()`.** The runner catches every exception a tool raises and sends it to the model as an error result. On its own, a hook raising or a connection's credentials being rejected becomes one more model turn, and the task can succeed on an answer written around the failure. The SDK has no hook to stop that. `run()` drives the loop and calls the runner's public `generate_tool_call_response()` inside a per-turn scope; the runner caches that response and sends it with the next request. That way a failure is raised before the next request, and the retry limit knows which calls share a turn. Driving the runner without `run()` still works, but a failure then goes to the model, and the adapter logs a warning when that happens.
**Tools run only on `tool_use` turns.** The runner itself runs a turn's tool calls only when the turn ended in `tool_use`. A turn cut off at `max_tokens` can hold a tool call with incomplete arguments, and `pause_turn` and `compaction` turns are resumed without running any tools. `run()` follows the same rule. The SDK's runner gained that routing in 1.1.0 (1.0.0 ran tools on every stop reason except `refusal`), so the floor of common.ai's `anthropic` extra moves from 1.0.0 to 1.1.0.
**After one call in a turn fails, the turn's other calls are skipped.** The runner calls them one at a time, and their results would never be sent, so a tool that writes would run for nothing.
**Why common.ai and not the Anthropic provider.** The adapter adapts common.ai's tool interface, which is experimental and 0.x, and it uses common.ai internals: the per-turn retry scope and the metric's framework label. common.ai's `anthropic` extra already carries the SDK for `@task.llm_batch`. The adapter accepts any `Anthropic` or `AsyncAnthropic` client, so it does not depend on the Anthropic provider. The guide points to `AnthropicHook(...).get_conn()` for Bedrock, Vertex AI and Foundry.
**`run_coroutine_sync` now copies the caller's context into its worker thread.** When a loop is already running (an `async def` task, a notebook), the shared helper runs the coroutine on a worker thread. Until now that thread did not see the caller's context variables, so the sync adapter lost track of the run and a fatal failure went to the model. Copying the context fixes this for every caller of the helper. The helper also now calls `asyncio.run` outside its `except RuntimeError` block, so a tool's exception no longer arrives in the task log chained to a "no running event loop" error.
## Gotchas
- `tool_runner` is a beta helper of the Anthropic SDK, which is one reason the adapter is experimental.
- The streaming runner (`tool_runner(..., stream=True)`) is not supported.
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Anthropic tool runner adapter for common.ai toolsets [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #73962:
URL: https://github.com/apache/airflow/pull/73962#issuecomment-6160204647
<!-- uv-lock-rebase-notice -->
`uv.lock` on `main` just moved via [#71725](https://redirect.github.com/apache/airflow/pull/71725) ("Add Boat backend for sandbox tools"), commit [`c5c2f19`](https://redirect.github.com/apache/airflow/commit/c5c2f194b574df45125b286c99daace57986333f) and this PR currently conflicts.
Quickest fix:
```bash
git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease
```
_Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future `uv.lock` bumps._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Anthropic tool runner adapter for common.ai toolsets [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #73962:
URL: https://github.com/apache/airflow/pull/73962
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Anthropic tool runner adapter for common.ai toolsets [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #73962:
URL: https://github.com/apache/airflow/pull/73962#discussion_r4267952853
##########
providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py:
##########
@@ -0,0 +1,264 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Give Airflow tools to an agent loop run by the Anthropic SDK's tool runner.
+
+.. note:: Experimental; see :mod:`airflow.providers.common.ai.tools`.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+from collections.abc import Callable, Coroutine, Iterator
+from contextlib import AbstractContextManager, contextmanager
+from contextvars import ContextVar
+from typing import TYPE_CHECKING, Any, TypeVar
+
+try:
+ from anthropic.lib.tools import (
+ BetaAsyncFunctionTool,
+ BetaAsyncToolRunner,
+ BetaFunctionTool,
+ BetaToolRunner,
+ ToolError,
+ beta_async_tool,
+ beta_tool,
+ )
+except ImportError as e:
+ from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(e)
+
+from airflow.providers.common.ai.tools import AirflowTool, ToolCallError, collect_tools
+from airflow.providers.common.ai.tools._from_toolset import tool_call_scope
+from airflow.providers.common.ai.utils.coroutines import run_coroutine_sync
+from airflow.providers.common.ai.utils.tool_definition import serialize_for_llm
+from airflow.providers.common.ai.utils.tool_metrics import calling_framework
+
+if TYPE_CHECKING:
+ from anthropic.types.beta import BetaMessage
+ from anthropic.types.beta.parsed_beta_message import ParsedBetaMessage
+
+ from airflow.providers.common.ai.tools import ToolProvider
+
+log = logging.getLogger(__name__)
+
+__all__ = ["AirflowTools", "AsyncAirflowTools"]
+
+# The ``output_format`` a runner parses its final message into.
+_OutputT = TypeVar("_OutputT")
+
+# The failures of the run in progress. ``run`` sets it, the tools append to it.
+_failures: ContextVar[list[ToolCallError] | None] = ContextVar("common_ai_anthropic_failures", default=None)
+
+
+class AirflowTools:
+ """
+ Airflow's tools for the tool runner of a synchronous ``Anthropic`` client.
+
+ Pass toolsets that implement
+ :class:`~airflow.providers.common.ai.tools.ToolProvider`, such as ``SQLToolset``
+ and ``HookToolset``, or individual
+ :class:`~airflow.providers.common.ai.tools.AirflowTool` objects. Give :attr:`tools`
+ to ``client.beta.messages.tool_runner`` and drive the runner with :meth:`run`:
+
+ .. code-block:: python
+
+ 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 :meth:`run` raise
+ :class:`~airflow.providers.common.ai.tools.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 :meth:`run` rather than
+ ``until_done()``. A runner created with ``stream=True`` is not supported.
+
+ For an ``AsyncAnthropic`` client, use :class:`AsyncAirflowTools`.
+
+ :param sources: Toolsets and tools to add, in order.
+ """
+
+ def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
+ self.tools: list[BetaFunctionTool[Callable[..., str]]] = [
+ _sync_tool(tool) for tool in collect_tools(sources)
+ ]
+
+ def run(self, runner: BetaToolRunner[_OutputT]) -> ParsedBetaMessage[_OutputT]:
+ """
+ Drive ``runner`` to its final message.
+
+ :raises ToolCallError: when a tool failed in a way the model cannot fix.
+ """
+ with _run() as turn:
+ for message in runner:
+ with turn(message.id):
+ if _runs_tools(message):
+ runner.generate_tool_call_response()
+ return runner.until_done()
Review Comment:
If the runner stops at `max_iterations` while the model is still asking for tools, I think this returns that last `tool_use` message. Callers that join the text blocks then get `""`, and the task succeeds with no answer and nothing in the log (the example DAG does exactly this).
Would it make sense to raise here when the final `stop_reason == "tool_use"`? That would cover the async `run()` below and any caller, not just the example. I didn't see a test for the exhaustion case either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add Anthropic tool runner adapter for common.ai toolsets [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73962:
URL: https://github.com/apache/airflow/pull/73962#discussion_r4268704474
##########
providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py:
##########
@@ -0,0 +1,264 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Give Airflow tools to an agent loop run by the Anthropic SDK's tool runner.
+
+.. note:: Experimental; see :mod:`airflow.providers.common.ai.tools`.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+from collections.abc import Callable, Coroutine, Iterator
+from contextlib import AbstractContextManager, contextmanager
+from contextvars import ContextVar
+from typing import TYPE_CHECKING, Any, TypeVar
+
+try:
+ from anthropic.lib.tools import (
+ BetaAsyncFunctionTool,
+ BetaAsyncToolRunner,
+ BetaFunctionTool,
+ BetaToolRunner,
+ ToolError,
+ beta_async_tool,
+ beta_tool,
+ )
+except ImportError as e:
+ from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(e)
+
+from airflow.providers.common.ai.tools import AirflowTool, ToolCallError, collect_tools
+from airflow.providers.common.ai.tools._from_toolset import tool_call_scope
+from airflow.providers.common.ai.utils.coroutines import run_coroutine_sync
+from airflow.providers.common.ai.utils.tool_definition import serialize_for_llm
+from airflow.providers.common.ai.utils.tool_metrics import calling_framework
+
+if TYPE_CHECKING:
+ from anthropic.types.beta import BetaMessage
+ from anthropic.types.beta.parsed_beta_message import ParsedBetaMessage
+
+ from airflow.providers.common.ai.tools import ToolProvider
+
+log = logging.getLogger(__name__)
+
+__all__ = ["AirflowTools", "AsyncAirflowTools"]
+
+# The ``output_format`` a runner parses its final message into.
+_OutputT = TypeVar("_OutputT")
+
+# The failures of the run in progress. ``run`` sets it, the tools append to it.
+_failures: ContextVar[list[ToolCallError] | None] = ContextVar("common_ai_anthropic_failures", default=None)
+
+
+class AirflowTools:
+ """
+ Airflow's tools for the tool runner of a synchronous ``Anthropic`` client.
+
+ Pass toolsets that implement
+ :class:`~airflow.providers.common.ai.tools.ToolProvider`, such as ``SQLToolset``
+ and ``HookToolset``, or individual
+ :class:`~airflow.providers.common.ai.tools.AirflowTool` objects. Give :attr:`tools`
+ to ``client.beta.messages.tool_runner`` and drive the runner with :meth:`run`:
+
+ .. code-block:: python
+
+ 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 :meth:`run` raise
+ :class:`~airflow.providers.common.ai.tools.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 :meth:`run` rather than
+ ``until_done()``. A runner created with ``stream=True`` is not supported.
+
+ For an ``AsyncAnthropic`` client, use :class:`AsyncAirflowTools`.
+
+ :param sources: Toolsets and tools to add, in order.
+ """
+
+ def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
+ self.tools: list[BetaFunctionTool[Callable[..., str]]] = [
+ _sync_tool(tool) for tool in collect_tools(sources)
+ ]
+
+ def run(self, runner: BetaToolRunner[_OutputT]) -> ParsedBetaMessage[_OutputT]:
+ """
+ Drive ``runner`` to its final message.
+
+ :raises ToolCallError: when a tool failed in a way the model cannot fix.
+ """
+ with _run() as turn:
+ for message in runner:
+ with turn(message.id):
+ if _runs_tools(message):
+ runner.generate_tool_call_response()
+ return runner.until_done()
Review Comment:
Good catch, confirmed against the SDK: the runner checks `max_iterations` only at the top of its loop, so a cut-off run's last message is the `tool_use` turn. `run()` (sync and async) now raises `ToolRunnerMaxIterationsError` when the final `stop_reason` is `tool_use`, and there are tests for both a cut-off run and one that answers on its last allowed turn. Fixed in eefc777fdfb.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org