ReasonBlocks

https://docs.reasonblocks.com/
Admin
ReasonBlocks is a drop-in Python SDK for production AI agents that provides live observability,...
Tokens:57,208
Snippets:646
Trust Score:5.4
Update:1 month ago
Show doc for...
Tokens:
Raw
### Installing dependencies

Source: https://docs.reasonblocks.com/api-reference/integrations/langchain-tools

Install the required packages for the LangChain integration.

```bash
pip install reasonblocks langchain-core
```

--------------------------------

### Install ReasonBlocks and Anthropic SDK

Source: https://docs.reasonblocks.com/guides/claude-messages

Install the necessary packages via pip.

```bash
pip install reasonblocks anthropic
```

--------------------------------

### Install reasonblocks and openai-agents

Source: https://docs.reasonblocks.com/api-reference/integrations/openai-tools

Install the necessary packages via pip.

```bash
pip install reasonblocks openai-agents
```

--------------------------------

### Initialize LangChain Tools

Source: https://docs.reasonblocks.com/api-reference/integrations/langchain-tools

Examples for creating LangChain tools with varying configurations for memory, import graphs, and storage.

```python
import pathlib
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

from reasonblocks import CodebaseMemory, ImportGraph, ReasonBlocks
from reasonblocks.integrations import make_langchain_tools

rb = ReasonBlocks(api_key="rb_live_...")
memory = CodebaseMemory(codebase_id="my-repo", api_key="rb_live_...")
graph = ImportGraph().build_from_files(
    {str(p): p.read_text() for p in pathlib.Path("myrepo").rglob("*.py")}
)

rb_tools = make_langchain_tools(
    memory,
    graph,
    recall_top_k=8,
    recall_threshold=0.3,
)

model = ChatAnthropic(model="claude-haiku-4-5-20251001", max_tokens=1024)

with rb.middleware(
    run_id="run-1",
    agent_name="reviewer",
    task="Review PR #42",
    model="claude-haiku-4-5-20251001",
    codebase_id="my-repo",
) as mw:
    agent = create_agent(
        model=model,
        tools=[*rb_tools, *your_tools],
        system_prompt="You are a senior Python reviewer.",
        middleware=[mw],
    )
    result = agent.invoke({"messages": [("user", "Review PR #42")]})
```

```python
rb_tools = make_langchain_tools(
    memory,
    graph,
    enable_store=False,
)
```

```python
rb_tools = make_langchain_tools(memory)
# impact_analysis is omitted — graph is None
```

--------------------------------

### Initialize and use OpenAI tools

Source: https://docs.reasonblocks.com/api-reference/integrations/openai-tools

Demonstrates the full setup of ReasonBlocks tools with an OpenAI agent, including memory and graph initialization.

```python
import asyncio
import pathlib
from agents import Agent, Runner
from reasonblocks import CodebaseMemory, ImportGraph, ReasonBlocks
from reasonblocks.integrations import make_openai_tools

rb = ReasonBlocks(api_key="rb_live_...")
memory = CodebaseMemory(codebase_id="my-repo", api_key="rb_live_...")
graph = ImportGraph().build_from_files(
    {str(p): p.read_text() for p in pathlib.Path("myrepo").rglob("*.py")}
)

rb_tools = make_openai_tools(memory, graph)

agent = Agent(
    name="code-reviewer",
    instructions="You are a senior engineer reviewing Python codebases.",
    tools=[*rb_tools, *your_tools],
)

hooks = rb.openai_hooks(
    run_id="run-1",
    agent_name="code-reviewer",
    task="Review auth module for security issues",
)

async def main():
    async with hooks:
        result = await Runner.run(
            agent,
            input="Review auth/session.py for security issues",
            hooks=hooks,
        )
    print(result.final_output)

asyncio.run(main())
```

--------------------------------

### Install ReasonBlocks SDK

Source: https://docs.reasonblocks.com/quickstart

Install the required package via pip. Python 3.10 or later is required.

```bash
pip install reasonblocks
```

--------------------------------

### Install specific version of ReasonBlocks

Source: https://docs.reasonblocks.com/installation

Command to pin the installation to a specific package version.

```bash
pip install "reasonblocks==0.1.0"
```

--------------------------------

### Install development dependencies

Source: https://docs.reasonblocks.com/installation

Install the package with development-specific dependencies.

```bash
pip install "reasonblocks[dev]"
```

--------------------------------

### Initialize ReasonBlocks API and Middleware

Source: https://docs.reasonblocks.com/configuration/reasonblocks-config

Demonstrates the full setup of the ReasonBlocks API client, configuration object, and middleware integration.

```python
from reasonblocks import (
    ReasonBlocksAPI, ReasonBlocksConfig, build_middleware,
)
from reasonblocks.fsm import DifficultyFSM
from reasonblocks.state import TraceStateManager
from reasonblocks.client import ReasonBlocks

api = ReasonBlocksAPI(api_key="rb_live_...")

config = ReasonBlocksConfig(
    # Tier toggles
    enable_e1=True,
    enable_e2=True,
    enable_e3=True,
    enable_monitor_steering=True,
    enable_general_monitor=False,
    enable_token_saving=True,

    # Token-saving
    ts_compress_threshold_chars=1800,
    ts_keep_recent_tool_messages=2,
    ts_enable_early_exit=True,

    # Routing / scoping
    monitor_task_profile="coding",
)

middleware = build_middleware(
    config, api,
    score_fn=ReasonBlocks.score_step,
    fsm=DifficultyFSM(),
    state_manager=TraceStateManager(),
    run_id="my-run-1",
    run_metadata={"agent_name": "bugfixer"},
)

agent = create_agent(model=..., tools=..., middleware=middleware)
```

