Introduce AWS Strands Agents hook to common AI provider and Skills integration

[gopidesupavan]

Add Strands Agents hook to common AI provider

Summary

Add AWS Strands Agents as a new agent backend for AgentOperator and @task.agent in the common AI provider, building on the BaseAIHook contract.

• Introduce StrandsHook (shared Strands SDK integration) and StrandsGeminiHook as the first backend (conn_type: strands-gemini, default connection ID: strands_default)

• Wire toolsets through _tool_spec_to_native, converting ToolSpec instances to Strands-native tools

• Add skills support end-to-end: SkillSpec dataclass on BaseAIHook, skills / skills_params on AgentRunRequest and AgentOperator, and Strands AgentSkills plugin integration for filesystem paths and inline skill definitions

• Register the new connection type in provider.yaml / get_provider_info.py and add optional dependency:

  pip install 'apache-airflow-providers-common-ai[strands]'

  (strands-agents[gemini]>=1.0.0)

• Add example DAGs (example_strands.py) covering basic operator usage, skills, inline SkillSpec + SQL toolset, direct hook usage, and @task.agent

• Document connection setup, hook usage, and operator skills in new/updated RST pages

Depends on

BaseAIHook PR #67438

---

Follow-ups

Durable execution for Strands (durable=True)

StrandsHook currently sets supports_durable=False. A follow-up PR should mirror the pydantic-ai durable path so Strands agents can resume from cached steps on task retry.

Out of scope for this PR: usage limits for Strands hooks.

Skills for Pydantic AI (pydantic-ai-skills)

PydanticAIHook currently leaves supports_skills=False, so AgentOperator.skills / skills_params only work with Strands backends in this PR. A follow-up should wire the same operator-level skills API to pydantic-ai via the pydantic-ai-skills library (Agent Skills / agentskills.io spec with progressive disclosure).

Files changed

Area	Files
Hooks	hooks/strands_ai.py, hooks/base_ai.py
Operator	operators/agent.py
Examples	example_dags/example_strands.py
Provider metadata	provider.yaml, get_provider_info.py, pyproject.toml
Docs	docs/connections/strands.rst, docs/hooks/strands_ai.rst, docs/operators/agent.rst, …
Tests	tests/unit/common/ai/hooks/test_strands_ai.py, test_base_ai.py, operators/test_agent.py

---

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

• Yes (please specify the tool below)

---

• 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.

[gopidesupavan]

Thanks you @kaxil — resolved comments.

[kaxil]

This constructs AgentRunResult with only output and model_name, so usage and tool_names stay at their None defaults. log_run_summary then skips both the token counts (logging.py:35) and the tool-call sequence (logging.py:49) for Strands even when tools ran, whereas the pydantic-ai path populates both. Strands' result exposes usage metrics and a tool trace; mapping them into AgentUsage / tool_names here would match the pydantic-ai path. Use getattr(..., None) guards if the metrics shape varies across SDK versions. (Low priority: no crash, just missing observability.)

[gopidesupavan]

Yeah this needs refactor , will updated..

[github-actions]

uv.lock on main just moved via commit fd0f7e2 and this PR currently conflicts.

Quickest fix:

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.

[aaron-y-chen]

A potential bug I found, I hope this helps :)

[aaron-y-chen]

Hi, I found that an AttributeError occurs when I run the code below, is this the expected behavior?

from strands import Agent
from strands.models import Model

class FakeModel(Model):
    def get_config(self): return {}
    def update_config(self, **kw): pass
    async def stream(self, *a, **k):
        raise AssertionError("should never reach inference")
    async def structured_output(self, *a, **k):
        raise AssertionError("should never reach inference")

agent = Agent(model=FakeModel(), tools=[], structured_output_model=str)
print("Agent constructed OK")
agent("hello")

• Output: AttributeError: type object 'str' has no attribute 'model_json_schema'
• strands-agents version: 1.41.0

[gopidesupavan]

are you using this hook @aaron-y-chen ?

[aaron-y-chen]

I ran another experiment and reproduced it through StrandsGeminiHook itself, against a real connection (my own Gemini API key, model gemini-3.6-flash):

Repro code

import json
import os
from unittest.mock import patch

from airflow.models.connection import Connection
from airflow.providers.common.ai.hooks.base import AgentRunRequest
from airflow.providers.common.ai.hooks.strands_ai import StrandsGeminiHook

hook = StrandsGeminiHook()
conn = Connection(
    conn_id="strands_default",
    conn_type="strands-gemini",
    password=os.environ.get("GOOGLE_API_KEY", "unused"),
    extra=json.dumps({"model": "gemini-3.6-flash"}),
)
request = AgentRunRequest(prompt="Reply with the single word: pong")  # output_type defaults to str

with patch.object(hook, "get_connection", return_value=conn):
    agent = hook.create_agent(request)
    hook.run_agent(agent, request)

# AttributeError: type object 'str' has no attribute 'model_json_schema'