# OpenAI Agents SDK

> Make an OpenAI Agents SDK run durable by giving it the journaled AsyncOpenAI: its model calls replay after a restart, and a tool's side effect runs once when it is a Pyrula step.

Source: https://docs.pyrula.io/agents/integrations/openai-agents-sdk/

The OpenAI Agents SDK needs no adapter. Give it Pyrula's journaled `AsyncOpenAI` as its client and run the agent inside a Pyrula `@workflow`: every model call the SDK makes, handoff and guardrail calls included, is recorded in the run's journal. After a crash the run resumes from the top and each call that had finished is answered from the journal, with no second request.

The SDK's function tools run in your process and are not journaled. A side effect is durable when the tool calls a Pyrula [`@step`](https://docs.pyrula.io/workflows/guides/steps.md) (see [Tools](#tools)).

```bash
pip install "pyrula[agents]" openai-agents
```
`openai-agents` 0.21 and later needs `openai` 3, so it can't share an environment with `pyrula[litellm]` until litellm lifts its `openai<3` cap. Install the two in separate environments. See [Versions](#versions).

## Use it

Set the client once when the worker starts, before the first run. The agent keeps its ordinary `model="gpt-6-luna"` string.

```python
import asyncio
import json

import httpx2
from agents import (
    Agent,
    RunConfig,
    Runner,
    function_tool,
    set_default_openai_client,
    set_tracing_disabled,
)
from agents.run_config import ToolExecutionConfig
from pyrula.contracts.events import EventKind
from pyrula.workflows import step, workflow
from pyrula.workflows.testing import WorkflowTestEnv

from pyrula.agents.compat.openai import AsyncOpenAI

requests: list[dict] = []


def fake_openai(request: httpx2.Request) -> httpx2.Response:
    """Stands in for api.openai.com so this page runs offline. Leave http_client out for the real API."""
    body = json.loads(request.content)
    requests.append(body)
    answered = any(item.get("type") == "function_call_output" for item in body["input"])
    output = (
        [{"id": "msg_1", "type": "message", "role": "assistant", "status": "completed",
          "content": [{"type": "output_text", "text": "Charged order A-1.", "annotations": []}]}]
        if answered
        else [{"id": "fc_1", "type": "function_call", "call_id": "call_1", "name": "charge_order",
               "arguments": json.dumps({"order": "A-1"}), "status": "completed"}]
    )
    return httpx2.Response(200, json={
        "id": f"resp_{len(requests)}", "object": "response", "created_at": 0, "model": "gpt-6-luna",
        "status": "completed", "output": output, "parallel_tool_calls": True,
        "tool_choice": "auto", "tools": [],
    })


set_default_openai_client(
    AsyncOpenAI(
        api_key="test",
        base_url="http://openai.test/v1",
        http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(fake_openai)),
    ),
    use_for_tracing=False,
)
set_tracing_disabled(True)

charged: list[str] = []
receipts: list[str] = []


@step
async def charge(order: str) -> str:
    charged.append(order)
    return f"charged {order}"


@step
async def send_receipt(text: str) -> str:
    receipts.append(text)
    if len(receipts) == 1:
        raise RuntimeError("the mail server is down")
    return "sent"


@function_tool(failure_error_function=None)
async def charge_order(order: str) -> str:
    """Charge an order."""
    return await charge(order)


support = Agent(name="support", instructions="Charge the order.", tools=[charge_order], model="gpt-6-luna")


@workflow(name="docs_oas_support")
async def answer(ctx, question: str) -> str:
    result = await Runner.run(
        support,
        question,
        run_config=RunConfig(tool_execution=ToolExecutionConfig(max_function_tool_concurrency=1)),
    )
    await send_receipt(result.final_output)
    return result.final_output


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.client.submit("docs_oas_support", {"question": "charge order A-1"})
    try:
        await run.result()
    except Exception:
        pass  # the receipt step failed after the agent finished
    assert len(requests) == 2 and charged == ["A-1"]

    retried = await run.retry()
    assert await retried.result() == "Charged order A-1."
    assert len(requests) == 2  # both model calls were served from the journal
    assert charged == ["A-1"]  # and the charge was not made again
    assert len(receipts) == 2  # the step that had failed ran again
    assert [
        e.payload["step_id"]
        for e in env.store.load_run("docs_oas_support", retried.run_id)
        if e.kind == EventKind.STEP_DONE
    ] == [
        "call:openai.responses.create:0",
        "charge",
        "call:openai.responses.create:1",
        "send_receipt",
    ]


asyncio.run(main())
```
Against the real API, build the client from your key and leave out `http_client`:

```python
from agents import set_default_openai_client
from pyrula.agents.compat.openai import AsyncOpenAI

set_default_openai_client(AsyncOpenAI(), use_for_tracing=False)  # reads OPENAI_API_KEY
```
Outside a Pyrula run the client passes calls through to OpenAI with a one-time warning, so the same code runs in a script. Call `await Runner.run(...)` or `Runner.run_streamed(...)` in the workflow body, never `Runner.run_sync`. Inside an `@agent` it works too; as with the other drop-in clients, the model calls are then also metered and held to `AgentLimits`, while in a plain `@workflow` they write `llm:done` but nothing holds a cap, see the [drop-in clients](https://docs.pyrula.io/agents/integrations/durable-vs-not.md#the-openai-drop-in).

`use_for_tracing=False` keeps the SDK from using this client's key to export traces. Trace export is a separate path and is sent again on a replay; `set_tracing_disabled(True)` turns it off.

## Tools

The SDK runs a function tool in your process, outside the journal. On a resume the model's request for the tool replays, and the tool body runs again, possibly more than once (a retry replays the body against the journal before its live attempt). Make an effect run once by putting it in a `@step` the tool calls, and set the two options the example sets:

- `@function_tool(failure_error_function=None)`. The SDK otherwise turns an exception from the tool into text for the model, and that includes the error of a failed step. The model answers the text and the run goes on as if the tool had only failed. A `ReplayInputChanged` from the step is hidden the same way and surfaces one model call later. With `None` the error ends the run with its own type, and a retry resumes it.
- `max_function_tool_concurrency=1` in `RunConfig(tool_execution=...)`. The SDK runs the tool calls of one model reply in parallel, and two calls into the same `@step` overlap, which fails the run with `ConcurrentStepIdError`. At 1 each call runs and replays in turn.

A step that is cut off after its effect and before its record runs again on resume, so an effect that can't repeat needs the step's `idempotency_key` ([the crash window](https://docs.pyrula.io/workflows/concepts/replay-and-recovery.md#the-window-where-a-step-can-run-twice)). A tool without a step is best effort, as for the other drop-in clients: `compat.openai`'s [tools cell](https://docs.pyrula.io/agents/integrations/guarantees.md). Computer and local-shell tools run in process and are not journaled either.

## What replays

| What the SDK does | On a resume |
|---|---|
| A model call, streamed or not, on the default Responses API or on `chat_completions` | Served from the journal. No request is sent. |
| A handoff | Nothing extra: a handoff is a model call and SDK-local bookkeeping. |
| An input guardrail that runs an agent | Its model calls are journaled. |
| A `WebSearchTool` or `FileSearchTool` call | Journaled as part of the model call that made it. |
| A function tool in a `@step` | The step's result is served. |
| A function tool without a step | Runs again, possibly more than once. |
| A call made through the SDK's own retry path | Journaled; the SDK asks the client for `with_options(max_retries=0)` and gets the journaled one. |

A resume replays a call only if what it asked is the same. A changed prompt, history, instructions or tool list raises `ReplayInputChanged` out of `Runner.run`, and so does a changed argument to a step inside a tool.

Each call is a `step:pending` and `step:done` pair with the step id `call:openai.responses.create:<index>` (`call:openai.chat.completions.create:<index>` on `chat_completions`, and a `.stream` suffix when streamed), and a tool's step keeps its own id. Cloud shows them like any other drop-in call.

To use Chat Completions instead of the default Responses API, add one line:

```python
from agents import set_default_openai_api

set_default_openai_api("chat_completions")
```
Code that must not set process globals passes the client to one run:

```python
from agents import RunConfig
from agents.models.openai_provider import OpenAIProvider

config = RunConfig(model_provider=OpenAIProvider(openai_client=client))
result = await Runner.run(support, question, run_config=config)
```
A `RunConfig` keeps the client its provider first resolved. Build a fresh one per run if the default client can change.

### Guardrails

A guardrail function is plain code. One that runs an agent has its model calls journaled. Give a guardrail that calls a model `run_in_parallel=False`. In parallel mode a tripwire cancels the main model call while it is in flight, and a cancelled call is not journaled, so it can be sent again on a retry. The order of the two calls rests on which task starts first, and a change to it raises `ReplayInputChanged` instead of serving the wrong reply.

## What is not durable

Not journaled, and nothing says so when the run starts:

- The Responses WebSocket transport (`set_default_openai_responses_transport("websocket")`, `use_responses_websocket=True`) connects without going through the client. Leave it at the default HTTP transport.
- `LitellmModel`, any-llm models, Realtime and voice agents call their provider directly.
- Function, computer and local-shell tool bodies outside a `@step`.

Not durable, and documented or loud:

- SDK `Session` memory. `Runner.run(session=...)` adds the first attempt's items to the session, so on a resume the first call's input has changed and the run raises `ReplayInputChanged`. Keep the history in the workflow and pass it as the input.
- Tool approval interruptions (`needs_approval`). They end the SDK run and are not a durable pause. Use a Pyrula [`@tool`](https://docs.pyrula.io/agents/authoring/tools.md) with `requires_approval` for that.
- Trace export, which is sent again on a replay.

The journaled client refuses what a replay could not repeat: `Runner.run(conversation_id=...)`, background responses, and the hosted tools that act (code interpreter, hosted MCP servers, image generation) raise `UnjournaledSurface` before any request. See [the OpenAI drop-in](https://docs.pyrula.io/agents/integrations/durable-vs-not.md#the-openai-drop-in).

## Versions

Tested with `openai-agents` 0.24 on `openai` 3.41, in a CI step of its own. `openai-agents` 0.21 and later needs `openai` 3, and `pyrula[litellm]` caps `openai` below 3, so the two can't share an environment until litellm lifts its cap. Releases 0.17 to 0.20 use `openai` 2.x and call the same two client methods; they are not tested.

The [guarantee matrix](https://docs.pyrula.io/agents/integrations/guarantees.md) row is `openai-agents (recipe)`.