--------------------------------

### Install ReasonBlocks dependencies

Source: https://docs.reasonblocks.com/guides/langchain

Install the necessary packages including reasonblocks, langchain, and relevant integrations.

```bash
pip install reasonblocks "langchain>=1.0" langchain-anthropic langgraph
```

--------------------------------

### Configure Token Saving Middleware

Source: https://docs.reasonblocks.com/api-reference/token-saving-middleware

Examples demonstrating standard compression, early-exit logic, and perplexity-based compression strategies.

```python
from reasonblocks import ReasonBlocks, TokenSavingMiddleware

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

mw = TokenSavingMiddleware(
    compress_threshold_chars=1800,
    keep_recent_tool_messages=2,
)

agent = create_agent(
    model=...,
    tools=...,
    middleware=[rb.middleware(agent_name="reviewer", task="Review PR #5"), mw],
)

result = agent.invoke({"messages": [HumanMessage(content="Review PR #5")]})

print(f"Compressions: {mw.stats.compressions}")
print(f"Chars saved:  {mw.stats.chars_saved}")
print(f"Early exits:  {mw.stats.early_exits}")
```

```python
from reasonblocks import ReasonBlocks, TokenSavingMiddleware

def my_signals(steps: list[dict]) -> dict[str, float]:
    # Return loop-likelihood signals in [0, 1]; the middleware reads
    # the "streak", "hedge", and "diversity" keys.
    return {"streak": 0.0, "hedge": 0.0, "diversity": 0.0}

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

mw = TokenSavingMiddleware(
    compress_threshold_chars=1800,
    keep_recent_tool_messages=2,
    early_exit_min_call_index=40,
    signals_fn=my_signals,
)

agent = create_agent(
    model=...,
    tools=...,
    middleware=[rb.middleware(agent_name="reviewer", task="Review PR #5"), mw],
)
```

```python
import anthropic
from reasonblocks import TokenSavingMiddleware, make_anthropic_classifier

classifier = make_anthropic_classifier(
    anthropic.Anthropic(),
    model="claude-haiku-4-5-20251001",
    target_keep_ratio=0.5,
)

mw = TokenSavingMiddleware(
    enable_perplexity_compression=True,
    perplexity_classifier=classifier,
    perplexity_recent_cutoff=3,
    perplexity_mid_cutoff=10,
    perplexity_keep_ratio_mid=0.55,
    perplexity_keep_ratio_old=0.30,
)

agent = create_agent(model=..., tools=..., middleware=[mw])
```

--------------------------------

### Implement a custom agent harness in Python

Source: https://docs.reasonblocks.com/api-reference/rest-api/custom-harness-quickstart

This example demonstrates the end-to-end workflow for a custom harness, including environment configuration, task execution, step monitoring, and final trace submission.

```python
import os, uuid, httpx

BASE = os.environ["RB_BASE_URL"]   # https://rb-api.reasonblocks.com
KEY  = os.environ["RB_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}"}
RUN_ID = f"harness-{uuid.uuid4()}"

def run_task(task: str, model: str):
    client = httpx.Client(base_url=BASE, headers=HEADERS, timeout=15.0)

    # 1) Pre-task: pull patterns to seed the system prompt.
    patterns = client.post("/v1/traces/retrieve", json={
        "context": task, "tier": "e1", "top_k": 3, "model": model,
    }).json()["traces"]
    system_prompt = build_prompt(task, patterns)

    # 2) Start the run for telemetry.
    client.post("/v1/monitor/runs", json={
        "run_id": RUN_ID, "agent_name": "my-harness", "task": task,
        "model": model, "framework": "custom", "task_profile": "coding",
    })

    steps = []
    for i in range(MAX_STEPS):
        step = your_agent_step(system_prompt, history=steps)
        steps.append(step)

        score = client.post(f"/v1/monitor/runs/{RUN_ID}/steps", json={
            "step_index": i,
            "action": step.action, "action_input": step.action_input,
            "thought": step.thought, "observation": step.observation,
            "is_error": step.is_error, "tokens": step.tokens,
        }).json()

        if score["fired"]:
            # Ask the server for an intervention to inject next turn.
            intervention = client.post("/v1/monitors/evaluate", json={
                "model": model, "task_profile": "coding",
                "steps": [s.to_dict() for s in steps],
            }).json()
            if intervention["inject"]:
                system_prompt += "\n\n" + intervention["intervention_text"]

        if step.is_terminal:
            break

    # 3) Post-task: submit the trace for distillation.
    client.post("/v1/traces", json={
        "trace_id": RUN_ID,
        "outcome": "success" if steps[-1].ok else "failure",
        "steps": [s.to_dict() for s in steps],
    })
    client.post(f"/v1/monitor/runs/{RUN_ID}/finish", json={
        "outcome": "success" if steps[-1].ok else "failure",
    })
```

