rb.middleware() returns a ReasonBlocksMiddleware — a langchain.agents.middleware.AgentMiddleware subclass that also acts as a context manager. You attach it to create_agent like any other middleware. It hooks before_agent, before_model, wrap_model_call, and after_agent to score each step, evaluate trajectory monitors server-side, inject steering text into the system message, and (optionally) route the model based on the FSM state.
Install
pip install reasonblocks "langchain>=1.0" langchain-anthropic langgraph
Initialize ReasonBlocks
Create one ReasonBlocks per process and call rb.middleware() per run. Each middleware instance is single-use.import os
from reasonblocks import ReasonBlocks
rb = ReasonBlocks(
api_key=os.environ["REASONBLOCKS_API_KEY"],
base_url=os.environ.get("REASONBLOCKS_BASE_URL"), # omit for the hosted API
)
The constructor does not auto-read REASONBLOCKS_API_KEY — your code must pass it explicitly. The base_url default is read from REASONBLOCKS_BASE_URL at import time by reasonblocks._settings; if you set that env var before importing reasonblocks, you can omit base_url here.
Wrap your agent in the context manager
The recommended pattern is with mw:. The context manager guarantees that the run_finish telemetry event fires on success, exception, or explicit failure marking.from langchain.agents import create_agent
mw = rb.middleware(
agent_name="bugfixer",
task="fix the TypeError in handler.py",
framework="langchain",
model="anthropic:claude-haiku-4-5-20251001",
codebase_id="my-org/my-repo",
)
agent = create_agent(
model="anthropic:claude-haiku-4-5-20251001",
tools=[search_codebase, read_file, edit_file, run_tests],
system_prompt="You are a senior software engineer. Fix bugs by searching, reading, editing, and testing.",
middleware=[mw],
)
with mw:
result = agent.invoke(
{"messages": [("user", "There's a TypeError in the request handler. Find and fix it.")]},
)
Tag runs for the dashboard
Every parameter on rb.middleware() is optional. Anything not consumed by a named field rides along in metadata on the run row.mw = rb.middleware(
run_id="pr-42-attempt-3", # auto-generated UUID if omitted
agent_name="bugfixer",
task="fix the TypeError in handler.py",
framework="langchain", # default
model="anthropic:claude-haiku-4-5-20251001", # display only
codebase_id="my-org/my-repo",
org_id="6d3f...", # "default" if omitted
project_id="a91b...", # "default" if omitted
metadata={"pr_number": 42, "experiment": "v2"},
)
When your api_key is a per-org rb_live_* key, the API overrides org_id / project_id with the key’s authoritative scope. Leaving them at "default" is fine.
Mark explicit failures
If the agent returns normally but the run was logically a failure (wrong answer, tests still failing), call mark_failure(reason=...) before the with block exits. Exceptions that escape the block are recorded automatically as failure: <ExceptionType>.with mw:
result = agent.invoke({"messages": [("user", "Fix the bug.")]})
if not tests_pass(result):
mw.mark_failure(reason="tests_still_failing")
Inspect step_log
mw.step_log is a list[StepLogEntry] populated as the agent runs. Each entry holds the FSM state, difficulty, resolved model id (when routed), monitors that fired, full intervention text, tool calls, tokens, and latency.for entry in mw.step_log:
print(
f"step {entry.step}: state={entry.fsm_state} "
f"difficulty={entry.difficulty} model_id={entry.model_id} "
f"tokens={entry.tokens}"
)
if entry.monitors_fired:
print(f" monitors fired: {entry.monitors_fired}")
for src, text in zip(entry.injection_sources, entry.intervention_texts):
print(f" [{src}] {text[:160]}")
entry.as_dict() for a JSON-shaped view.make_langchain_tools wraps a CodebaseMemory (and optionally an ImportGraph) into @tool-decorated callables. The agent uses them to recall prior findings before re-reading files and to persist new findings during a run.
from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.langchain_tools import make_langchain_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_langchain_tools(
memory,
graph,
recall_top_k=5,
recall_threshold=0.25,
enable_recall=True,
enable_store=True,
enable_impact=True,
)
agent = create_agent(
model="anthropic:claude-haiku-4-5-20251001",
tools=[*rb_tools, search_codebase, read_file, edit_file, run_tests],
system_prompt="Call recall_findings before reading any file.",
middleware=[mw],
)
recall_findings — semantic search over prior findings.store_finding — persist a new finding for future runs (hidden when enable_store=False).impact_analysis — blast-radius lookup via ImportGraph (only added when a graph is passed).
ImportGraph.build_from_files requires networkx. Install with pip install networkx.
Complete example
import os
from langchain.agents import create_agent
from langchain_core.tools import tool
from reasonblocks import CodebaseMemory, ReasonBlocks
from reasonblocks.integrations.langchain_tools import make_langchain_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_langchain_tools(memory)
@tool
def search_codebase(query: str) -> str:
"""Search the codebase for files matching a query."""
...
@tool
def read_file(path: str) -> str:
"""Read the contents of a file."""
...
mw = rb.middleware(
agent_name="bugfixer",
task="fix the TypeError in handler.py",
codebase_id="my-org/my-repo",
)
agent = create_agent(
model="anthropic:claude-haiku-4-5-20251001",
tools=[*rb_tools, search_codebase, read_file],
system_prompt="You are a senior software engineer. Fix bugs methodically.",
middleware=[mw],
)
with mw:
result = agent.invoke({"messages": [("user", "Fix the bug.")]})
Each mw is single-use — call rb.middleware(...) again for the next run.
