Skip to main content
ReasonBlocks is the main entry point for the SDK. Create one instance per process, passing your API key and any global configuration, then call middleware() or openai_hooks() once per agent run to get a fresh, run-scoped object.

​
Constructor

from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key="rb_live_...",
    token_budget=50_000,
    task_profile="pr_review",
)
​
api_key
string
required
Your ReasonBlocks API key. Production keys start with rb_live_; test keys with rb_test_. When you use a per-customer live key, the server overrides org_id and project_id with the key’s authoritative scope automatically.
​
base_url
string
Override the API base URL. Set this when pointing the SDK at a self-hosted deployment or a staging environment. When omitted, the SDK connects to the ReasonBlocks hosted API.
​
token_budget
integer
Soft cap on tokens tracked by the trace state manager. Stored on TraceState.token_budget and used by downstream consumers that surface budget pressure. Leave unset to run without a cap.
​
fsm_thresholds
object
Override the difficulty FSM’s transition thresholds. Forwarded as kwargs to DifficultyFSM; recognized keys include fast_threshold, slow_threshold, skip_threshold, hysteresis_margin, fast_window, slow_window, skip_window. Leave unset to use the defaults.
​
model_routing
object
Map FSM state names to model identifiers, e.g. {"FAST": "anthropic:claude-haiku-4-5-20251001", "SLOW": "anthropic:claude-sonnet-4-6"}. The middleware swaps the model on the next call when the FSM enters that state.
​
e_traces_enabled
boolean
default:"True"
Whether to retrieve and inject E-trace patterns from the server. Set to False to skip the retrieval round-trip — useful in local development or CI environments without rb-api access.
​
monitor_names
list[str] | None
default:"None"
Accepted for backward compatibility but currently a no-op on the middleware path — the monitor suite runs server-side. Custom local monitors require a custom injection.
​
live_streaming_enabled
boolean
default:"True"
Whether to stream run_start, per-step, and run_finish events to rb-api as the agent runs. Disable for fully offline or air-gapped deployments. When False, no MonitorClient or StreamingEmitter is constructed and the dashboard will not see this run.
​
task_profile
string
default:"\"coding\""
Selects the monitor weight vector on the server. Built-in profiles are "coding" (the default), "pr_review" (weights semantic_loop highest for read-only review agents), and "qa" (weights claim_contradiction and silent_topic_drift highest for question-answering agents). See Monitor profiles for the per-profile weight tables. The profile is stamped on every run_start event and persisted on the run row.

​
middleware()

Creates a fresh ReasonBlocksMiddleware instance scoped to a single agent run. Call this once per run, not once per step. 
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...")

with rb.middleware(
    run_id="run-abc-123",
    agent_name="pr-reviewer",
    task="Review PR #42 for correctness",
    framework="langchain",
    model="claude-sonnet-4-6",
    codebase_id="myorg_myrepo",
    metadata={"pr_number": 42, "experiment": "v2"},
) as mw:
    agent = create_agent(
        model=ChatAnthropic(model="claude-sonnet-4-6", max_tokens=2048),
        tools=[...],
        system_prompt="...",
        middleware=[mw],
    )
    result = agent.invoke({"messages": [("user", "Review PR #42")]})
​
run_id
string
Unique identifier for this run. When omitted, the SDK generates one from the trace state manager. Provide your own when you want to correlate the dashboard row with an external ID (CI job, PR number, etc.).
​
agent_name
string
default:"\"\""
Human-readable name for the agent. Displayed in the dashboard run list.
​
task
string
default:"\"\""
Short description of what this run is trying to accomplish. Stored on the run row.
​
framework
string
default:"\"langchain\""
Agent framework identifier. Stored for diagnostic filtering.
​
model
string
default:"\"\""
Model identifier used by the agent. Stored on the run row.
​
codebase_id
string
default:"\"\""
Associates this run with a specific codebase in the findings store. Use the same value you pass to CodebaseMemory.
​
org_id
string
default:"\"default\""
Multi-tenant organization scope. When the api_key is a per-customer live key, rb-api overrides this with the key’s authoritative org.
​
project_id
string
default:"\"default\""
Multi-tenant project scope within org_id. Same override behavior as org_id.
​
metadata
object
Free-form key–value tags attached to the run row and surfaced as JSON in the dashboard.