--------------------------------

### Verify installation via CLI

Source: https://docs.reasonblocks.com/installation

Use this command to confirm the package is importable in your current environment.

```bash
python -c "import reasonblocks; print('ReasonBlocks installed')"
```

--------------------------------

### Install ReasonBlocks and LangGraph dependencies

Source: https://docs.reasonblocks.com/guides/langgraph

Install the necessary packages to enable ReasonBlocks integration with LangGraph.

```bash
pip install reasonblocks "langgraph>=0.2" langchain-anthropic
```

--------------------------------

### Run Messages Agent Loop Usage

Source: https://docs.reasonblocks.com/api-reference/integrations/claude-tools

Example demonstrating how to initialize tools and execute the agent loop with a specific model and system prompt.

```python
from reasonblocks.integrations.claude_tools import (
    make_claude_tools,
    run_messages_agent_loop,
)

tool_specs, dispatch = make_claude_tools(memory, graph)

outcome = run_messages_agent_loop(
    client,
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Audit the payment module"}],
    tool_specs=tool_specs,
    dispatch=dispatch,
    system="You are a senior Python code reviewer.",
    max_steps=30,
)

print(outcome["final_text"])
for name, inp, result in outcome["tool_calls"]:
    print(f"  {name}({inp}) -> {result[:80]}")
```

--------------------------------

### Install LangChain dependency

Source: https://docs.reasonblocks.com/troubleshooting/common-issues

Ensure the required version of LangChain is installed.

```bash
pip install "langchain>=1.0"
```

--------------------------------

### Configure tool permissions and sensitivity

Source: https://docs.reasonblocks.com/api-reference/integrations/openai-tools

Examples for disabling the store functionality or adjusting recall parameters.

```python
rb_tools = make_openai_tools(
    memory,
    graph,
    enable_store=False,
)
```

```python
rb_tools = make_openai_tools(
    memory,
    graph,
    recall_top_k=10,
    recall_threshold=0.35,
)
```

--------------------------------

### Install networkx dependency

Source: https://docs.reasonblocks.com/troubleshooting/common-issues

Install the optional networkx dependency required for ImportGraph.

```bash
pip install networkx
```

--------------------------------

### Implement Manual Anthropic Tool-Use Loop

Source: https://docs.reasonblocks.com/api-reference/integrations/claude-tools

Example demonstrating how to use make_claude_tools with the Anthropic client to manage a tool-use loop.

```python
import anthropic
from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.claude_tools import make_claude_tools

import pathlib

client = anthropic.Anthropic()
memory = CodebaseMemory(codebase_id="my-repo", api_key="rb_live_...")
graph = ImportGraph().build_from_files(
    {str(p): p.read_text() for p in pathlib.Path("myrepo").rglob("*.py")}
)

tool_specs, dispatch = make_claude_tools(memory, graph)

messages = [{"role": "user", "content": "Find the bug in auth/session.py"}]

while True:
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        tools=tool_specs,
        messages=messages,
    )
    messages.append({"role": "assistant", "content": resp.content})

    if resp.stop_reason != "tool_use":
        break

    tool_results = []
    for block in resp.content:
        if block.type == "tool_use":
            result = dispatch(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })
    messages.append({"role": "user", "content": tool_results})
```

--------------------------------

### Initialize and Configure PR Review ETrace

Source: https://docs.reasonblocks.com/guides/pr-review-etrace

Setup the store and agent middleware to enable trajectory caching for PR reviews.

```python
from reasonblocks import for_code_review, PRReviewETraceStore
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
# or: from langchain_anthropic import ChatAnthropic

store = PRReviewETraceStore(save_path="./.rb/pr_etrace.json")

agent = create_agent(
    model=ChatOpenAI(model="gpt-5.2"),       # or any LangChain-compatible model
    tools=[bash, search_codebase, read_file],
    middleware=for_code_review(etrace_store=store),
)

# After each successful review, persist the trajectory so future similar
# PRs can hit the cache.
store.store(
    pr_signature=current_pr_diff,
    review=review_json,
    files_inspected=files_read_during_run,
)
```

--------------------------------

### Build ImportGraph from files

Source: https://docs.reasonblocks.com/api-reference/import-graph

Populate the graph by providing a dictionary of file paths and their source code content. Requires the networkx library to be installed.

```python
import pathlib

# Load all .py files from a local repo
files = {
    str(p): p.read_text()
    for p in pathlib.Path("myrepo").rglob("*.py")
}

graph = ImportGraph()
graph.build_from_files(files)

print(graph.stats())
# {'nodes': 214, 'edges': 891}
```

```python
graph = ImportGraph().build_from_files(files)
```

--------------------------------

### Initialize ReasonBlocks

Source: https://docs.reasonblocks.com/guides/openai-agents

Set up the ReasonBlocks client using an API key from environment variables.

