---
title: OpenAI Agents Adapter
description: Wrap an OpenAI Agents SDK Agent with KitaruRunner so calls are durable and replayable inside Kitaru flows
---

Kitaru's OpenAI Agents adapter lets you keep your existing OpenAI Agents SDK
agent logic while adding Kitaru durability around it.

```python
from agents import Agent
from kitaru.adapters.openai_agents import KitaruRunner

agent = Agent(name="researcher", model=your_model)
runner = KitaruRunner(agent, checkpoint_strategy="runner_call")
```

The runtime default is `checkpoint_strategy="calls"` (per-call checkpoints —
see below); pass `"runner_call"` whenever you want a single terminal
checkpoint so `flow.run(...).wait()` returns the run result directly.

You run the agent through `runner.run(...)` or `runner.run_sync(...)` with an
`OpenAIRunRequest`.

## Install

```bash
uv add "kitaru[openai-agents,local]"
```

Then initialize/login as usual:

```bash
kitaru init
kitaru login
kitaru status
```

<Callout type="info">
  Migrating an existing OpenAI Agents SDK project? The
  [`zenml-io/kitaru-skills`](https://github.com/zenml-io/kitaru-skills) package
  includes `/kitaru:kitaru-openai-agents-migration` for moving to `KitaruRunner`,
  choosing `calls` vs `runner_call`, and checking approval/resume state. See
  [Agent Skills](/agent-native/claude-code-skill/).
</Callout>

## Minimal flow

```python
from kitaru import flow
from kitaru.adapters.openai_agents import KitaruRunner, OpenAIRunRequest

runner = KitaruRunner(agent, checkpoint_strategy="runner_call")

@flow
def research(prompt: str) -> str:
    result = runner.run_sync(OpenAIRunRequest.start(prompt))
    return str(result.final_output)
```

## Fresh-run context

OpenAI Agents SDK tools and guardrails often use a local application context:
for example, "which team is this user in?", "which thread is this request part
of?", or "which plugin settings are active?" Pass that object to Kitaru the same
way you pass it to the OpenAI SDK: as a runner-call argument, not as part of the
serializable `OpenAIRunRequest`.

```python
from dataclasses import dataclass
from typing import Any

from agents import RunContextWrapper, function_tool
from kitaru.adapters.openai_agents import KitaruRunner, OpenAIRunRequest

@dataclass(frozen=True)
class WorkerContext:
    team_id: str
    user_id: str
    thread_id: str
    message_id: str
    tool_settings: dict[str, Any]

@function_tool
def lookup_customer(ctx: RunContextWrapper[WorkerContext], customer_id: str) -> str:
    # The context stays local to your Python process. The model only sees what
    # your tool chooses to return.
    return f"team={ctx.context.team_id}, customer={customer_id}"

runner = KitaruRunner(
    agent,
    context_cache_identity=lambda ctx: {
        "team_id": ctx.team_id,
        "user_id": ctx.user_id,
        "thread_id": ctx.thread_id,
        "tool_settings": ctx.tool_settings,
    },
)

result = runner.run_sync(
    OpenAIRunRequest.start("Look up customer 123"),
    context=WorkerContext(
        team_id="team_abc",
        user_id="user_123",
        thread_id="thread_456",
        message_id="msg_this_run_only",
        tool_settings={"include_private_notes": False},
    ),
)
```

A concrete way to think about this: the `OpenAIRunRequest` is the written travel
plan Kitaru can save and replay. `context=` is the live badge the worker carries
while doing the trip. Tools and guardrails can inspect the badge through
`RunContextWrapper.context`, but Kitaru does not save that badge as a visible
artifact or send it to the model automatically. Kitaru still uses the context
identity internally for safe replay, without adding your raw context or projection
to visible tool input artifacts.

Context does matter for safe replay. Imagine two teams both call
`lookup_customer(customer_id="123")`. The visible tool arguments are identical,
but team A and team B may be allowed to see different customer records. Kitaru
therefore includes a context identity in adapter cache keys. If your context is
plain data, Kitaru can derive a structural identity. For production contexts,
prefer `context_cache_identity=` so you can include stable fields such as team,
user, thread, project, plugin, and JSON-primitive `tool_settings`, while
excluding per-run fields such as `message_id`, `trace_id`, or a changing
document cursor. That keeps replay safe without making every new message miss
the cache unnecessarily.

`context=` is different from `metadata=` on `OpenAIRunRequest.start(...)`:
metadata is Kitaru run/checkpoint metadata; context is local OpenAI Agents SDK
runtime state for your tools, guardrails, handoffs, and hooks.

Fresh context is only for new `kind="start"` requests. Interrupted/resumed runs
use the saved OpenAI `RunState`; `context_serializer=` and
`context_deserializer=` on `KitaruRunner` remain the way to serialize and rebuild
context that is already inside an interrupted SDK state. With
`checkpoint_strategy="calls"`, tool checkpoint cache keys use that restored SDK
context identity when it is available, so an approved resumed tool call for team
A does not accidentally reuse a cached tool result from team B. Kitaru uses only
the derived cache key for that separation; it does not save the raw context or
your `context_cache_identity=` projection in visible tool input artifacts.

One more boundary to remember: if your guardrail manually calls raw
`agents.Runner.run(...)` for a nested evaluator, that nested call is not managed
by Kitaru automatically. Wrap the nested evaluator with its own `KitaruRunner` if
you need Kitaru checkpoints there too.

## Checkpoint strategy choices

You choose how Kitaru places checkpoints with `checkpoint_strategy=`.

### `checkpoint_strategy="runner_call"` (recommended for `.wait()`)

Kitaru places one checkpoint around the outer OpenAI `Runner.run(...)` call. That
single checkpoint becomes the flow's terminal artifact, so
`flow.run(...).wait()` returns the run result cleanly. The name
`"runner_call"` is deliberately specific: it means Kitaru is wrapping the outer
OpenAI runner call, not claiming to own every SDK-internal step.

Use this when you want one coarse replay boundary for the whole agent run, or
whenever you want a clean Python value back from `.wait()`.

### `checkpoint_strategy="calls"` (default)

Kitaru catches supported model/tool calls individually as separate peer
checkpoints under the flow.

Use this when you want finer replay units (for example: if call 6 fails, calls
1–5 can come from cache).

Because the per-call checkpoints are siblings under the flow with no single
sink, `flow.run(...).wait()` cannot pick one as "the" return value and raises
`KitaruAmbiguousFlowResultError`. The per-checkpoint artifacts are still
fully visible in the Kitaru UI and retrievable via `KitaruClient` — the
error message points at them. If you need a clean `.wait()` return value,
switch to `checkpoint_strategy="runner_call"`. Wrapping the `runner.run_sync()`
call in your own `@checkpoint` is **not** a workaround here — the adapter
guards against it and will raise, because per-call checkpoints cannot be
nested inside another Kitaru checkpoint.

## Streaming with Kitaru durability

Use `run_stream(...)` / `run_stream_sync(...)` when you want OpenAI Agents SDK
stream updates while the Kitaru checkpoint is still running. This is supported
for `checkpoint_strategy="runner_call"` only.

The concrete story is: OpenAI sends radio chatter while the agent runs. Kitaru
forwards useful pieces of that chatter as live events named
`openai_agents.stream.*`. When OpenAI's stream iterator finishes, Kitaru saves
the same durable `OpenAIRunResult` shape that `run(...)` / `run_sync(...)` save.
The live events help a UI or terminal feel alive; the final `OpenAIRunResult` is
the saved record.

Async flow:

```python
from kitaru import flow
from kitaru.adapters.openai_agents import KitaruRunner, OpenAIRunRequest

runner = KitaruRunner(agent, checkpoint_strategy="runner_call")

@flow
async def research(prompt: str) -> str:
    result = await runner.run_stream(OpenAIRunRequest.start(prompt))
    return str(result.final_output)
```

Sync flow:

```python
from kitaru import flow
from kitaru.adapters.openai_agents import KitaruRunner, OpenAIRunRequest

runner = KitaruRunner(agent, checkpoint_strategy="runner_call")

@flow
def research(prompt: str) -> str:
    result = runner.run_stream_sync(OpenAIRunRequest.start(prompt))
    return str(result.final_output)
```

To watch the live events, submit the flow and read execution events from the
client:

```python
import threading

from kitaru.adapters.openai_agents import (
    OPENAI_STREAM_EVENT_KINDS,
    OPENAI_STREAM_TERMINAL_EVENT_KINDS,
)
from kitaru.client import KitaruClient

handle = research.run("summarize the incident", cache=False)
stop_watching = threading.Event()


def watch_openai_events() -> None:
    try:
        for event in KitaruClient().executions.events(
            handle.exec_id,
            kinds=list(OPENAI_STREAM_EVENT_KINDS),
        ):
            if stop_watching.is_set():
                return
            data = event.payload.get("data", {})
            print(data.get("display", event.kind))
            if event.kind in OPENAI_STREAM_TERMINAL_EVENT_KINDS:
                return
    except Exception as exc:
        print(f"Live event watching unavailable; reading saved result instead: {exc}")


watcher = threading.Thread(target=watch_openai_events, daemon=True)
watcher.start()

result = handle.wait()
stop_watching.set()
watcher.join(timeout=1.0)
```

The watcher is optional and runs in the background so it cannot block the durable
result path. If event streaming is unavailable on the active backend, or if the
watcher sees no live events before it is stopped, `handle.wait()` still reads the
saved result after the flow finishes.

The runnable version lives at
[`examples/integrations/openai_agents_agent/openai_agents_streaming.py`](https://github.com/zenml-io/kitaru/blob/develop/examples/integrations/openai_agents_agent/openai_agents_streaming.py).
It uses a real customer-support agent, submits the flow, prints
`openai_agents.stream.*` events from `KitaruClient().executions.events(...)`, and
then prints the final durable `OpenAIRunResult`.

A few gotchas matter:

- Live events are best effort. Missing stream events do **not** mean the agent
  failed; check the final `OpenAIRunResult`.
- Replay may emit stream events again. If the process crashes after chunk six,
  replay may run the OpenAI stream again and you may see chunks one through six a
  second time.
- Cache hits may emit no OpenAI stream events. Kitaru found the saved final
  runner-call result, so there is no fresh OpenAI stream to forward.
- Kitaru does not save a durable token-by-token transcript in this slice. The
  durable value is the final `OpenAIRunResult`.
- `checkpoint_strategy="calls"` streaming is intentionally unsupported for now.
  Per-call streaming needs a buffering design so replay knows which chunks were
  already emitted.

Streaming also works with approval interruptions. The first streamed run may
finish with `status="interrupted"`; after the approval bridge returns a resume
request, stream the resumed request without passing fresh `context=`:

```python
from kitaru import flow
from kitaru.adapters.openai_agents import (
    KitaruRunner,
    OpenAIRunRequest,
    wait_for_approval,
)

runner = KitaruRunner(agent, checkpoint_strategy="runner_call")

@flow
async def publish_with_gate(prompt: str) -> str:
    result = await runner.run_stream(OpenAIRunRequest.start(prompt))

    if result.status == "interrupted":
        resume_request = wait_for_approval(
            result,
            name="approve_openai_tool",
            timeout=600,
        )
        result = await runner.run_stream(resume_request)

    return str(result.final_output)
```

## Structured outputs, guardrails, and nested agents

OpenAI Agents SDK structured outputs work through the adapter. If your agent is
created with `Agent(output_type=...)`, Kitaru preserves the SDK result object and
its typed `final_output` in both supported strategies:

- `checkpoint_strategy="runner_call"` records the outer runner call and returns
  the structured result from `.wait()` cleanly.
- `checkpoint_strategy="calls"` records supported model and tool calls
  individually, while the SDK still produces the typed final output for your
  Python code.

For tool-input guardrails, use `checkpoint_strategy="calls"` when you need to
see blocked tool attempts. In that strategy, Kitaru records a rejected tool
attempt as an existing `tool_call` event with guardrail metadata before the tool
function runs. It does not create a new event type, and it does not save a tool
checkpoint for arguments that the guardrail rejected.

Privacy follows the capture policy here too. If `save_input=False`, Kitaru omits
raw tool input artifacts and also redacts guardrail rejection messages and
unexpected guardrail exception details from persisted event metadata, because
those strings may repeat the user/tool input the guardrail just inspected. The
event still shows that a guardrail blocked the call, which guardrail did it, and
whether the behavior was `reject_content`, `raise_exception`, or an exception.

`checkpoint_strategy="runner_call"` still only sees the outer `Runner.run(...)`
boundary. That is useful for a single durable result, but it cannot show each
individual tool guardrail decision. Choose `"calls"` when per-tool guardrail
observability matters.

One more boundary to remember: raw nested `agents.Runner.run(...)` calls remain
outside Kitaru unless you wrap that evaluator agent with its own `KitaruRunner`.
Raw nested agents are fine for quick ephemeral checks. If their inputs, outputs,
or guardrail decisions need Kitaru observability, run them through
`KitaruRunner` too.

## Important guardrail

`checkpoint_strategy="calls"` must run from flow scope (not from inside another
`@checkpoint`), because the adapter needs room to open inner checkpoints for
model/tool calls.

## Approval interruptions

The adapter preserves OpenAI Agents SDK runs that stop for human approval. In story form: the agent reaches a tool approval, the SDK returns an interrupted run, Kitaru stores the serialized run state, and your flow can turn that interruption into a normal durable `kitaru.wait()`.

```python
from kitaru import flow
from kitaru.adapters.openai_agents import (
    KitaruRunner,
    OpenAIRunRequest,
    wait_for_approval,
)

runner = KitaruRunner(agent, checkpoint_strategy="runner_call")

@flow
def publish_with_gate(prompt: str) -> str:
    result = runner.run_sync(OpenAIRunRequest.start(prompt))

    if result.status == "interrupted":
        resume_request = wait_for_approval(
            result,
            name="approve_openai_tool",
            timeout=600,
        )
        result = runner.run_sync(resume_request)

    return str(result.final_output)
```

`wait_for_approval(...)` asks Kitaru to wait for a boolean approval and then returns an `OpenAIRunRequest.resume(...)` object. Approving resumes the saved OpenAI run state; rejecting sends the SDK a rejection message. If you already collected the decision somewhere else, use `build_resume_request(result, approve=True)` or pass an explicit `OpenAIApprovalDecision` to `OpenAIRunRequest.resume(...)`.

Keep this bridge at flow scope. If you put it inside a Kitaru checkpoint, the flow would be trying to pause from inside a step that is meant to finish or fail as one unit.

## Capture and checkpoint configuration

`KitaruRunner` exposes the same two kinds of knobs most teams need in production:

- **Capture policy:** what gets saved for observability.
- **Checkpoint policy:** how retries and dashboard grouping apply to adapter-created checkpoints.

```python
from kitaru.adapters.openai_agents import KitaruRunner, OpenAICapturePolicy

runner = KitaruRunner(
    agent,
    checkpoint_strategy="calls",
    capture=OpenAICapturePolicy(
        save_input=False,              # privacy: do not persist full user input
        save_final_output=True,
        save_run_state=True,           # needed for approval resume
        save_interruption_payloads=True,
        save_response_items=False,      # opt in only when you need raw SDK items
        save_usage=True,
    ),
    model_checkpoint_config={"retries": 2},
    tool_checkpoint_config={"retries": 1},
    tool_checkpoint_config_by_name={
        "charge_card": False,          # do not checkpoint side-effectful tool
        "search_docs": {"retries": 3},
    },
)
```

`OpenAICapturePolicy` defaults are designed for useful traces: child events, input, final output, run state, interruption payloads, usage, and OTel correlation are on; raw response items are off by default because they can be noisy.

Two privacy switches are worth calling out:

- `save_input=False` keeps raw model/tool inputs out of artifacts and redacts
  tool-input guardrail messages or exception text that may contain those inputs.
- `save_interruption_payloads=False` keeps approval interruption summaries
  usable for resume decisions — index, kind, tool name, call ID, and message when
  the SDK exposes them — but omits raw `arguments` and `arguments_preview`.

Checkpoint config accepts `retries`, `type`, and `runtime`. `runtime="isolated"` is rejected for adapter-managed checkpoints today because those synthetic checkpoint closures capture live OpenAI SDK objects; use inline runtime or omit `runtime`.

For interrupted OpenAI runs, the SDK stores its own `RunState` so the run can resume later. If that saved `RunState` contains context objects that are not JSON-serializable, pass `context_serializer=` and `context_deserializer=` to `KitaruRunner`. These hooks are for serializing resume state after an interruption; they do not control the fresh-run `context=` object you pass when starting a new run. By default `strict_context=True`, so Kitaru fails loudly instead of saving a resume state that cannot be reconstructed later.

## Runnable example

This example uses the real OpenAI API (not a stub model), so set your key:

```bash
uv sync --extra local --extra openai-agents
export OPENAI_API_KEY='OPENAI_API_KEY_VALUE'
# default model in the example is gpt-5-nano
# optional override: any OpenAI model you have access to
# export OPENAI_AGENTS_MODEL='<another-openai-model>'
uv run python examples/integrations/openai_agents_agent/openai_agents_adapter.py

# streaming runner-call example
uv run python examples/integrations/openai_agents_agent/openai_agents_streaming.py
```

## End-to-end research bot example

For a larger example, run the OpenAI research bot:

```bash
cd examples/end_to_end/openai_research_bot
uv sync --extra local --extra openai-agents
uv run kitaru init
export OPENAI_API_KEY='OPENAI_API_KEY_VALUE'
uv run python research_bot.py "AI agent durability" --max-searches 2
```

The workflow keeps the original research-bot shape:

```text
planner → submitted search fan-out → writer report
```

The planner and writer run at flow scope through `KitaruRunner` with
`checkpoint_strategy="runner_call"`. The planned searches fan out with
`run_search_item.submit(...)`, so each search is its own durable checkpoint.
Ordinary Kitaru checkpoints publish stable dashboard artifacts such as the
normalized research plan, search summaries, and final report.

The example also uses a local OpenAI Agents SDK `@function_tool` named
`search_web` instead of the hosted `WebSearchTool`. The local tool calls the
OpenAI Responses API with `web_search`, which makes the checkpoint trace clearer
with the adapter's current public behavior.

Look for these artifacts in the Kitaru UI:

- `research_plan`
- `search_summaries`
- `durability_drill`
- `final_report`
- `research_report_metadata`

To test the durable-retry story directly, set
`KITARU_RESEARCH_BOT_FAIL_AFTER_SEARCHES=1` before running the example. It will
fail after the submitted searches complete. Unset the flag and run
`kitaru executions replay <EXECUTION_ID> --from durability_drill_gate`; the
replay should reuse the completed planner/search checkpoints and continue into
the writer. `retry` tries to restart the same failed execution and may be
unavailable on server-backed stacks after a run has concluded.

See also: [Replay and overrides](/guides/replay-and-overrides).