​
Returns

​
ReasonBlocksMiddleware
middleware
A LangChain AgentMiddleware subclass. Also a context manager (with rb.middleware(...) as mw:) that emits run_finish and closes the streaming emitter on exit. Inside an __exit__ an exception sets the outcome to failure: <ExceptionName>; a clean exit defaults to success. Call mw.mark_failure(reason="...") before the context exits to override the default. Call mw.close(timeout=5.0) to drain the emitter explicitly outside a with block.

​
ab_middleware()

Builds a run-scoped middleware for an A/B evaluation. Flips a deterministic coin (assign_arm) and returns either the full pipeline (on) or a vanilla passthrough control (off). Both arms force live telemetry on and stamp experiment_id / arm / assignment_unit / rb_version onto the run. See the A/B evaluation guide for the end-to-end flow and how to read the report. 
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...")

mw = rb.ab_middleware(
    experiment_id="cust-acme-q2",
    unit_id="ticket-1843",      # stable -> retries stay in the same arm
    on_fraction=0.5,
    agent_name="reviewer",
    task="Review PR #42",
    org_id="acme",
)
# mw.arm is recorded on the run; use mw with create_agent(..., middleware=[mw])
​
experiment_id
string
required
Identifier shared by every run in the experiment. Stamped on the run and used as the report key (GET /v1/monitor/experiments/{experiment_id}/report). Immutable on the run once set.
​
unit_id
string
Randomization unit. Hashed with experiment_id to pick the arm, so the same unit always lands in the same arm. Pass a stable task/ticket id if you want retries to stay in their arm; defaults to the run id (each attempt drawn independently).
​
on_fraction
float
default:"0.5"
Probability of the on arm, in [0, 1]. Also passed to the report’s SRM check as the expected split.
The remaining keyword arguments mirror middleware() (run_id, agent_name, task, framework, model, codebase_id, org_id, project_id, metadata). Experiment tags are stamped last, so a caller’s metadata cannot override the arm.

​
Returns

​
ReasonBlocksMiddleware
middleware
A run-scoped middleware, same type and lifecycle as middleware(). For the off arm it runs in passthrough mode — scoring and telemetry only, with the model request left untouched. The chosen arm is on mw._run_metadata["arm"].

​
claude_messages_session()

Builds a SteeringSession wired for the Anthropic Messages API loop. Pair with run_messages_agent_loop(..., session=session) to run the full pipeline at every turn. 
from anthropic import Anthropic
from reasonblocks import ReasonBlocks
from reasonblocks.integrations.claude_tools import (
    make_claude_tools, run_messages_agent_loop,
)

rb = ReasonBlocks(api_key="rb_live_...")
client = Anthropic()
tool_specs, dispatch = make_claude_tools(memory)

with rb.claude_messages_session(
    agent_name="reviewer", task="review PR #42",
    model="claude-haiku-4-5-20251001",
) as session:
    outcome = run_messages_agent_loop(
        client,
        model="claude-haiku-4-5-20251001",
        messages=[{"role": "user", "content": "..."}],
        tool_specs=tool_specs, dispatch=dispatch,
        session=session,
    )
Same parameters as middleware() (run_id, agent_name, task, framework, model, codebase_id, org_id, project_id, metadata). framework defaults to "claude-messages".

​
Returns

​
SteeringSession
A SteeringSession instance — the framework-agnostic core. See the SteeringSession reference for its full API. Sync + async context manager; emits run_start on enter, run_finish on exit (with the same outcome semantics as middleware()).

​
openai_model()

Wraps an openai-agents Model so the steering pipeline runs before each get_response call. This is the parity path with middleware() for the OpenAI Agents SDK — FSM scoring, server-side monitor evaluation, E1 / E2 / E3 injection, and (with a model_factory) model routing all run. 
from agents import Agent, Runner, OpenAIChatCompletionsModel
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...", 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",
)

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

with wrapped:
    await Runner.run(agent, input="...")
