This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f82542a453a Name the caller's own MCP client and durable setup in
Common AI errors (#76495)
f82542a453a is described below
commit f82542a453a63d1b5f5b3fd73f99b2cf7ca9dae4
Author: Kaxil Naik <[email protected]>
AuthorDate: Thu Nov 5 14:55:23 2026 +0000
Name the caller's own MCP client and durable setup in Common AI errors
(#76495)
Handing MCPToolset to a framework adapter's AirflowTools told every user
about Strands' MCPClient and ADK's McpToolset, and called the adapters
"the framework-neutral tools". The error is now MCPToolsetNotSupportedError,
still a NotImplementedError, and each adapter this provider ships names its
own MCP route: mcp_servers with MCPServerStreamableHttp for the OpenAI
Agents SDK, async_mcp_tool() with AsyncAnthropic for the Anthropic tool
runner, build_options(mcp_servers=...) plus allowed_tools for the Claude
Agent SDK, MCPClient for Strands and McpToolset for ADK. The public
collect_tools() signature is unchanged; the adapters pass their hint
through a private _collect_tools().
The missing durable_cache_path error on Airflow before 3.3 said
"durable=True requires ...", which run_durably() callers never pass. It now
talks about durable execution, and so does the option's description.
---
providers/common/ai/docs/durable_execution.rst | 2 +-
providers/common/ai/docs/frameworks/anthropic.rst | 4 +-
.../common/ai/docs/frameworks/claude_agent_sdk.rst | 4 +-
.../common/ai/docs/frameworks/openai_agents.rst | 7 ++--
providers/common/ai/docs/troubleshooting.rst | 2 +-
providers/common/ai/provider.yaml | 7 ++--
.../airflow/providers/common/ai/durable/storage.py | 3 +-
.../src/airflow/providers/common/ai/exceptions.py | 10 +++++
.../providers/common/ai/get_provider_info.py | 2 +-
.../airflow/providers/common/ai/tools/__init__.py | 28 +++++++++++++-
.../src/airflow/providers/common/ai/tools/adk.py | 8 +++-
.../airflow/providers/common/ai/tools/anthropic.py | 14 +++++--
.../providers/common/ai/tools/claude_agent_sdk.py | 9 ++++-
.../providers/common/ai/tools/openai_agents.py | 9 ++++-
.../airflow/providers/common/ai/tools/strands.py | 6 ++-
.../airflow/providers/common/ai/toolsets/mcp.py | 14 +++----
.../tests/unit/common/ai/durable/test_storage.py | 9 ++++-
.../ai/tests/unit/common/ai/tools/test_adk.py | 6 +++
.../tests/unit/common/ai/tools/test_anthropic.py | 13 ++++++-
.../unit/common/ai/tools/test_claude_agent_sdk.py | 10 +++++
.../unit/common/ai/tools/test_openai_agents.py | 6 +++
.../ai/tests/unit/common/ai/tools/test_strands.py | 6 +++
.../ai/tests/unit/common/ai/tools/test_tools.py | 43 +++++++++++++++++++++-
.../ai/tests/unit/common/ai/toolsets/test_mcp.py | 8 +++-
24 files changed, 193 insertions(+), 37 deletions(-)
diff --git a/providers/common/ai/docs/durable_execution.rst
b/providers/common/ai/docs/durable_execution.rst
index b6004ed0e69..87bffbbed7a 100644
--- a/providers/common/ai/docs/durable_execution.rst
+++ b/providers/common/ai/docs/durable_execution.rst
@@ -122,7 +122,7 @@ Without the option, the task fails with:
.. code-block:: text
- ValueError: durable=True requires [common.ai] durable_cache_path to be
set. Example: durable_cache_path = file:///tmp/airflow_durable_cache
+ ValueError: Durable execution on Airflow before 3.3 needs [common.ai]
durable_cache_path to be set, to say where its journal is kept. Example:
durable_cache_path = file:///tmp/airflow_durable_cache
How a retry replays
-------------------
diff --git a/providers/common/ai/docs/frameworks/anthropic.rst
b/providers/common/ai/docs/frameworks/anthropic.rst
index 3124206709c..1d83e75d1d3 100644
--- a/providers/common/ai/docs/frameworks/anthropic.rst
+++ b/providers/common/ai/docs/frameworks/anthropic.rst
@@ -84,7 +84,9 @@ What each tool does
``AirflowTools`` accepts any toolset that implements
:class:`~airflow.providers.common.ai.tools.ToolProvider`, and individual
:class:`~airflow.providers.common.ai.tools.AirflowTool` objects.
``MCPToolset`` is the
-exception; give the runner the MCP server through the SDK instead. Each tool:
+exception; run the agent with ``AsyncAnthropic`` and
+:class:`~airflow.providers.common.ai.tools.anthropic.AsyncAirflowTools`, and
convert the MCP
+server's tools with the SDK's ``anthropic.lib.tools.mcp.async_mcp_tool()``
instead. Each tool:
- Keeps the toolset tool's name, description and argument schema.
- Runs through the toolset, so connection resolution, SQL validation,
diff --git a/providers/common/ai/docs/frameworks/claude_agent_sdk.rst
b/providers/common/ai/docs/frameworks/claude_agent_sdk.rst
index 6b27e4f8dbe..54910ca5047 100644
--- a/providers/common/ai/docs/frameworks/claude_agent_sdk.rst
+++ b/providers/common/ai/docs/frameworks/claude_agent_sdk.rst
@@ -94,7 +94,9 @@ What each tool does
``AirflowTools`` accepts any toolset that implements
:class:`~airflow.providers.common.ai.tools.ToolProvider`, and individual
:class:`~airflow.providers.common.ai.tools.AirflowTool` objects.
``MCPToolset`` is the
-exception; give the SDK the MCP server directly instead. Each tool:
+exception; give the SDK the MCP server directly instead, with
+``build_options(mcp_servers=...)``, and list its tools (``mcp__<name>__*``) in
+``allowed_tools`` so the agent may call them. Each tool:
- Keeps the toolset tool's name, description and argument schema, served as an
MCP tool
named ``mcp__<server_name>__<tool name>``.
diff --git a/providers/common/ai/docs/frameworks/openai_agents.rst
b/providers/common/ai/docs/frameworks/openai_agents.rst
index 23f3d6908fa..8b55cdd03c0 100644
--- a/providers/common/ai/docs/frameworks/openai_agents.rst
+++ b/providers/common/ai/docs/frameworks/openai_agents.rst
@@ -468,7 +468,7 @@ classes, such as ``MCPServerStreamableHttp``.
``MCPToolset`` does not work throu
.. code-block:: text
- NotImplementedError: MCPToolset works in Pydantic AI agents and through
the LangChain bridge, not through the framework-neutral tools. Connect your
framework's own MCP client to the server instead, such as Strands' MCPClient or
ADK's McpToolset.
+ airflow.providers.common.ai.exceptions.MCPToolsetNotSupportedError:
MCPToolset works in Pydantic AI agents and through the LangChain bridge, not
through AirflowTools. Pass the server to the agent's mcp_servers with one of
the SDK's MCPServer classes, such as MCPServerStreamableHttp, instead.
Results from ``mcp_servers`` do not pass through Airflow's secret masker, and
under
``run_durably`` their calls run again on every retry.
@@ -542,8 +542,9 @@ Each model call and function tool call is recorded in the
task instance's durabl
journal as it completes (see :doc:`../durable_execution`). On Airflow 3.3 and
later the
journal is kept in the task state store. On earlier versions, set ``[common.ai]
durable_cache_path`` to where it is kept; without it the call raises
``ValueError:
-durable=True requires [common.ai] durable_cache_path to be set.`` Outside a
task there
-is nothing to replay into, and the agent runs as ``Runner.run`` runs it.
+Durable execution on Airflow before 3.3 needs [common.ai] durable_cache_path
to be set``.
+Outside a task there is nothing to replay into, and the agent runs as
``Runner.run`` runs
+it.
What was recorded is deleted when the agent's run succeeds, so clearing the
task later
starts over. That happens when ``run_durably_sync`` returns, not when the task
diff --git a/providers/common/ai/docs/troubleshooting.rst
b/providers/common/ai/docs/troubleshooting.rst
index cc4cb5ecf0a..545318a9392 100644
--- a/providers/common/ai/docs/troubleshooting.rst
+++ b/providers/common/ai/docs/troubleshooting.rst
@@ -164,7 +164,7 @@ Run-time errors
it either, so the task fails without retrying. Return text, numbers,
lists, dicts or a
Pydantic model. See :doc:`durable_execution`.
-``durable=True`` on Airflow below 3.3 fails with a ``ValueError`` about
``durable_cache_path``
+Durable execution on Airflow below 3.3 fails with a ``ValueError`` about
``durable_cache_path``
On Airflow versions older than 3.3 the step cache lives in object storage
and
``[common.ai] durable_cache_path`` must be set. On 3.3 and later the task
state store
is used and the option is ignored. See :doc:`durable_execution`.
diff --git a/providers/common/ai/provider.yaml
b/providers/common/ai/provider.yaml
index 7196a6955d1..eaffa676ab8 100644
--- a/providers/common/ai/provider.yaml
+++ b/providers/common/ai/provider.yaml
@@ -116,15 +116,14 @@ config:
options:
durable_cache_path:
description: |
- ObjectStorage URI used to persist per-step caches when running
- ``AgentOperator`` / ``@task.agent`` with ``durable=True`` on Airflow
- **< 3.3**. Each task execution writes a single JSON file under this
+ ObjectStorage URI used to persist per-step caches for durable
execution
+ on Airflow **< 3.3**. Each task execution writes a single JSON file
under this
path containing its cached model responses and tool results, so that
on retry the agent can replay completed steps instead of re-issuing
LLM calls and tool invocations. The file is deleted on successful
task
completion.
- Required for ``durable=True`` only on Airflow < 3.3. On Airflow >=
3.3
+ Required for durable execution only on Airflow < 3.3. On Airflow >=
3.3
the cache is stored in the AIP-103 task state store and this option
is
ignored. Any scheme supported by ``airflow.sdk.ObjectStoragePath`` is
accepted (``file://``, ``s3://``, ``gs://``, ``azure://``, ...).
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
b/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
index 3950b6f88b0..5a41b788dcb 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/durable/storage.py
@@ -38,7 +38,8 @@ def _get_base_path():
path = conf.get(SECTION, "durable_cache_path", fallback="")
if not path:
raise ValueError(
- "durable=True requires [common.ai] durable_cache_path to be set. "
+ "Durable execution on Airflow before 3.3 needs [common.ai]
durable_cache_path "
+ "to be set, to say where its journal is kept. "
"Example: durable_cache_path = file:///tmp/airflow_durable_cache"
)
return ObjectStoragePath(path)
diff --git a/providers/common/ai/src/airflow/providers/common/ai/exceptions.py
b/providers/common/ai/src/airflow/providers/common/ai/exceptions.py
index 1be907e6998..2386a40e73b 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/exceptions.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/exceptions.py
@@ -33,6 +33,16 @@ class ToolRunnerMaxIterationsError(RuntimeError):
"""
+class MCPToolsetNotSupportedError(NotImplementedError):
+ """
+ Raised when ``MCPToolset`` is handed to a framework adapter's
``AirflowTools``.
+
+ An MCP session belongs to the event loop that opened it, so the framework
connects to the
+ server with its own MCP client instead. The adapters this provider ships
name that client
+ in the message.
+ """
+
+
class ToolApprovalError(RuntimeError):
"""
Raised when a run paused for tool approval cannot resume safely.
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
index ffea5c04250..331957e8030 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
@@ -119,7 +119,7 @@ def get_provider_info():
"description": "Options for the
``apache-airflow-providers-common-ai`` provider.\n",
"options": {
"durable_cache_path": {
- "description": "ObjectStorage URI used to persist
per-step caches when running\n``AgentOperator`` / ``@task.agent`` with
``durable=True`` on Airflow\n**< 3.3**. Each task execution writes a single
JSON file under this\npath containing its cached model responses and tool
results, so that\non retry the agent can replay completed steps instead of
re-issuing\nLLM calls and tool invocations. The file is deleted on successful
task\ncompletion.\n\nRequired for ``durable= [...]
+ "description": "ObjectStorage URI used to persist
per-step caches for durable execution\non Airflow **< 3.3**. Each task
execution writes a single JSON file under this\npath containing its cached
model responses and tool results, so that\non retry the agent can replay
completed steps instead of re-issuing\nLLM calls and tool invocations. The file
is deleted on successful task\ncompletion.\n\nRequired for durable execution
only on Airflow < 3.3. On Airflow >= 3.3\n [...]
"version_added": "0.1.0",
"type": "string",
"example": "file:///tmp/airflow_durable_cache",
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/tools/__init__.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/__init__.py
index 5f116b5569d..a7b6e84e4b9 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/tools/__init__.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/tools/__init__.py
@@ -39,6 +39,7 @@ from collections.abc import Awaitable, Callable, Iterable,
Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Protocol
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.utils.masking import mask_secrets
from airflow.providers.common.ai.utils.tool_metrics import calling_framework,
current_framework
@@ -49,6 +50,12 @@ log = logging.getLogger(__name__)
__all__ = ["AirflowTool", "ToolCallError", "ToolProvider", "ToolResult",
"collect_tools"]
+# The first sentence of the error for an MCPToolset handed to an adapter; the
second says what
+# to use instead.
+_MCP_TOOLSET_NOT_SUPPORTED = (
+ "MCPToolset works in Pydantic AI agents and through the LangChain bridge,
not through AirflowTools."
+)
+
class ToolCallError(Exception):
"""
@@ -148,16 +155,35 @@ def collect_tools(sources: Iterable[ToolProvider |
AirflowTool]) -> list[Airflow
"""
Flatten toolsets and individual tools, as a framework adapter receives
them, into tools.
+ :raises MCPToolsetNotSupportedError: when a source is an ``MCPToolset``.
Connect the
+ framework's own MCP client to the server instead.
:raises ValueError: when two tools share a name. Frameworks route a call
by name, so one
of them would silently never run; give the toolsets different
``tool_prefix`` values,
where they take one.
"""
+ return _collect_tools(sources)
+
+
+def _collect_tools(
+ sources: Iterable[ToolProvider | AirflowTool], *, mcp_hint: str | None =
None
+) -> list[AirflowTool]:
+ """
+ :func:`collect_tools` for the adapters this provider ships.
+
+ :param mcp_hint: How the adapter's framework connects to an MCP server, as
a sentence. It
+ replaces the generic advice in the error raised for an ``MCPToolset``.
+ """
tools: list[AirflowTool] = []
for source in sources:
if isinstance(source, AirflowTool):
tools.append(source)
- else:
+ continue
+ try:
tools.extend(source.airflow_tools())
+ except MCPToolsetNotSupportedError:
+ if mcp_hint is None:
+ raise
+ raise MCPToolsetNotSupportedError(f"{_MCP_TOOLSET_NOT_SUPPORTED}
{mcp_hint}") from None
names = [tool.name for tool in tools]
if duplicates := sorted({name for name in names if names.count(name) > 1}):
raise ValueError(
diff --git a/providers/common/ai/src/airflow/providers/common/ai/tools/adk.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/adk.py
index 5520dd7b060..a876e17195e 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/tools/adk.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/tools/adk.py
@@ -34,7 +34,7 @@ except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
-from airflow.providers.common.ai.tools import AirflowTool, collect_tools
+from airflow.providers.common.ai.tools import AirflowTool, _collect_tools
from airflow.providers.common.ai.tools._from_toolset import tool_call_scope
from airflow.providers.common.ai.utils.tool_metrics import calling_framework
@@ -46,6 +46,8 @@ if TYPE_CHECKING:
__all__ = ["AirflowTools"]
+_MCP_HINT = "Connect ADK's McpToolset to the server instead."
+
class AirflowTools(BaseToolset):
"""
@@ -82,7 +84,9 @@ class AirflowTools(BaseToolset):
def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
super().__init__()
- self._adk_tools: list[BaseTool] = [_AirflowAdkTool(tool) for tool in
collect_tools(sources)]
+ self._adk_tools: list[BaseTool] = [
+ _AirflowAdkTool(tool) for tool in _collect_tools(sources,
mcp_hint=_MCP_HINT)
+ ]
async def get_tools(self, readonly_context: ReadonlyContext | None = None)
-> list[BaseTool]:
return self._adk_tools
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py
index a436441bb6a..e1f79d2d393 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/tools/anthropic.py
@@ -45,7 +45,7 @@ except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
from airflow.providers.common.ai.exceptions import ToolRunnerMaxIterationsError
-from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
collect_tools
+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
@@ -61,6 +61,14 @@ log = logging.getLogger(__name__)
__all__ = ["AirflowTools", "AsyncAirflowTools"]
+_ASYNC_MCP_HINT = "Convert the server's tools with
anthropic.lib.tools.mcp.async_mcp_tool() instead."
+# The SDK's sync mcp_tool() needs an AnyIO worker thread with the session open
on its loop,
+# which a sync task does not have.
+_MCP_HINT = (
+ "Run the agent with AsyncAnthropic and AsyncAirflowTools, and convert the
server's tools "
+ "with anthropic.lib.tools.mcp.async_mcp_tool(), instead."
+)
+
# The ``output_format`` a runner parses its final message into.
_OutputT = TypeVar("_OutputT")
@@ -117,7 +125,7 @@ class AirflowTools:
def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
self.tools: list[BetaFunctionTool[Callable[..., str]]] = [
- _sync_tool(tool) for tool in collect_tools(sources)
+ _sync_tool(tool) for tool in _collect_tools(sources,
mcp_hint=_MCP_HINT)
]
def run(self, runner: BetaToolRunner[_OutputT]) ->
ParsedBetaMessage[_OutputT]:
@@ -158,7 +166,7 @@ class AsyncAirflowTools:
def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
self.tools: list[BetaAsyncFunctionTool[Callable[..., Coroutine[Any,
Any, str]]]] = [
- _async_tool(tool) for tool in collect_tools(sources)
+ _async_tool(tool) for tool in _collect_tools(sources,
mcp_hint=_ASYNC_MCP_HINT)
]
async def run(self, runner: BetaAsyncToolRunner[_OutputT]) ->
ParsedBetaMessage[_OutputT]:
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/tools/claude_agent_sdk.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/claude_agent_sdk.py
index 49cae35fda9..4a9da23d161 100644
---
a/providers/common/ai/src/airflow/providers/common/ai/tools/claude_agent_sdk.py
+++
b/providers/common/ai/src/airflow/providers/common/ai/tools/claude_agent_sdk.py
@@ -42,7 +42,7 @@ import copy
import logging
from typing import TYPE_CHECKING, Any
-from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
collect_tools
+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.tool_definition import serialize_for_llm
from airflow.providers.common.ai.utils.tool_metrics import calling_framework
@@ -56,6 +56,11 @@ log = logging.getLogger(__name__)
__all__ = ["AirflowTools"]
+_MCP_HINT = (
+ "Pass the server to build_options(mcp_servers=...) and its tools
(mcp__<name>__*) to "
+ "allowed_tools, or set both on ClaudeAgentOptions, instead."
+)
+
_MAX_ERROR_DETAIL_CHARS = 500
_MAX_ERROR_ENTRIES = 5
@@ -186,7 +191,7 @@ class AirflowTools:
def __init__(self, *sources: ToolProvider | AirflowTool, server_name: str
= "airflow") -> None:
self.server_name = server_name
- airflow_tools = collect_tools(sources)
+ airflow_tools = _collect_tools(sources, mcp_hint=_MCP_HINT)
self.allowed_tools: list[str] = [f"mcp__{server_name}__{t.name}" for t
in airflow_tools]
# ``_run`` is the run in progress (guards re-entry); ``_last_run`` is
the one that most
# recently ended (refuses late calls). It is only read while ``_run``
is None.
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/tools/openai_agents.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/openai_agents.py
index 7da74585ecb..8d855171f82 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/tools/openai_agents.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/tools/openai_agents.py
@@ -52,7 +52,7 @@ except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
from airflow.providers.common.ai.observability import _content_capture_enabled
-from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
collect_tools
+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.tools.tracing import _task_identity
from airflow.providers.common.ai.utils.masking import dumps_masked,
mask_secrets
@@ -71,6 +71,11 @@ if TYPE_CHECKING:
__all__ = ["AirflowTools", "AirflowTraceProcessor"]
+_MCP_HINT = (
+ "Pass the server to the agent's mcp_servers with one of the SDK's
MCPServer classes, "
+ "such as MCPServerStreamableHttp, instead."
+)
+
class AirflowTools(list[Tool]):
"""
@@ -116,7 +121,7 @@ class AirflowTools(list[Tool]):
"""
def __init__(self, *sources: ToolProvider | AirflowTool) -> None:
- super().__init__(_to_function_tool(tool) for tool in
collect_tools(sources))
+ super().__init__(_to_function_tool(tool) for tool in
_collect_tools(sources, mcp_hint=_MCP_HINT))
class _FailedToolCall(ToolCallError, AgentsException):
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/tools/strands.py
b/providers/common/ai/src/airflow/providers/common/ai/tools/strands.py
index 2fb7d3416f0..a927c7e8429 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/tools/strands.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/tools/strands.py
@@ -36,7 +36,7 @@ except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
-from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
collect_tools
+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.tool_metrics import calling_framework
@@ -56,6 +56,8 @@ if TYPE_CHECKING:
__all__ = ["AirflowTools"]
+_MCP_HINT = "Connect Strands' MCPClient to the server instead."
+
# Strands refuses two plugins with the same name on one agent, so each
instance gets its own.
_instance_numbers = itertools.count(1)
@@ -102,7 +104,7 @@ class AirflowTools(Plugin):
self._name = f"airflow-tools-{next(_instance_numbers)}"
self._run = object()
self._airflow_tools: list[AgentTool] = [
- _to_strands_tool(tool, lambda: self._run) for tool in
collect_tools(sources)
+ _to_strands_tool(tool, lambda: self._run) for tool in
_collect_tools(sources, mcp_hint=_MCP_HINT)
]
super().__init__()
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/toolsets/mcp.py
b/providers/common/ai/src/airflow/providers/common/ai/toolsets/mcp.py
index 1b7cc80992a..43b860995ea 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/toolsets/mcp.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/toolsets/mcp.py
@@ -20,6 +20,8 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, Self
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
+from airflow.providers.common.ai.tools import _MCP_TOOLSET_NOT_SUPPORTED
from airflow.providers.common.ai.utils.toolset_base import AirflowToolset
if TYPE_CHECKING:
@@ -141,12 +143,10 @@ class MCPToolset(AirflowToolset):
"""
Not supported: use the agent framework's own MCP client instead.
- An MCP session belongs to the event loop that opened it, and the
framework-neutral
- tools run outside the Pydantic AI run that manages it, so every call
would reconnect
- to the server, and a stdio server would restart each time.
+ An MCP session belongs to the event loop that opened it, and a
framework adapter's
+ ``AirflowTools`` run outside the Pydantic AI run that manages it, so
every call would
+ reconnect to the server, and a stdio server would restart each time.
"""
- raise NotImplementedError(
- "MCPToolset works in Pydantic AI agents and through the LangChain
bridge, not through "
- "the framework-neutral tools. Connect your framework's own MCP
client to the server "
- "instead, such as Strands' MCPClient or ADK's McpToolset."
+ raise MCPToolsetNotSupportedError(
+ f"{_MCP_TOOLSET_NOT_SUPPORTED} Connect your framework's own MCP
client to the server instead."
)
diff --git a/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
b/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
index aeae08f9e1a..a9098cc1922 100644
--- a/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
+++ b/providers/common/ai/tests/unit/common/ai/durable/test_storage.py
@@ -127,11 +127,16 @@ class TestDeleteSteps:
class TestMissingConfig:
- def test_a_missing_cache_path_names_the_option(self):
+ def
test_a_missing_cache_path_names_the_option_without_naming_durable_true(self):
+ """run_durably() callers reach this too, and they never pass
durable=True."""
storage_module._get_base_path.cache_clear()
try:
with conf_vars({("common.ai", "durable_cache_path"): ""}):
- with pytest.raises(ValueError, match=r"\[common.ai\]
durable_cache_path"):
+ with pytest.raises(
+ ValueError, match=r"^Durable execution .* needs
\[common.ai\] durable_cache_path"
+ ) as exc_info:
DurableStorage(dag_id="d", task_id="t",
run_id="r").load_step("k")
+
+ assert "durable=True" not in str(exc_info.value)
finally:
storage_module._get_base_path.cache_clear()
diff --git a/providers/common/ai/tests/unit/common/ai/tools/test_adk.py
b/providers/common/ai/tests/unit/common/ai/tools/test_adk.py
index 00cb64ba766..e7193132629 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_adk.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_adk.py
@@ -32,8 +32,10 @@ from google.adk.models.llm_response import LlmResponse
from google.adk.runners import InMemoryRunner
from google.genai import types
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult
from airflow.providers.common.ai.tools.adk import AirflowTools
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from unit.common.ai.toolsets.test_sql import _make_mock_db_hook
@@ -120,6 +122,10 @@ class TestAirflowTools:
assert tool._detect_error_in_response({"error": "Unknown column"}) ==
"TOOL_ERROR"
assert tool._detect_error_in_response({"result": "ok"}) is None
+ def test_an_mcp_toolset_points_to_adks_mcp_toolset(self):
+ with pytest.raises(MCPToolsetNotSupportedError, match="ADK's
McpToolset"):
+ AirflowTools(MCPToolset("mcp_default"))
+
class TestAgentRun:
def test_the_agent_reads_a_toolset_result(self):
diff --git a/providers/common/ai/tests/unit/common/ai/tools/test_anthropic.py
b/providers/common/ai/tests/unit/common/ai/tools/test_anthropic.py
index b8a2d736f6d..2c1d07be24e 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_anthropic.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_anthropic.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import asyncio
import copy
import json
+import re
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
from uuid import uuid4
@@ -32,9 +33,10 @@ from anthropic import Anthropic, AsyncAnthropic,
DefaultAsyncHttpxClient, Defaul
from anthropic.types.beta import BetaTextBlock
from pydantic_ai.exceptions import ModelRetry
-from airflow.providers.common.ai.exceptions import ToolRunnerMaxIterationsError
+from airflow.providers.common.ai.exceptions import
MCPToolsetNotSupportedError, ToolRunnerMaxIterationsError
from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult
from airflow.providers.common.ai.tools.anthropic import AirflowTools,
AsyncAirflowTools
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from unit.common.ai.tools.test__from_toolset import _scripted
@@ -205,6 +207,15 @@ def
test_each_toolset_tool_becomes_a_runner_tool_with_its_schema(tools_class):
assert definitions[3]["input_schema"]["required"] == ["sql"]
[email protected]("tools_class", [AirflowTools, AsyncAirflowTools])
+def
test_an_mcp_toolset_points_to_the_anthropic_sdks_async_mcp_helper(tools_class):
+ """The SDK's sync mcp_tool() needs an AnyIO worker thread, so both runners
point at the async one."""
+ with pytest.raises(
+ MCPToolsetNotSupportedError,
match=re.escape("anthropic.lib.tools.mcp.async_mcp_tool()")
+ ):
+ tools_class(MCPToolset("mcp_default"))
+
+
@pytest.mark.parametrize("tools_class", [AirflowTools, AsyncAirflowTools])
def test_a_runner_tool_does_not_share_the_source_schema(tools_class):
"""A toolset's schema can be a module-level constant that the pydantic-ai
path also uses."""
diff --git
a/providers/common/ai/tests/unit/common/ai/tools/test_claude_agent_sdk.py
b/providers/common/ai/tests/unit/common/ai/tools/test_claude_agent_sdk.py
index 8788f02cb07..02b9f6f495e 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_claude_agent_sdk.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_claude_agent_sdk.py
@@ -20,6 +20,7 @@ import asyncio
import copy
import importlib
import json
+import re
import sys
from collections.abc import AsyncIterator
from typing import Any
@@ -32,8 +33,10 @@ pytest.importorskip("claude_agent_sdk")
from claude_agent_sdk import AssistantMessage, Message, ResultMessage,
TextBlock, UserMessage
from pydantic_ai.exceptions import ModelRetry
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult, collect_tools
from airflow.providers.common.ai.tools.claude_agent_sdk import AirflowTools
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
@@ -179,6 +182,13 @@ async def _request_real_server(tools: AirflowTools,
method: str, params: dict[st
class TestTools:
+ def test_an_mcp_toolset_points_to_the_sdks_mcp_servers(self):
+ with pytest.raises(
+ MCPToolsetNotSupportedError,
+ match=re.escape("build_options(mcp_servers=...) and its tools
(mcp__<name>__*) to allowed_tools"),
+ ):
+ AirflowTools(MCPToolset("mcp_default"))
+
def test_each_toolset_tool_becomes_an_mcp_tool_with_its_schema(self):
tools, handlers = _build_tools(_tool(ToolResult("ok")),
SQLToolset("pg_default"))
diff --git
a/providers/common/ai/tests/unit/common/ai/tools/test_openai_agents.py
b/providers/common/ai/tests/unit/common/ai/tools/test_openai_agents.py
index 949f163dc2b..afb82eee50c 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_openai_agents.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_openai_agents.py
@@ -51,6 +51,7 @@ from opentelemetry.trace import SpanKind, StatusCode
from pydantic_ai.exceptions import ModelRetry
from pydantic_ai.toolsets.function import FunctionToolset
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult
from airflow.providers.common.ai.tools._from_toolset import
airflow_tools_from_toolset
from airflow.providers.common.ai.tools.openai_agents import (
@@ -60,6 +61,7 @@ from airflow.providers.common.ai.tools.openai_agents import (
_mask_content,
)
from airflow.providers.common.ai.tools.tracing import agent_framework_tracing
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from tests_common.test_utils.config import conf_vars
@@ -189,6 +191,10 @@ class TestAirflowTools:
assert source.parameters == before
+ def test_an_mcp_toolset_points_to_the_sdks_mcp_servers(self):
+ with pytest.raises(MCPToolsetNotSupportedError, match="mcp_servers .*
MCPServerStreamableHttp"):
+ AirflowTools(MCPToolset("mcp_default"))
+
class TestAgentRun:
def test_the_agent_reads_a_toolset_result(self):
diff --git a/providers/common/ai/tests/unit/common/ai/tools/test_strands.py
b/providers/common/ai/tests/unit/common/ai/tools/test_strands.py
index 9bda221008e..593cc8e635b 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_strands.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_strands.py
@@ -33,9 +33,11 @@ from strands.models.model import Model
from strands.tools.registry import ToolRegistry
from strands.types.exceptions import EventLoopException
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult
from airflow.providers.common.ai.tools._from_toolset import
airflow_tools_from_toolset
from airflow.providers.common.ai.tools.strands import AirflowTools
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from unit.common.ai.toolsets.test_sql import _make_mock_db_hook
@@ -128,6 +130,10 @@ def _run_agent(plugin: AirflowTools, tool_name: str,
tool_input: dict[str, Any])
class TestAirflowTools:
+ def test_an_mcp_toolset_points_to_strands_mcp_client(self):
+ with pytest.raises(MCPToolsetNotSupportedError, match="Strands'
MCPClient"):
+ AirflowTools(MCPToolset("mcp_default"))
+
def test_adds_every_tool_of_a_toolset(self):
plugin = AirflowTools(SQLToolset("pg_default"))
diff --git a/providers/common/ai/tests/unit/common/ai/tools/test_tools.py
b/providers/common/ai/tests/unit/common/ai/tools/test_tools.py
index 35d1975382c..be80d3b3fc7 100644
--- a/providers/common/ai/tests/unit/common/ai/tools/test_tools.py
+++ b/providers/common/ai/tests/unit/common/ai/tools/test_tools.py
@@ -18,11 +18,20 @@ from __future__ import annotations
import asyncio
import traceback
+from functools import partial
from typing import Any
import pytest
-from airflow.providers.common.ai.tools import AirflowTool, ToolCallError,
ToolResult, collect_tools
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
+from airflow.providers.common.ai.tools import (
+ AirflowTool,
+ ToolCallError,
+ ToolResult,
+ _collect_tools,
+ collect_tools,
+)
+from airflow.providers.common.ai.toolsets.mcp import MCPToolset
def _tool(function) -> AirflowTool:
@@ -112,6 +121,38 @@ class TestCollectTools:
):
collect_tools([_Provider(_named("query")),
_Provider(_named("query"))])
+ @pytest.mark.parametrize(
+ ("collect", "advice"),
+ [
+ pytest.param(
+ collect_tools, "Connect your framework's own MCP client to the
server instead.", id="public"
+ ),
+ pytest.param(
+ partial(_collect_tools, mcp_hint="Connect Acme's McpClient
instead."),
+ "Connect Acme's McpClient instead.",
+ id="adapter-hint",
+ ),
+ ],
+ )
+ def test_an_mcp_toolset_error_says_what_to_connect_instead(self, collect,
advice):
+ with pytest.raises(MCPToolsetNotSupportedError) as exc_info:
+ collect([MCPToolset("mcp_default")])
+
+ assert str(exc_info.value) == (
+ "MCPToolset works in Pydantic AI agents and through the LangChain
bridge, not through "
+ f"AirflowTools. {advice}"
+ )
+
+ def test_other_not_implemented_errors_pass_through_unchanged(self):
+ """The hint is MCP-specific; another provider's NotImplementedError
keeps its own message."""
+
+ class _Unfinished:
+ def airflow_tools(self) -> list[AirflowTool]:
+ raise NotImplementedError("not yet")
+
+ with pytest.raises(NotImplementedError, match="^not yet$"):
+ _collect_tools([_Unfinished()], mcp_hint="Connect Acme's McpClient
instead.")
+
@pytest.mark.enable_redact
class TestAirflowToolMasking:
diff --git a/providers/common/ai/tests/unit/common/ai/toolsets/test_mcp.py
b/providers/common/ai/tests/unit/common/ai/toolsets/test_mcp.py
index 11745066c63..5391d62e0e0 100644
--- a/providers/common/ai/tests/unit/common/ai/toolsets/test_mcp.py
+++ b/providers/common/ai/tests/unit/common/ai/toolsets/test_mcp.py
@@ -21,6 +21,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
+from airflow.providers.common.ai.exceptions import MCPToolsetNotSupportedError
from airflow.providers.common.ai.toolsets.mcp import MCPToolset
_HOOK_PATH = "airflow.providers.common.ai.hooks.mcp.MCPHook"
@@ -136,5 +137,10 @@ class TestMCPToolsetDelegation:
class TestMCPToolsetOutsidePydanticAI:
def test_points_other_frameworks_to_their_own_mcp_client(self):
- with pytest.raises(NotImplementedError, match="framework's own MCP
client"):
+ with pytest.raises(MCPToolsetNotSupportedError) as exc_info:
MCPToolset(mcp_conn_id="mcp_default").airflow_tools()
+
+ assert str(exc_info.value) == (
+ "MCPToolset works in Pydantic AI agents and through the LangChain
bridge, not through "
+ "AirflowTools. Connect your framework's own MCP client to the
server instead."
+ )
