airflow.providers.common.ai.toolsets.skills

A pydantic-ai toolset that loads Agent Skills.

AgentSkillsToolset is a normal pydantic-ai AbstractToolset: it can be passed to AgentOperator via toolsets= or used directly with a pydantic_ai.Agent anywhere the Airflow connection backend is reachable (i.e. inside a worker/task runtime).

Skill sources are resolved lazily when the agent enters the toolset (run time, on the worker), never at DAG-parse time, so a Git token resolved from an Airflow connection is never baked into the serialized DAG. Cloned repositories are removed when the toolset context exits.

Attributes

AbstractToolset

Classes

AgentSkillsToolset

A pydantic-ai toolset that loads Agent Skills, with Git credentials from Airflow connections.

Module Contents

airflow.providers.common.ai.toolsets.skills.AbstractToolset[source]
class airflow.providers.common.ai.toolsets.skills.AgentSkillsToolset(sources, *, exclude_tools=None)[source]

Bases: pydantic_ai.toolsets.abstract.AbstractToolset

A pydantic-ai toolset that loads Agent Skills, with Git credentials from Airflow connections.

Sources are local directory paths and/or GitSkills.

Parameters:

  • sources (list[airflow.providers.common.ai.skills.SkillSource]) – Skill sources – local directory paths and/or GitSkills.

  • exclude_tools (set[str] | None) – Optional set of skill tool names to hide from the agent (e.g. {"run_skill_script"} to disable on-worker script execution).

Requires the skills extra: pip install "apache-airflow-providers-common-ai[skills]".

property id: str | None[source]

An ID for the toolset that is unique among all toolsets registered with the same agent.

If you’re implementing a concrete implementation that users can instantiate more than once, you should let them optionally pass a custom ID to the constructor and return that here.

A toolset needs to have an ID in order to be used in a durable execution environment like Temporal, in which case the ID will be used to identify the toolset’s activities within the workflow.

async for_run(ctx)[source]

Return the toolset to use for this agent run.

Called once per run, before __aenter__. Override this to return a fresh instance for per-run state isolation. Default: return self (shared across runs).

async __aenter__()[source]

Enter the toolset context.

This is where you can set up network connections in a concrete implementation.

async __aexit__(*args)[source]

Exit the toolset context.

This is where you can tear down network connections in a concrete implementation.

async get_tools(ctx)[source]

The tools that are available in this toolset.

async call_tool(name, tool_args, ctx, tool)[source]

Call a tool with the given arguments.

Args:

name: The name of the tool to call. tool_args: The arguments to pass to the tool. ctx: The run context. tool: The tool definition returned by [get_tools][pydantic_ai.toolsets.AbstractToolset.get_tools] that was called.

async get_instructions(ctx)[source]

Return instructions for how to use this toolset’s tools.

Override this method to provide instructions that help the agent understand how to use the tools in this toolset effectively.

Simple implementations can return a plain str; advanced implementations can return [InstructionPart][pydantic_ai.messages.InstructionPart] objects to indicate whether each instruction block is static or dynamic for caching purposes.

Args:

ctx: The run context for this agent run.

Returns:

Instruction string, InstructionPart, list of either, or None. Plain str values are treated as dynamic instructions by default.

Was this entry helpful?