```python
import os
from reasonblocks import ReasonBlocks

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

--------------------------------

### Initialize ReasonBlocks instance

Source: https://docs.reasonblocks.com/api-reference/reasonblocks

Create a single instance per process with the required API key and optional configuration settings.

```python
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key="rb_live_...",
    token_budget=50_000,
    task_profile="pr_review",
)
```

--------------------------------

### begin_step

Source: https://docs.reasonblocks.com/api-reference/steering-session

Runs scoring, FSM transition, monitor evaluation, E-trace retrieval, and pattern rendering for one turn.

```APIDOC
## begin_step

### Description
Runs scoring, FSM transition, monitor evaluation, E-trace retrieval, and pattern rendering for one turn. Returns a StepDecision carrying the resolved FSM state, pending injections, optional model override, the rendered injection text, and a fresh StepLogEntry.

### Method
session.begin_step(*, thought, action=None, action_input=None, observation=None) -> StepDecision

### Parameters
- **thought** (str) - Required - The thought process for the step.
- **action** (str) - Optional - The action to perform.
- **action_input** (str) - Optional - Input for the action.
- **observation** (str) - Optional - Observation data.
```

--------------------------------

### Initialize ReasonBlocks client

Source: https://docs.reasonblocks.com/guides/langchain

Create a ReasonBlocks instance using your API key. The constructor requires an explicit API key, while the base URL can be configured via environment variables.

```python
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
)
```

--------------------------------

### GET /health

Source: https://docs.reasonblocks.com/api-reference/rest-api/setup

Performs a sanity check on the service to ensure it is operational.

```APIDOC
## GET /health

### Description
Sanity-check the service to verify it is running.

### Method
GET

### Endpoint
https://rb-api.reasonblocks.com/health

### Response
#### Success Response (200)
- **status** (string) - The status of the service.
- **version** (string) - The current version of the API.
```

--------------------------------

### Initialize ReasonBlocks with a task profile

Source: https://docs.reasonblocks.com/configuration/monitor-profiles

Select a predefined monitor profile by setting the task_profile parameter during client initialization.

```python
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key="rb_live_...",
    task_profile="coding",   # or "pr_review", "qa"
)
```

--------------------------------

### Using SteeringSession as a context manager

Source: https://docs.reasonblocks.com/api-reference/steering-session

Demonstrates the sync/async context manager usage for beginning and ending LLM steps.

```python
with SteeringSession(...) as session:
    decision = session.begin_step(thought=...)
    # ... call your LLM, then ...
    session.end_step(decision, model_id=..., tokens=..., tool_calls=...)
```

--------------------------------

### Initialize ReasonBlocks and Anthropic Client

Source: https://docs.reasonblocks.com/guides/claude-messages

Configure the ReasonBlocks client with model routing and initialize the Anthropic SDK client.

```python
import os
from anthropic import Anthropic
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key=os.environ["REASONBLOCKS_API_KEY"],
    model_routing={
        "FAST": "anthropic:claude-haiku-4-5-20251001",
        "SLOW": "anthropic:claude-sonnet-4-6",
    },
)
client = Anthropic()
```

--------------------------------

### API Request Endpoint Example

Source: https://docs.reasonblocks.com/api-reference/rest-api/setup

The standard format for API requests, using the versioned path for all new integrations.

```http
POST https://rb-api.reasonblocks.com/v1/traces/retrieve
```

--------------------------------

### Configure ReasonBlocks for self-hosted deployment

Source: https://docs.reasonblocks.com/installation

Initialize the ReasonBlocks client by explicitly providing an API key and a custom base URL.

```python
import os
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key=os.environ["REASONBLOCKS_API_KEY"],
    base_url=os.environ.get("REASONBLOCKS_BASE_URL", "http://localhost:9000"),
    e_traces_enabled=True,
)
```

--------------------------------

### Serializing StepLogEntry Data

Source: https://docs.reasonblocks.com/api-reference/types/step-record

Example of iterating through the middleware step log and serializing entries to JSON using as_dict().

```python
import json

with rb.middleware(...) as mw:
    result = agent.invoke({"messages": [...]})

print(json.dumps([e.as_dict() for e in mw.step_log], indent=2))
```

--------------------------------

### Retrieve Reasoning Patterns

Source: https://docs.reasonblocks.com/api-reference/rest-api/custom-harness-quickstart

Call this endpoint at the start of a task to fetch relevant reasoning patterns for prompt injection.

```bash
curl $RB_BASE_URL/v1/traces/retrieve \
  -H "Authorization: Bearer $RB_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "context": "refactor auth middleware to use JWT",
    "tier": "e1",
    "top_k": 3,
    "model": "claude-opus-4-7"
  }'
