OpenAI agents

The OpenAI Agents SDK is a lightweight framework for building multi-tool, multi-agent applications. Union.ai ships a first-party integration, flyteplugins-openai, that lets you expose durable Flyte tasks as Agents SDK tools with a single decorator.

The integration provides function_tool from flyteplugins.openai.agents. Stack it on top of @env.task and the resulting object is both a durable Flyte task and an OpenAI Agents SDK tool. When the agent calls the tool, it executes on-cluster — durable, retryable, and observable in the dashboard.

Tools that are also durable tasks

Each tool below is a Flyte task wrapped as an Agents SDK tool. The agent task builds an Agents SDK Agent, hands it the tools, and runs it with Runner.run:

openai_agents_agent.py 
import asyncio
from typing import Optional

from agents import Agent, Runner
from flyteplugins.openai.agents import function_tool

import flyte

agent_env = flyte.TaskEnvironment(
    "openai-agent",
    resources=flyte.Resources(cpu=1),
    secrets=[flyte.Secret(key="OPENAI_API_KEY")],
    image=flyte.Image.from_debian_base(python_version=(3, 13)).with_pip_packages(
        "flyteplugins-openai",
        "openai-agents",
    ),
)


@function_tool
@agent_env.task
async def get_bread() -> str:
    await asyncio.sleep(1)
    return "bread"


@function_tool
@agent_env.task
async def get_peanut_butter() -> str:
    await asyncio.sleep(1)
    return "peanut butter"


@function_tool
@agent_env.task
async def spread_peanut_butter(bread: str, peanut_butter: str) -> str:
    await asyncio.sleep(1)
    return f"{bread} with {peanut_butter}"


@function_tool
@agent_env.task
async def assemble_sandwich(pb_bread: Optional[str] = None, j_bread: Optional[str] = None) -> str:
    await asyncio.sleep(1)
    return f"{pb_bread} and {j_bread} combined"


@agent_env.task
async def agent(goals: list[str]) -> list[str]:
    async def run_agent(goal: str, index: int) -> str:
        with flyte.group(f"sandwich-maker-{index}"):
            result = await Runner.run(
                Agent(
                    name="sandwich_maker",
                    instructions="You are a sandwich-making assistant.",
                    tools=[
                        get_bread,
                        get_peanut_butter,
                        spread_peanut_butter,
                        assemble_sandwich,
                    ],
                ),
                input=goal,
            )
            return result.final_output

    # Each goal runs its own agent loop, fanned out in parallel.
    tasks = [run_agent(goal, idx) for idx, goal in enumerate(goals, start=1)]
    return list(await asyncio.gather(*tasks))

What’s happening under the hood:

  • @function_tool (from flyteplugins.openai.agents) adapts the durable @env.task into an OpenAI Agents SDK tool, so the agent can call it through the SDK’s normal tool-calling mechanism.
  • Each tool call executes as a Flyte task on the cluster — durable and observable — even though the agent loop itself orchestrates them.
  • flyte.group(...) groups each agent’s tool calls under a named span, and asyncio.gather() fans out the per-goal agents in parallel, each in its own container.

Next steps