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",
)
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.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).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.
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],
)
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=...) |
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. 