```

```json
{
  "traces": [
    {
      "tier": "e1",
      "pattern_id": "p_abc123",
      "similarity": 0.87,
      "fields": { "situation": "...", "dead_ends": "...", "unlock": "..." }
    }
  ]
}
```

--------------------------------

### Deploy with Docker

Source: https://docs.reasonblocks.com/api-reference/rest-api/setup

Build and run the ReasonBlocks API service using Docker with environment variables loaded from a .env file.

```bash
docker build -t rb-api .
docker run -p 8000:8000 --env-file .env rb-api
```

--------------------------------

### Initialize ReasonBlocks Client

Source: https://docs.reasonblocks.com/installation

Instantiate the ReasonBlocks client with custom API credentials, FSM thresholds, and model routing configurations.

```python
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key="rb_live_...",
    base_url="https://your-rb-api-host.example.com",
    token_budget=100_000,
    fsm_thresholds={
        "fast_threshold": 0.2,
        "slow_threshold": 0.6,
        "skip_threshold": 0.85,
    },
    model_routing={
        "FAST":   "anthropic:claude-haiku-4-5-20251001",
        "NORMAL": "anthropic:claude-sonnet-4-6",
        "SLOW":   "anthropic:claude-sonnet-4-6",
        "SKIP":   "anthropic:claude-sonnet-4-6",
    },
    e_traces_enabled=True,
    live_streaming_enabled=True,
    task_profile="coding",
)
```

--------------------------------

### Constructor

Source: https://docs.reasonblocks.com/api-reference/token-saving-middleware

Configuration parameters for initializing the Token Saving Middleware.

```APIDOC
## Constructor

### Parameters
- **compress_threshold_chars** (integer, default: 1800) - Minimum character length a ToolMessage body must reach before it is compressed.
- **head_keep_chars** (integer, default: 900) - Number of characters to keep from the start of a tool output when compressing.
- **tail_keep_chars** (integer, default: 700) - Number of characters to keep from the end of a tool output when compressing.
- **keep_recent_tool_messages** (integer, default: 2) - Number of the most recent ToolMessage objects to exempt from compression.
- **early_exit_min_call_index** (integer, default: 40) - Minimum number of model calls that must have occurred before an early-exit nudge can be injected.
- **early_exit_text** (string, default: "You appear to be stuck in a loop...") - The text injected as a HumanMessage when an early-exit nudge fires.
- **signals_fn** (callable, default: None) - A function (steps: list[dict]) -> dict[str, float] that evaluates the agent's trajectory and returns loop-likelihood signals.
- **enable_compression** (boolean, default: True) - Whether to enable head+tail tool-output compression.
- **enable_early_exit** (boolean, default: True) - Whether to enable the early-exit nudge.
- **enable_perplexity_compression** (boolean, default: False) - Whether to enable word-level perplexity-based compression.
- **perplexity_classifier** (callable) - A WordClassifier callable (words: list[str]) -> list[bool] that returns a keep/drop decision for each word.
- **perplexity_recent_cutoff** (integer, default: 3) - Messages from fewer than this many model calls ago are considered "recent" and are excluded from perplexity compression.
- **perplexity_mid_cutoff** (integer, default: 10) - Messages from between perplexity_recent_cutoff and this many calls ago are in the "mid" tier.
- **perplexity_keep_ratio_mid** (number, default: 0.55) - Target fraction of words to keep in "mid" tier messages.
- **perplexity_keep_ratio_old** (number, default: 0.30) - Target fraction of words to keep in "old" tier messages.
- **perplexity_window_words** (integer, default: 50) - The number of words per window passed to the classifier in a single call.
- **perplexity_min_content_words** (integer, default: 30) - Texts shorter than this many words are returned unchanged.
```

--------------------------------

### Initialize and run an OpenAI Agent with ReasonBlocks

Source: https://docs.reasonblocks.com/guides/openai-agents

Configures an agent with codebase memory tools and executes a task using the ReasonBlocks runner.

```python
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())
```

--------------------------------

### Example AB Testing JSON Response

Source: https://docs.reasonblocks.com/guides/ab-testing

A sample JSON response demonstrating the structure of arms, comparison metrics, SRM checks, and timeseries data.

```json
{
  "arms": {
    "on":  {"n_runs": 6, "success_rate": 0.83,
            "tokens_per_run": {"median": 1996},
            "cost": {"cache_read_tokens": 14400, "cost_per_run_usd": 0.0101}},
    "off": {"n_runs": 6, "success_rate": 0.83,
            "tokens_per_run": {"median": 12600},
            "cost": {"cache_read_tokens": 0, "cost_per_run_usd": 0.063}}
  },
  "comparison": {
    "cost_per_run_usd":      {"on": 0.0101, "off": 0.063, "pct": -0.84},
    "tokens_per_run_median": {"on": 1996,  "off": 12600, "pct": -0.70},
    "success_rate": {"on": 0.83, "off": 0.83, "delta": 0.0,
                     "ci_95": [-0.42, 0.42], "method": "newcombe"}
  },
  "srm": {"observed": {"on": 6, "off": 6}, "p_value": 1.0, "flagged": false},
  "timeseries": [{"day": "2026-05-19", "arm": "on", "tokens_per_run": 1996, "success_rate": 1.0}],
  "caveats": ["Accuracy CI is wide (±42%); N is too small to bound a tight margin ..."]
}
```

--------------------------------

### Initialize CodebaseMemory

Source: https://docs.reasonblocks.com/api-reference/codebase-memory

Instantiate the client by providing the codebase identifier and authentication credentials.

```python
from reasonblocks import CodebaseMemory

