Use ReasonBlocks with OpenAI Agents

ReasonBlocks integrates with the OpenAI Agents SDK two ways depending on how much of the pipeline you want:

rb.openai_model(default_model, ...) — wraps any openai-agents Model so the agent runs FSM step scoring, server-side monitor steering, E-trace injection, and (with a model_factory) model routing on every get_response call. This is the parity path with the LangChain middleware.
rb.openai_hooks(...) — telemetry-only RunHooks adapter. Use this when you want dashboard rows + token / tool capture but don’t need steering. Cheaper and stays out of the model path entirely.

The make_openai_tools factory for CodebaseMemory + ImportGraph works alongside either entry point.

Path 1: full steering with `rb.openai_model`

Install

pip install reasonblocks openai-agents

Build a wrapped Model

Pass an underlying Model instance (typically OpenAIChatCompletionsModel or OpenAIResponsesModel) and an optional model_factory to enable FSM-driven model routing.

import os
from agents import Agent, Runner, OpenAIChatCompletionsModel
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key=os.environ["REASONBLOCKS_API_KEY"],
    model_routing={
        "FAST": "gpt-4o-mini",
        "SLOW": "gpt-4o",
    },
)

wrapped = rb.openai_model(
    default_model=OpenAIChatCompletionsModel("gpt-4o"),
    model_factory=lambda mid: OpenAIChatCompletionsModel(mid),
    agent_name="reviewer",
    task="describe what this run is doing",
    model="gpt-4o",
)

Without model_factory, routing overrides log to the step entry but the wrapped default model is always used. With it, the wrapper builds and caches alternates lazily.

Use as the agent's model + run

wrapped is itself a sync and async context manager that flushes the live-telemetry emitter on exit.

Async
Sync

import asyncio

async def main():
    agent = Agent(
        name="reviewer",
        instructions="You are a senior code reviewer.",
        model=wrapped,
        tools=[...],
    )

    async with wrapped:
        result = await Runner.run(agent, input="Review PR #42")

    return result

asyncio.run(main())

agent = Agent(name="reviewer", instructions="...", model=wrapped, tools=[...])

with wrapped:
    result = Runner.run_sync(agent, input="Review PR #42")

Inspect the step log

The wrapper exposes its SteeringSession. Read wrapped.session.step_log after the run for per-step difficulty, FSM state, monitors fired, injection text, model id used, tokens, and latency.

for entry in wrapped.session.step_log:
    print(entry.as_dict())

Streaming via stream_response is currently a pass-through to the wrapped model — no steering is applied to streamed responses in the first cut. Non-streaming get_response calls run the full pipeline.

Path 2: telemetry-only with `rb.openai_hooks`

Use this when you want dashboard rows but don’t want the wrapper sitting in front of model calls.

Install

pip install reasonblocks openai-agents

Initialize ReasonBlocks

import os
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key=os.environ["REASONBLOCKS_API_KEY"])

Build hooks and run the agent

rb.openai_hooks(...) returns a ReasonBlocksHooks instance — a RunHooks subclass that is also a sync and async context manager. Pass it to Runner.run (or Runner.run_sync).

Async
Sync

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(
        name="reviewer",
        instructions="You are a senior code reviewer.",
        tools=[...],
    )

    hooks = rb.openai_hooks(
        run_id="pr-42",
        agent_name="reviewer",
        task="review PR #42",
        model="gpt-4o",
    )

    async with hooks:
        result = await Runner.run(agent, input="Review PR #42", hooks=hooks)

    return result

asyncio.run(main())

from agents import Agent, Runner

agent = Agent(name="reviewer", instructions="...", tools=[...])

hooks = rb.openai_hooks(agent_name="reviewer", task="review PR #42")

with hooks:
    result = Runner.run_sync(agent, input="Review PR #42", hooks=hooks)

Tag runs for the dashboard

rb.openai_hooks(...) accepts the same identifying fields as rb.middleware().

hooks = rb.openai_hooks(
    run_id="pr-42",                              # auto-generated if omitted
    agent_name="pr-reviewer",
    task="review PR #42",
    framework="openai-agents",                   # default
    model="gpt-4o",
    codebase_id="my-org/my-repo",
    org_id="6d3f...",                            # "default" if omitted
    project_id="a91b...",                        # "default" if omitted
    metadata={"pr_number": 42, "base_branch": "main"},
)