​
default_model
Model
required
Any object satisfying the openai-agents Model protocol — typically OpenAIChatCompletionsModel or OpenAIResponsesModel.
​
model_factory
Callable[[str], Model]
default:"None"
Optional callable that builds alternate Model instances on demand when model_routing on the client maps the current FSM state to a different identifier. Without a factory, routing overrides log on the step entry but the wrapped default model is still used.
The remaining keyword arguments mirror middleware() (run_id, agent_name, task, framework, model, codebase_id, org_id, project_id, metadata). framework defaults to "openai-agents".

​
Returns

​
ReasonBlocksModel
A Model-compatible wrapper that is also a sync + async context manager. Use as the model field on an Agent. The underlying SteeringSession is exposed via wrapped.session for inspecting step_log / calling mark_failure.
Streaming via stream_response is currently a pass-through to the wrapped model. Non-streaming get_response calls run the full pipeline.

​
claude_agent_telemetry()

Builds a telemetry-only adapter for the claude-agent-sdk query() stream. The Claude Agent SDK runs the agent loop inside the Claude Code CLI process, so the steering pipeline cannot inject — this adapter parses the message stream for tool_use / tool_result blocks and emits run_start / per-tool step / run_finish events to the dashboard. 
from claude_agent_sdk import query
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...")

async with rb.claude_agent_telemetry(
    agent_name="reviewer", task="...", model="claude-haiku-4-5-20251001",
) as tele:
    async for msg in tele.wrap(query(prompt="...", options={"tools": tools})):
        ...
Parameters mirror middleware(). framework defaults to "claude-agent-sdk".

​
Returns

​
ClaudeAgentTelemetry
Sync + async context manager exposing wrap(async_iter) to drive an async query stream, plus mark_failure(reason=...) / close(timeout=...). No FSM, no monitor evaluation, no E-trace retrieval — purely an observer.

​
openai_hooks()

Builds a RunHooks adapter for the openai-agents SDK. Telemetry-only — emits the same run_start / step / run_finish events as the steering integrations, but does not run scoring, monitors, or injection. Use this when you want dashboard rows but don’t want a Model wrapper sitting in front of model calls; use openai_model() for the full pipeline.
If you adopt openai_model, you don’t also need openai_hooks — the wrapper emits the same telemetry from inside the steering pipeline.
from agents import Agent, Runner
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...")

hooks = rb.openai_hooks(
    run_id="run-oai-1",
    agent_name="reviewer",
    task="Review PR #42",
)

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

with hooks:
    result = Runner.run_sync(agent, input="Review this PR", hooks=hooks)
openai_hooks() accepts the same keyword parameters as middleware(). The only behavioral difference is that framework defaults to "openai-agents" instead of "langchain".

​
Returns

​
RunHooks
hooks
A RunHooks-compatible object that is also a sync + async context manager. Exposes mark_failure(reason=...) and close(timeout=...) with the same semantics as the LangChain middleware. The hooks emit run_start on on_agent_start, a per-tool step event on on_tool_end, and run_finish on on_agent_end (or via __exit__ when an exception escapes).

​
score_step()

@staticmethod. Heuristic difficulty scorer for a single step’s text. The middleware uses this internally; you can call it directly for debugging or testing. 
score = ReasonBlocks.score_step(
    "I'm not sure — this could be a ValidationError or a TypeError"
)
print(score)  # e.g. 0.73
​
text
string
required
The agent step text to score. An empty string returns 0.5.
​
score
float
A value between 0 and 1. Higher scores mean the step looks harder — more hedging, longer text, more error language, more entities.
The score is a weighted combination of four signals, passed through a sigmoid centered at raw=0.5:
SignalWeightDescription
Hedging density0.30Frequency of uncertainty words such as "maybe", "not sure", "hmm", "reconsider"
Length0.25Word count, capped at 500
Error language0.25Frequency of words like "error", "exception", "traceback", "failed"
Entity density0.20Count of file paths, dotted identifiers, and quoted strings (capped at 10)
The raw weighted sum is mapped via 1 / (1 + exp(-6 * (raw - 0.5))) to keep the output in a usable range.