memory = CodebaseMemory(
    codebase_id="myorg_myrepo",
    api_key="rb_live_...",
    base_url="https://rb-api.mycompany.com",
)
```

--------------------------------

### Initialize and query Claude tools

Source: https://docs.reasonblocks.com/api-reference/integrations/claude-tools

Demonstrates creating tools from memory and graph objects and passing them to the Claude SDK query function.

```python
from claude_agent_sdk import query
from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.claude_tools import make_claude_agent_sdk_tools

import pathlib

memory = CodebaseMemory(codebase_id="my-repo", api_key="rb_live_...")
graph = ImportGraph().build_from_files(
    {str(p): p.read_text() for p in pathlib.Path("myrepo").rglob("*.py")}
)

rb_tools = make_claude_agent_sdk_tools(memory, graph)

async for message in query(
    prompt="Find and explain the auth bug",
    tools=[*rb_tools, *your_tools],
):
    print(message)
```

--------------------------------

### Configure E-trace with Prompt Caching

Source: https://docs.reasonblocks.com/guides/pr-review-etrace

Initializes an agent with both prompt caching and E-trace middleware enabled for optimized PR review performance.

```python
agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5"),
    tools=[...],
    middleware=for_code_review(
        prompt_caching=True,
        etrace_store=store,
    ),
)
```

--------------------------------

### Verify constructor wiring

Source: https://docs.reasonblocks.com/installation

Initialize the ReasonBlocks client and middleware to ensure the API connection and internal wiring are healthy.

```python
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(api_key="rb_live_...")
mw = rb.middleware(agent_name="test")
print("Middleware created:", mw)
```

--------------------------------

### ReasonBlocks Constructor

Source: https://docs.reasonblocks.com/api-reference/reasonblocks

Initializes the ReasonBlocks SDK instance. Create one instance per process and provide the necessary configuration for your agent run.

```APIDOC
## ReasonBlocks(api_key, base_url, token_budget, fsm_thresholds, model_routing, e_traces_enabled, monitor_names, live_streaming_enabled, task_profile)

### Description
Initializes the main ReasonBlocks SDK client. This instance should be created once per process.

### Parameters
- **api_key** (string) - Required - Your ReasonBlocks API key (starts with rb_live_ or rb_test_).
- **base_url** (string) - Optional - Override the API base URL for self-hosted or staging environments.
- **token_budget** (integer) - Optional - Soft cap on tokens tracked by the trace state manager.
- **fsm_thresholds** (object) - Optional - Override difficulty FSM transition thresholds (e.g., fast_threshold, slow_threshold).
- **model_routing** (object) - Optional - Map FSM state names to model identifiers.
- **e_traces_enabled** (boolean) - Optional - Whether to retrieve and inject E-trace patterns (default: True).
- **monitor_names** (list[str]) - Optional - Backward compatibility field for monitor suites.
- **live_streaming_enabled** (boolean) - Optional - Whether to stream run events to the API (default: True).
- **task_profile** (string) - Optional - Selects the monitor weight vector (default: "coding").
```

--------------------------------

### Initialize and integrate CodebaseMemory tools

Source: https://docs.reasonblocks.com/guides/langchain

Wraps memory and graph objects into LangChain tools and adds them to an agent configuration. Requires the networkx library for ImportGraph functionality.

```python
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],
)
```

--------------------------------

### Initialize ReasonBlocks with advanced parameters

Source: https://docs.reasonblocks.com/configuration/advanced

Configures the ReasonBlocks client with specific tuning parameters. Note that all parameters except api_key are keyword-only.

```python
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key="rb_live_...",
    base_url=None,                  # default: hosted API
    token_budget=None,              # tracked-only; not enforced
    monitor_names=None,             # accepted but unused on the middleware path
    fsm_thresholds=None,            # default: see DifficultyFSM
    model_routing=None,             # default: no routing
    e_traces_enabled=True,
    live_streaming_enabled=True,
    task_profile="coding",
)
```

--------------------------------

### Begin a steering session step

Source: https://docs.reasonblocks.com/api-reference/steering-session

Executes scoring, FSM transition, and pattern rendering for a single turn. Pass thought=None for the initial call to trigger the first-call path.

```python
session.begin_step(
    *,
    thought,
    action=None,
    action_input=None,
    observation=None,
) -> StepDecision
```

--------------------------------

### Configure ReasonBlocks with monitor weights

Source: https://docs.reasonblocks.com/configuration/reasonblocks-config

Initializes the ReasonBlocks configuration with a specific task profile and custom monitor weight overrides.

```python
config = ReasonBlocksConfig(
    monitor_task_profile="coding",
    monitor_weights={"semantic_loop": 0.5, "verification_skip": 0.05},
)
```

--------------------------------

### Configure Perplexity Compression

Source: https://docs.reasonblocks.com/guides/token-saving

Set up the WordClassifier and enable perplexity compression in the middleware.

```python
import anthropic
from reasonblocks.token_saving import (
    TokenSavingMiddleware,
    make_anthropic_classifier,
)

client = anthropic.Anthropic()

classifier = make_anthropic_classifier(
    client,
    model="claude-haiku-4-5-20251001",
    target_keep_ratio=0.5,
)

