[vikramkoka]

Initial PR to create a common interface for invoking Managed Agents across clouds, which would enable:

1. Automated agent failover across cloud providers, and
2. Composite agent invocations.

The common interface is an abstraction, which would then be implemented across Managed Agent implementations for individual clouds including AWS Bedrock, Azure Foundry, and Google Vertex.

Just in case it wasn't clear above, this is intended to be part of the Common AI provider.

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

Generated-by: Claude Code (Sonnet 5)

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[kaxil]

call_tool resolves self.agent_ref before it calls invoke(), and on a failover group that property (line 306) joins agent_ref for every member, standbys included, so a standby whose identity comes from a missing or misconfigured connection makes the whole tool call fail before the healthy primary is tried. Since the standby usually lives on a different cloud, a broken standby connection is the likely case rather than the exotic one. I ran it against this code: group.invoke(prompt) returns the primary's answer, while group.call_tool(...) on the same group raises RuntimeError: Connection 'azure_standby' not found.

[kaxil]

ref = member.agent_ref sits outside the try, and the standby lookup on line 323 sits inside the except handler but is equally unguarded, so a failure to resolve identity bypasses failover entirely: a broken primary ref raises before any member is invoked, and a broken standby ref replaces the primary's transient failure with the standby's config error without the standby ever being tried. The docs say a transient failure moves to the next member, but with a misconfigured standby a primary 503 surfaces as a connection error instead. Resolving each ref lazily, or wrapping both lookups so a failed lookup degrades to the name alone, would keep failover independent of identity resolution.

[kaxil]

timeout is documented as the seconds to wait for a single invocation, but nothing in the base class reads self._timeout, so the only way a subclass can honor it is to reach into a private attribute of another package's class. agent_ref and replayable are public for exactly that reason. A public timeout property, plus a docstring line saying that honoring it is the implementation's job, would give provider subclasses a supported way in.

[kaxil]

Inside a group, each member's tool_name, description and max_retries are inert: the group builds its single ToolDefinition from its own values and only ever calls member.invoke(), never member.get_tools(). The docs example still has to give each member a tool_name that nothing reads, so a line in the docs would stop someone tuning max_retries on a member and wondering why it does nothing.

[kaxil]

managed_agent.served only ever fires inside a failover group, so a plain BaseManagedAgentToolset, which the docs call the normal case of one toolset per task, emits no metrics at all, and the failover counter is left without a denominator. The name reads as though it covers every managed agent invocation while counting only the grouped ones. Should the base class emit it too, taking care not to double count when the group's inherited call_tool runs?