Track failures

Exceptions that escape the with/async with block are recorded as failure: <ExceptionType>. To mark an explicit logical failure when the agent returns normally, call hooks.mark_failure(reason=...) before exit:

async with hooks:
    result = await Runner.run(agent, input=task, hooks=hooks)
    if not outcome_is_valid(result):
        hooks.mark_failure(reason="invalid_output")

What gets captured

The hooks subscribe to the openai-agents lifecycle:

on_agent_start emits run_start once.
on_llm_end accumulates input + output tokens from each LLM response.
on_tool_start stamps a per-tool start time.
on_tool_end emits a step event with the tool name, observation (truncated to 8000 chars), latency, and accumulated tokens. Tokens are attributed to the first tool call after each LLM response; subsequent calls in the same response stamp 0. The total across the run is exact.
on_agent_end emits run_finish with outcome="success" (overridable via mark_failure).
__exit__ / __aexit__ flushes a failure outcome if an exception escaped, then closes the emitter.

Add CodebaseMemory tools

make_openai_tools returns function_tool-decorated callables ready for Agent(tools=[...]). Same contract as the LangChain factory — only the framework decoration differs.

from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.openai_agents import make_openai_tools

memory = CodebaseMemory(
    codebase_id="my-org/my-repo",
    api_key=os.environ["REASONBLOCKS_API_KEY"],
)
graph = ImportGraph().build_from_files({path: open(path).read() for path in py_files})

rb_tools = make_openai_tools(
    memory,
    graph,
    recall_top_k=5,
    recall_threshold=0.25,
    enable_recall=True,
    enable_store=True,
    enable_impact=True,
)

agent = Agent(
    name="reviewer",
    instructions="Call recall_findings before reading any file.",
    tools=[*rb_tools, *your_tools],
)

The factory adds up to three tools: recall_findings, store_finding, and (when graph is provided) impact_analysis. See CodebaseMemory for the storage API.

ImportGraph.build_from_files requires networkx. Install with pip install networkx.

Complete example

import asyncio
import os
from agents import Agent, Runner
from reasonblocks import CodebaseMemory, ReasonBlocks
from reasonblocks.integrations.openai_agents import make_openai_tools

rb = ReasonBlocks(api_key=os.environ["REASONBLOCKS_API_KEY"])

memory = CodebaseMemory(
    codebase_id="my-org/my-repo",
    api_key=os.environ["REASONBLOCKS_API_KEY"],
)
rb_tools = make_openai_tools(memory)

agent = Agent(
    name="pr-reviewer",
    instructions=(
        "You are a senior code reviewer. "
        "Call recall_findings before reading any file. "
        "Store key findings before you finish."
    ),
    tools=[*rb_tools],
)

async def main():
    hooks = rb.openai_hooks(
        agent_name="pr-reviewer",
        task="review PR #42",
        codebase_id="my-org/my-repo",
        model="gpt-4o",
    )

    async with hooks:
        result = await Runner.run(
            agent,
            input="Review the changes in PR #42 and flag any regressions.",
            hooks=hooks,
        )

    print(result.final_output)

asyncio.run(main())

Each hooks object — and each wrapped from rb.openai_model(...) — is single-use. Call the factory again for the next run.

Which to pick

You want…	Use
Steering + dashboard rows	`rb.openai_model(...)`
Dashboard rows only (no model wrapping)	`rb.openai_hooks(...)`
Codebase memory tools (either path)	`make_openai_tools(memory, graph)`
Model routing across FSM states	`rb.openai_model(model_factory=...)`

If you adopt rb.openai_model, you don’t also need rb.openai_hooks — the wrapper emits the same run_start / step / run_finish telemetry from inside the steering pipeline.

Getting Started

Concepts

Using ReasonBlocks

Connectors and sync

Use ReasonBlocks with OpenAI Agents

Path 1: full steering with `rb.openai_model`

Path 2: telemetry-only with `rb.openai_hooks`

What gets captured

Add CodebaseMemory tools

Complete example

Which to pick

​Path 1: full steering with rb.openai_model

​Path 2: telemetry-only with rb.openai_hooks

​What gets captured

​Add CodebaseMemory tools

​Complete example

​Which to pick

Path 1: full steering with `rb.openai_model`

Path 2: telemetry-only with `rb.openai_hooks`

What gets captured

Add CodebaseMemory tools

Complete example

Which to pick