ts = TokenSavingMiddleware(
    enable_perplexity_compression=True,
    perplexity_classifier=classifier,
)
```

--------------------------------

### Initialize LangChain Agent with ReasonBlocks

Source: https://docs.reasonblocks.com/guides/langchain

Sets up CodebaseMemory and middleware to provide an agent with codebase context. Note that each middleware instance is single-use and must be re-initialized for subsequent runs.

```python
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.")]})
```

--------------------------------

### Inspect and Analyze Step Logs

Source: https://docs.reasonblocks.com/api-reference/types/step-record

Demonstrates iterating through step logs, filtering for specific FSM states, and calculating average difficulty metrics.

```python
from reasonblocks import ReasonBlocks

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

with rb.middleware(run_id="run-1", agent_name="reviewer", task="review PR #42") as mw:
    result = agent.invoke({"messages": [("user", "Review the changes in PR #42")]})

for entry in mw.step_log:
    print(
        f"  [{entry.step}] difficulty={entry.difficulty}  "
        f"state={entry.fsm_state}  tools={entry.tool_calls}"
    )
```

```python
slow_steps = [
    e for e in mw.step_log
    if e.fsm_state in ("SLOW", "SKIP")
]
print(f"{len(slow_steps)} slow/skip out of {len(mw.step_log)} total")
```

```python
scored = [e for e in mw.step_log if e.difficulty is not None]
if scored:
    avg = sum(e.difficulty for e in scored) / len(scored)
    print(f"Average difficulty: {avg:.3f}")
```

--------------------------------

### POST /v1/monitor/runs

Source: https://docs.reasonblocks.com/api-reference/rest-api/custom-harness-quickstart

Initializes a new monitoring run for an agent.

```APIDOC
## POST /v1/monitor/runs

### Description
Initializes a run for server-side monitor scoring and dashboard visibility.

### Method
POST

### Endpoint
/v1/monitor/runs

### Request Body
- **run_id** (string) - Required - Unique identifier for the run.
- **agent_name** (string) - Required - Name of the agent.
- **task** (string) - Required - Description of the task.
- **model** (string) - Required - Model name.
- **framework** (string) - Required - Framework name.
- **task_profile** (string) - Required - Profile type.
```

--------------------------------

### Configure Agent with Prompt Caching

Source: https://docs.reasonblocks.com/guides/prompt-caching

Initialize a LangChain agent with ReasonBlocks middleware, enabling prompt caching and chain-of-draft directives to reduce token usage.

```python
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_agent
from reasonblocks import for_code_review, chain_of_draft_directive

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-6", max_tokens=2048),
    tools=[bash, ...],
    system_prompt=task_prompt + chain_of_draft_directive(),  # CoD on output reasoning
    middleware=for_code_review(                              # D-arm + caching, one middleware list
        fail_to_pass_tests=task.fail_to_pass,
        max_tool_calls=50,
        prompt_caching=True,                                 # ← turns on Anthropic prompt caching
    ),
)
```

--------------------------------

### Log Telemetry and Run Lifecycle

Source: https://docs.reasonblocks.com/api-reference/rest-api/custom-harness-quickstart

Use these endpoints to initialize a run, log individual agent steps, and mark the completion of a task.

```bash
curl $RB_BASE_URL/v1/monitor/runs \
  -H "Authorization: Bearer $RB_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "run_id": "harness-run-2026-05-12-001",
    "agent_name": "my-custom-harness",
    "task": "refactor auth middleware to use JWT",
    "model": "claude-opus-4-7",
    "framework": "custom",
    "task_profile": "coding"
  }'
```

```bash
curl $RB_BASE_URL/v1/monitor/runs/harness-run-2026-05-12-001/steps \
  -H "Authorization: Bearer $RB_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "step_index": 0,
    "action": "read_file",
    "action_input": "src/auth.py",
    "thought": "Need to find the current session-token logic.",
    "observation": "Found `verify_session()` at line 42.",
    "is_error": false,
    "tokens": 1834
  }'
```

```json
{
  "scores": { "verification_skip": 0.0, "claim_contradiction": 0.0 },
  "total_score": 0.12,
  "fired": []
}
```

```bash
curl $RB_BASE_URL/v1/monitor/runs/harness-run-2026-05-12-001/finish \
  -H "Authorization: Bearer $RB_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"outcome": "success"}'
```

--------------------------------

### Implement telemetry with ReasonBlocks

Source: https://docs.reasonblocks.com/guides/claude-agent-sdk

Wraps the query function with a telemetry adapter to track agent execution steps and performance metrics.

```python
import asyncio
from claude_agent_sdk import query
from reasonblocks import ReasonBlocks

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

async def main():
    async with rb.claude_agent_telemetry(
        agent_name="reviewer",
        task="investigate MarkerWidget",
        model="claude-haiku-4-5-20251001",
    ) as tele:
        async for message in tele.wrap(
            query(prompt="...", options={"tools": tools, "model": "claude-haiku-4-5-20251001"}),
        ):
            # process message yourself; telemetry was already emitted
            ...

asyncio.run(main())
```

--------------------------------

### rb.middleware()

Source: https://docs.reasonblocks.com/

LangChain 1.0 AgentMiddleware reference implementation for token-saving and general-monitor middleware.

```APIDOC
## rb.middleware()

