OpenAI Agents SDK

You have an agent built with the OpenAI Agents SDK and you want it to run as an Airflow task, reading data through connections your deployment already manages. Keep the agent as it is, and add Airflow’s toolsets to its tools with AirflowTools.

Note

AirflowTools is experimental: it may change in a minor release of this provider. The framework-neutral tool interface it builds on is stable. See Stable and experimental features.

Run an OpenAI Agents SDK agent in a task

The agent below answers a question about a database with a SQLToolset. Its model’s API key, and optionally the base URL of another OpenAI-compatible endpoint, come from an Airflow connection:

airflow/providers/common/ai/example_dags/example_openai_agents_agent.py[source]

@dag(tags=["example"])
def example_openai_agents_agent():
    """Answer a question about a database with an OpenAI Agents SDK agent."""

    @task
    def run_openai_agent(question: str = DEFAULT_QUESTION) -> str:
        from agents import Agent, OpenAIChatCompletionsModel, Runner, set_trace_processors
        from openai import AsyncOpenAI

        from airflow.providers.common.ai.tools.openai_agents import AirflowTools, AirflowTraceProcessor
        from airflow.providers.common.ai.tools.tracing import agent_framework_tracing
        from airflow.providers.common.ai.toolsets.sql import SQLToolset

        # Record the SDK's traces as the task's OpenTelemetry spans instead of sending them to OpenAI.
        set_trace_processors([AirflowTraceProcessor()])

        llm = BaseHook.get_connection(LLM_CONN_ID)
        agent = Agent(
            name="analyst",
            model=OpenAIChatCompletionsModel(
                model=LLM_MODEL,
                openai_client=AsyncOpenAI(api_key=llm.password, base_url=llm.host or None),
            ),
            instructions=(
                "You are a SQL analyst. Use list_tables and get_schema to explore "
                "the database, then run read-only queries to answer the question."
            ),
            tools=AirflowTools(SQLToolset(db_conn_id=DB_CONN_ID)),
        )
        with agent_framework_tracing():
            return Runner.run_sync(agent, question).final_output

    run_openai_agent()

Everything except AirflowTools is plain OpenAI Agents SDK: the agent, its runner, handoffs, guardrails and your own function tools work as the SDK documentation describes. AirflowTools is a list of FunctionTool objects, so add your own tools next to them with tools=[*AirflowTools(warehouse), my_tool].

What the tools give the agent

AirflowTools takes the same arguments as the Strands plugin (see Strands Agents): toolsets that implement ToolProvider, such as SQLToolset and HookToolset, and individual AirflowTool objects. MCPToolset is the exception: attach an MCP server with the SDK’s own mcp_servers=. Each tool keeps the toolset tool’s name, description and argument schema, runs through the toolset, and passes its result through Airflow’s secret masker.

  • The schema goes to the model as it is, with strict_json_schema=False. The SDK’s strict mode would mark every argument required, including the ones a toolset leaves optional.

  • A failure the model can correct, such as a rejected SQL statement or an invalid argument, reaches it as {"error": ...}. The tool’s retry limit bounds how many times in a row the model can try again, counted afresh on every run of the agent.

  • A refusal the tool reports as final, such as a file that does not exist, also comes back as {"error": ...} without counting against that limit, and so do arguments that are not a JSON object. The max_turns argument of Runner.run bounds how long a model can keep asking.

  • Any other failure raises ToolCallError out of Runner.run and Runner.run_sync, with the message masked. The task fails and Airflow’s own retry takes over.

Outside AgentOperator, a toolset’s connection ID is used as written: it is not rendered as a template.

An agent used as a tool

Agent.as_tool() turns any exception from the inner agent’s run into text for the outer model by default, so a tool failure that should fail the task would only reach the orchestrating model as a message. When the inner agent holds AirflowTools, pass failure_error_function=None:

analyst = Agent(name="analyst", model=model, tools=AirflowTools(warehouse))
lead = Agent(
    name="lead",
    model=model,
    tools=[analyst.as_tool("ask_analyst", "Ask the SQL analyst.", failure_error_function=None)],
)

The ToolCallError then fails the outer run, and with it the task. The same applies to a function tool of your own that runs another agent: let the exception propagate. A nested run shares the outer run’s retry count for each tool.

Tracing

The SDK has a tracing system of its own rather than OpenTelemetry. By default it sends every run’s traces to OpenAI’s Traces backend using OPENAI_API_KEY from the environment, whatever model the agent runs, including a self-hosted one, and the traces hold the prompts and model output. Tool results are masked before the model sees them; nothing else in those traces is.

AirflowTraceProcessor is a trace processor for the SDK that records each trace as OpenTelemetry spans under the task’s span instead. The example replaces the SDK’s processors with it, so nothing goes to OpenAI, and runs the agent inside agent_framework_tracing(), so the spans carry the task’s identity:

from agents import Runner, set_trace_processors

from airflow.providers.common.ai.tools.openai_agents import AirflowTraceProcessor
from airflow.providers.common.ai.tools.tracing import agent_framework_tracing

set_trace_processors([AirflowTraceProcessor()])
with agent_framework_tracing():
    answer = Runner.run_sync(agent, question).final_output

set_trace_processors replaces every processor registered with the SDK. To keep the export to OpenAI, or a processor another instrumentation registered, add this one with agents.add_trace_processor instead. The spans use the names of the OpenTelemetry GenAI conventions (invoke_agent, chat with the model and token counts, execute_tool), and the SDK’s other spans are named openai_agents.<kind>. Prompts, completions, and tool arguments and results are left out unless [common.ai] otel_export_enabled and capture_content are both on, and pass through Airflow’s secret masker when they are recorded; see Observability (OpenTelemetry tracing).

Use AirflowTraceProcessor when span content should follow [common.ai] capture_content and Airflow’s secret masker.

To turn the SDK’s tracing off altogether, call agents.set_tracing_disabled(True) or set OPENAI_AGENTS_DISABLE_TRACING=1 in the worker’s environment. That stops AirflowTraceProcessor from receiving anything too.

Differences from AgentOperator

  • A task retry runs the agent again from the start, so tools that change data must be safe to repeat.

  • There is no human review step like AgentOperator(enable_hitl_review=True), and the SDK’s own needs_approval interruptions are not connected to Airflow’s approval UI.

Installation

pip install "apache-airflow-providers-common-ai[sql]" "openai-agents>=0.20.0"

The SDK requires openai>=3 from 0.21.0 on. Airflow’s constraints file pins openai below 3, so with that file the command above installs openai-agents 0.20, the last release that accepts openai 2.x. AirflowTools works with both.

Was this entry helpful?