### Description
Provides the reference implementation for LangChain 1.0 AgentMiddleware, allowing token-saving and general-monitor middleware to compose alongside it.
```

--------------------------------

### Initialize ReasonBlocks Middleware with LangChain

Source: https://docs.reasonblocks.com/api-reference/reasonblocks

Use the middleware as a context manager to wrap an agent execution. Ensure it is called once per agent run rather than per step.

```python
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")]})
```

--------------------------------

### Initialize Codebase Memory Tools for LangGraph

Source: https://docs.reasonblocks.com/guides/langgraph

Create LangChain-compatible tools from a CodebaseMemory instance for use in LangGraph nodes.

```python
from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.langchain_tools import make_langchain_tools

memory = CodebaseMemory(codebase_id="my-org/my-repo", api_key=rb._api_key)
graph_tools = make_langchain_tools(memory)
```

--------------------------------

### Assemble middleware via unified config

Source: https://docs.reasonblocks.com/guides/code-review-mode

Uses ReasonBlocksConfig and build_middleware to automatically handle the ordering of the general monitor and token-saving components.

```python
from reasonblocks import ReasonBlocksAPI, ReasonBlocksConfig, build_middleware
from reasonblocks.fsm import DifficultyFSM
from reasonblocks.state import TraceStateManager
from reasonblocks.client import ReasonBlocks

api = ReasonBlocksAPI(api_key="rb_live_...")

cfg = ReasonBlocksConfig(
    enable_token_saving=True,
    ts_compress_threshold_chars=1800,
    ts_keep_recent_tool_messages=2,
    ts_enable_early_exit=True,
    enable_general_monitor=True,   # the accuracy-lift arm; drop for token-saving only
    gm_max_tool_calls=50,
)

middleware = build_middleware(
    cfg, api,
    score_fn=ReasonBlocks.score_step,
    fsm=DifficultyFSM(),
    state_manager=TraceStateManager(),
)
```

--------------------------------

### Inspect FSM step logs

Source: https://docs.reasonblocks.com/troubleshooting/common-issues

Print step details to diagnose why E-tracing might be skipped.

```python
for entry in mw.step_log:
    print(entry.step, entry.fsm_state, entry.difficulty, entry.skipped_reason)
```

--------------------------------

### Import make_openai_tools

Source: https://docs.reasonblocks.com/api-reference/integrations/openai-tools

Import the factory function from the reasonblocks integrations module.

```python
from reasonblocks.integrations import make_openai_tools
```

--------------------------------

### ReasonBlocks constructor

Source: https://docs.reasonblocks.com/configuration/advanced

Initializes the ReasonBlocks client with advanced configuration options for workload tuning and instrumentation.

```APIDOC
## ReasonBlocks Constructor

### Description
Initializes a new instance of the ReasonBlocks client. All parameters except `api_key` are keyword-only.

### Signature
`ReasonBlocks(api_key, base_url=None, token_budget=None, monitor_names=None, fsm_thresholds=None, model_routing=None, e_traces_enabled=True, live_streaming_enabled=True, task_profile="coding")`

### Parameters
- **api_key** (string) - Required - The API key for authentication.
- **base_url** (string) - Optional - The base URL for the API. Defaults to the hosted API.
- **token_budget** (int) - Optional - Token budget for tracking purposes.
- **monitor_names** (list) - Optional - List of monitor names.
- **fsm_thresholds** (dict) - Optional - FSM thresholds configuration.
- **model_routing** (dict) - Optional - Model routing configuration.
- **e_traces_enabled** (bool) - Optional - Whether to enable e-traces. Defaults to True.
- **live_streaming_enabled** (bool) - Optional - Whether to enable live streaming. Defaults to True.
- **task_profile** (string) - Optional - The task profile to use. Defaults to "coding".
```

--------------------------------

### POST /v1/traces/retrieve

Source: https://docs.reasonblocks.com/api-reference/rest-api/custom-harness-quickstart

Retrieves reasoning patterns based on a task description to inject into the system prompt.

```APIDOC
## POST /v1/traces/retrieve

### Description
Call this once at the start of a task to get reasoning patterns to inject into your system prompt.

### Method
POST

### Endpoint
/v1/traces/retrieve

### Parameters
#### Request Body
- **context** (string) - Required - One-sentence description of the task you're about to run.
- **tier** (string) - Optional - e1 (project-specific), e2 (commons), or e3 (universal).
- **top_k** (integer) - Optional - 1–50. Default is 5.
- **model** (string) - Optional - Target model name.
- **failure_type** (string) - Optional - Narrow to one failure category.

### Request Example
{
  "context": "refactor auth middleware to use JWT",
  "tier": "e1",
  "top_k": 3,
  "model": "claude-opus-4-7"
}

### Response
#### Success Response (200)
- **traces** (array) - List of retrieved patterns.

#### Response Example
{
  "traces": [
    {
      "tier": "e1",
      "pattern_id": "p_abc123",
      "similarity": 0.87,
      "fields": { "situation": "...", "dead_ends": "...", "unlock": "..." }
    }
  ]
}
```
Tokens:
Raw