Skip to content
Tenbri pangolinTenbri

LangGraph and LangChain

Two lines record every tool call of every LangGraph graph and LangChain agent (create_agent, createAgent) in a Tenbri stream: tenbri.init() opens the stream, tenbri.instrument_langchain() (instrumentLangChain() in TypeScript) registers Tenbri’s callback handler for the whole process. Each tool call becomes an intent that is durable before the tool runs, then its outcome. When a person answers a HumanInTheLoopMiddleware interrupt and you resume with Command(resume=...), the answer is recorded as an approval or a refusal. No callbacks to pass.

Quick start

Section titled “Quick start”

1. Install

Section titled “1. Install”

Python 3.10 or later.

Terminal window
pip install --pre tenbri-langgraph langchain

This brings the tenbri binding (0.1.0-beta.2), langchain-core 1.5 or later and langgraph 1.2 or later. Add your model’s integration too, for example langchain-anthropic.

2. Instrument your agent

Section titled “2. Instrument your agent”

Say what a refund is in tenbri.toml, next to your code. The target names the tool’s arguments in braces and is filled from each call; the policy is the rule recorded as having allowed it:

tenbri.toml
[actions.issue_refund]
target = "stripe:{charge}"     # {charge} is filled from the tool call's arguments
policy = "policy:refund@14"    # the rule that allowed it: ref@version

Then add the two lines to your agent. Nothing else in it changes:

agent.py
import tenbri
from langchain.agents import create_agent
from langchain_core.tools import tool


tenbri.init()  # reads TENBRI_STREAM... and ./tenbri.toml
tenbri.instrument_langchain()  # every tool call from here on is recorded




@tool
def issue_refund(charge: str, amount: int) -> str:
    """Refund `amount` EUR of a charge."""
    return f"refunded {amount} EUR of {charge}"




agent = create_agent("anthropic:claude-sonnet-5-5", tools=[issue_refund], name="support")
result = agent.invoke({"messages": [("user", "Refund 20 EUR of ch_1")]})
print(result["messages"][-1].content)
print("pin this:", tenbri.default().fingerprint)

Every tool call of every agent and graph the process runs is recorded from then on, in every thread, sync or async.

The agent’s name is who acts: agent:support, no TENBRI_AGENT needed.

3. Run it

Section titled “3. Run it”

Point Tenbri at a stream, with your model’s API key set (ANTHROPIC_API_KEY here):

Terminal window
export TENBRI_STREAM=./audit
python agent.py
Refunded 20 EUR of ch_1.
pin this: sha256:5ab4e13077d0f33d4d03ec05ce702827958a6c1847247e02df7aca54eabe7bff

init() creates the stream and its signing keys on first use. The keys go to $TENBRI_KEYS, else ~/.config/tenbri/keys, never into the stream (keys). Keep the fingerprint: it is what you verify against. This output comes from a run with a scripted chat model in place of the real one (see testing offline) that asks for issue_refund once, then answers.

4. See the entries

Section titled “4. See the entries”
Terminal window
tenbri tail ./audit -n 2 --no-follow
{"action":{"target":"stripe:ch_1","type":"issue_refund"},"durability":"local","ext":{"langchain":{"run_id":"01a15a14-268e-7184-85da-706077931b7c","tool_call_id":"call_1"}},"id":"01M5D189MXPWD8YFTD905M485F","inputs":[{"digest":"hmac-sha256:ebf4398c…","ref":"langchain:tool_call/call_1","version":"args"}],"kind":"decision","phase":"intent","prev":"sha256:8b01c989…","principal":{"id":"support","kind":"agent"},"rule":{"ref":"policy:refund","result":"allow","version":"14"},"seq":3,"sig":"ed25519:1c258c09…","signer":"sess:2026-10-20:d665b2aec157","source":"sdk","stream":"tnb1:5ab4e13077d0f33d4d03ec05ce702827","ts":"2026-10-20T16:10:32.480Z","v":1}
{"action":{"target":"stripe:ch_1","type":"issue_refund"},"id":"01M5D189NHV3YJCTT8853JXCZK","intent":"01M5D189MXPWD8YFTD905M485F","kind":"decision","phase":"outcome","prev":"sha256:90689266…","principal":{"id":"support","kind":"agent"},"seq":4,"sig":"ed25519:4a261ee2…","signer":"sess:2026-10-20:d665b2aec157","source":"sdk","status":"ok","stream":"tnb1:5ab4e13077d0f33d4d03ec05ce702827","ts":"2026-10-20T16:10:32.500Z","v":1}
Terminal window
tenbri verify ./audit --key sha256:5ab4e130…   # your full fingerprint
verified 4 entries, stream tnb1:5ab4e13077d0f33d4d03ec05ce702827, pinned to sha256:5ab4e13077d0f33d4d03ec05ce702827958a6c1847247e02df7aca54eabe7bff

Entries 1 and 2 are the stream’s key entries. Entry 3 is the intent, written before issue_refund ran: the target comes from tenbri.toml, filled with the call’s charge, and the rule is its policy. Entry 4 is the outcome. The call’s arguments are pinned by keyed digest, never stored; the LangChain run id and tool call id go into ext.langchain, so a LangSmith run and a decision find each other.

Configuration

Section titled “Configuration”

Environment variables

Section titled “Environment variables”

tenbri.init() (init()) reads these; an argument (init(approver=...), init({ approver })) wins over its variable.

Variable Argument What it is
TENBRI_STREAM stream Where entries go: a directory, file://..., s3://bucket/prefix or r2://bucket/prefix. Required.
TENBRI_KEYS keys The keys directory, apart from the stream. Default ~/.config/tenbri/keys.
TENBRI_AGENT agent Who acts when the agent has no name: agent:support. Without either, agent:langgraph.
TENBRI_APPROVER approver Who answers the interrupts: person:alice.
TENBRI_SCHEMA schema The action schemas. Default ./tenbri.toml when there is one.
fail_open (failOpen) Let tools run, unrecorded and with a warning, when their intent cannot be written. Default: block them.

init() flushes and closes the stream when the process exits. In Python, with tenbri.init(...) as log: scopes it to a block; in TypeScript, await using log = await init(...).

Map tool calls to actions

Section titled “Map tool calls to actions”

A tool call is the action of the tool’s name, or of the [actions.*] table in tenbri.toml that its name ends with (an MCP tool mcp__billing__issue_refund is issue_refund). From that table, the record takes target, with each {argument} filled from the call’s arguments ({order.id} reads a nested field), and policy, "ref@version", as the intent’s rule. In tenbri.toml, a target without placeholders is a pattern for validation only and is not recorded.

tenbri.toml is the same file tenbri schema check and tenbrid validate entries against. A plain dict (object) in code says the same and wins over the file, per tool name. It can rename the action, and take functions of the arguments:

tenbri.instrument_langchain(
    actions={
        "issue_refund": {"target": "stripe:{charge}", "policy": "policy:refund@14"},
        "close": {"action": "close_ticket", "target": lambda args: f"ticket:{args['id']}"},
    }
)

A tool neither names is recorded under its own name, with no target.

Options

Section titled “Options”

Python: tenbri.instrument_langchain(**options). TypeScript: instrumentLangChain({ ... }).

Python TypeScript Default What it does
actions actions tenbri.toml Per tool name: action, target, subject (templates or functions of the arguments) and policy. Wins over the file.
agent agent see below Who acts, for every tool call.
approver approver TENBRI_APPROVER Who answers the interrupts: a principal, or a function of the tool and its arguments returning one.
tools tools every tool Names to record, or a predicate of the name.
fail_open failOpen init()’s, False Run the tool, unrecorded and with a warning, when its intent could not be written.

Who acts, from the narrowest: the agent option, then the innermost with tenbri.agent("agent:...") (TypeScript: agent('agent:...', fn)), then the agent’s name (create_agent(name=...), createAgent({ name })), then TENBRI_AGENT, else agent:langgraph. tenbri_langgraph.uninstrument() (uninstrumentLangChain()) stops recording.

For trace ids on entries, install OpenTelemetry (pip install --pre "tenbri-langgraph[otel]"; the npm package already depends on @opentelemetry/api). Each entry then carries the active span’s trace.

Approvals and refusals

Section titled “Approvals and refusals”
What happens Entries
A person approves (or edits) a HumanInTheLoopMiddleware interrupt, and you resume with Command(resume=...) An approval by the approver; the tool’s intent names it in approval: {by, entry}
A person rejects the interrupt An intent and a refused outcome, with the reviewer in ext.langchain
A person responds instead of the tool (respond) A note by the reviewer
Your tool raises tenbri.Refused (throw new Refused(...)) The intent, then the outcome refused

HumanInTheLoopMiddleware needs a checkpointer. Resume the graph the LangGraph way; the answers in the Command are recorded as it starts, by TENBRI_APPROVER:

refunds.py
import tenbri
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


tenbri.init()  # TENBRI_STREAM, TENBRI_APPROVER, ./tenbri.toml
tenbri.instrument_langchain()




@tool
def issue_refund(charge: str, amount: int) -> str:
    """Refund `amount` EUR of a charge."""
    return f"refunded {amount} EUR of {charge}"




agent = create_agent(
    "anthropic:claude-sonnet-5-5",
    tools=[issue_refund],
    name="support",
    middleware=[
        HumanInTheLoopMiddleware(interrupt_on={
            "issue_refund": {
                "allowed_decisions": ["approve", "reject"],
                "when": lambda r: r.tool_call["args"]["amount"] > 50,  # over 50 EUR: ask a person
            }
        })
    ],
    checkpointer=InMemorySaver(),  # interrupts need a checkpointer
)
config = {"configurable": {"thread_id": "ticket-42"}}


agent.invoke({"messages": [("user", "Refund 120 EUR of ch_2")]}, config)  # paused for review
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)  # an approval by TENBRI_APPROVER


agent.invoke({"messages": [("user", "Refund 900 EUR of ch_3")]}, config)  # paused for review
agent.invoke(Command(resume={"decisions": [{"type": "reject", "message": "over the limit"}]}), config)


for m in tenbri.default().query(action="issue_refund"):
    e = m.entry
    who = f"{e['principal']['kind']}:{e['principal']['id']}"
    print(e["seq"], e["kind"], e.get("phase", ""), e.get("status", ""), who, e["action"]["target"])
Terminal window
export TENBRI_STREAM=./audit TENBRI_APPROVER=person:alice
python refunds.py

ainvoke and astream record the answers the same way (on Python 3.11 or later, which LangGraph’s interrupt() needs under asyncio).

With a scripted model asking for each refund in turn, both print:

3 approval   person:alice stripe:ch_2
4 decision intent  agent:support stripe:ch_2
5 decision outcome ok agent:support stripe:ch_2
6 decision intent  agent:support stripe:ch_3
7 decision outcome refused agent:support stripe:ch_3

The approval (seq 3) is written before the intent that names it (seq 4). The rejected refund never ran: its intent and refused outcome are written as the graph resumes.

  • The Command’s resume can be {"decisions": [...]}, a single decision, or a mapping of interrupt ids to those when several interrupts are pending. Per-call and batched interrupts both work. In batched mode, LangChain.js runs none of a batch’s calls when one is rejected; their approvals stay in the stream, unused.
  • When the person who answers changes per call, pass approver as a function of the tool and its arguments: tenbri.instrument_langchain(approver=lambda tool, args: f"person:{reviewer_of(args)}") (instrumentLangChain({ approver: (tool, args) => ... })), where reviewer_of is your own lookup.
  • With no approver at all, an approval writes no entry and logs tenbri: c2 approved by nobody known: set TENBRI_APPROVER; the call then runs with an intent that names no approval.

Choose where entries go

Section titled “Choose where entries go”

The stream is wherever TENBRI_STREAM points; the code does not change.

Location TENBRI_STREAM Credentials
A local directory ./audit or file:///var/lib/audit/support none
AWS S3, MinIO s3://bucket/prefix AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION; AWS_ENDPOINT_URL (and AWS_ALLOW_HTTP=true for plain HTTP) for MinIO
Cloudflare R2 r2://bucket/prefix CLOUDFLARE_ACCOUNT_ID, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY

Signing keys stay on the machine that writes; only entries go to the bucket, with "durability":"replicated". See Storage for every variable. The bindings do not write through a tenbrid server; to send entries to tenbrid over REST, record by hand as in any other framework.

Read back and verify

Section titled “Read back and verify”

Anyone with read access to the stream and its fingerprint can check it; no keys needed. Pin the fingerprint you got at creation: without one, tenbri verify only reports consistent and NOT pinned: this only shows the stream agrees with its own genesis key.

Terminal window
tenbri verify ./audit --key sha256:5ab4e130…                    # the whole stream
tenbri query ./audit --action issue_refund --key sha256:5ab4e130…
tenbri query ./audit --principal person:alice --key sha256:5ab4e130…
tenbri schema check ./audit                                     # entries against tenbri.toml
tenbri tail ./audit --key sha256:5ab4e130…                      # follow new entries

Each match carries verification: verified when a key is pinned, unpinned when not. See the CLI reference.

Test your agent offline

Section titled “Test your agent offline”

Give the agent a chat model that plays back prepared messages, tool calls included. The graph runs as usual, with no API key and no network, and you assert on what the stream holds. Give init() a stream and keys in a temporary directory, so tests never touch your real keys:

test_agent.py
import pytest
import tenbri
from langchain.agents import create_agent
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_core.tools import tool




class FakeModel(GenericFakeChatModel):
    """Plays back AI messages, tool calls included: no API key, no network."""


    def bind_tools(self, tools, **kwargs):
        return self




def calls(*tool_calls):
    return AIMessage("", tool_calls=[{"name": n, "args": a, "id": i} for n, i, a in tool_calls])




@tool
def issue_refund(charge: str, amount: int) -> str:
    """Refund `amount` EUR of a charge."""
    if amount > 500:
        raise tenbri.Refused("over the refund ceiling")
    return "refunded"




def test_refunds_are_recorded(tmp_path):
    # A stream and keys of its own: the test never touches your real keys directory.
    with tenbri.init(tmp_path / "audit", keys=tmp_path / "keys") as log:
        tenbri.instrument_langchain(actions={"issue_refund": {"target": "stripe:{charge}"}})
        model = FakeModel(messages=iter([
            calls(("issue_refund", "c1", {"charge": "ch_1", "amount": 20})),
            AIMessage("Refunded."),
            calls(("issue_refund", "c2", {"charge": "ch_2", "amount": 900})),
        ]))
        agent = create_agent(model, tools=[issue_refund], name="support")


        agent.invoke({"messages": [("user", "Refund ch_1")]})
        with pytest.raises(tenbri.Refused):  # ToolNode re-raises what the tool raised
            agent.invoke({"messages": [("user", "Refund ch_2")]})


        entries = [m.entry for m in log.query(action="issue_refund")]
        assert [(e["phase"], e.get("status"), e["action"]["target"]) for e in entries] == [
            ("intent", None, "stripe:ch_1"),
            ("outcome", "ok", "stripe:ch_1"),
            ("intent", None, "stripe:ch_2"),
            ("outcome", "refused", "stripe:ch_2"),
        ]
        assert log.verify(log.fingerprint).ok
Terminal window
python -m pytest -q test_agent.py
.                                                                        [100%]
1 passed in 0.36s

The two tests show a difference between the frameworks: Python’s ToolNode re-raises a tool’s exception (unless you set handle_tool_errors), so invoke raises Refused. LangChain.js hands the error to the model as a tool message, and invoke returns. Both record refused.

Troubleshooting

Section titled “Troubleshooting”

These messages come from real runs; paths are shortened.

Message Cause What to do
InvalidRequestError: no stream: pass tenbri.init(stream) or set TENBRI_STREAM to a directory or s3://bucket/prefix (TypeScript: pass init({ stream })) init() found no stream. Set TENBRI_STREAM, or pass the stream to init().
Python: NoSessionKeyError: read-only: no stream key for tnb1:… in …/keys raised by invoke, after LangChain logs Error in _Instrumented.on_tool_start callback: NoSessionKeyError(...) The intent could not be written (here the stream’s keys are not in this keys directory), so the tool did not run. Set TENBRI_KEYS to the directory the stream was created with. Any write failure surfaces this way.
TypeScript: LangChain logs Error in handler TenbriCallbackHandler, handleToolStart: NoSessionKeyError: read-only: no stream key for tnb1:… in …/keys, and the model gets the error as a tool message The same; createAgent hands the error to the model instead of throwing. The tool did not run. The same.
InvalidRequestError: 'stripe:{charge_id}': no argument 'charge_id' (TypeScript: "stripe:{charge_id}": no argument charge_id) The target names an argument the tool does not have. Match the placeholder to the tool’s schema.
InvalidRequestError: tenbri.toml: actions.issue_refund: policy: expected ref@version A policy without its version, raised by init(). Write policy = "policy:refund@14".
tenbri: c2 approved by nobody known: set TENBRI_APPROVER (a warning) An approval in Command(resume=...) with no approver configured: no approval entry is written. Set TENBRI_APPROVER, or pass approver.
tenbri: issue_refund ran unrecorded (fail-open): … (a warning) fail_open is on and the intent could not be written; the tool ran. Fix the stream; the call is not in the record.
several interrupts are pending: key the answers by interrupt id One answer while several interrupts wait. Pass a mapping of interrupt ids to answers in resume.
Entries say agent:langgraph The agent has no name, and nothing else says who acts. Pass name= to create_agent (name to createAgent), or set TENBRI_AGENT.
ImportError: …: pip install --pre tenbri-langgraph tenbri.instrument_langchain() without the adapter installed. Install it.

An outcome that cannot be written is logged (tenbri: outcome of … not recorded: …), not raised: the tool already ran. A process that dies mid-call leaves its intent open; the next writer of the stream records it as unknown. A LangGraph interrupt() inside a tool ends that call as unknown: the tool runs again from the start on resume, as a new decision.

What is written

Section titled “What is written”
What happens Entry
A tool starts An intent: principal the agent, action.type, action.target and rule from the mapping, the arguments pinned by keyed digest (langchain:tool_call/<id>@args), ext.langchain with the run id, tool call id and thread id. Durable before the tool runs.
The tool returns The outcome ok (error if it returned an error ToolMessage, from handle_tool_error)
The tool raises The outcome error, refused for tenbri.Refused, unknown for a LangGraph interrupt
Command(resume=...) with approve or edit An approval by the approver; the call’s intent names it
Command(resume=...) with reject An intent and a refused outcome; ext.langchain holds decision and reviewer
Command(resume=...) with respond A note by the reviewer

Arguments and results are never stored. The digests are keyed by the stream’s local key store, per subject, so no secret goes in your code.

The entries of the approvals run above, as stored (signatures and hashes shortened). Alice’s approval and the intent that names it:

{
  "action": { "target": "stripe:ch_2", "type": "issue_refund" },
  "ext": {
    "langchain": { "checkpoint_id": "1f1cca0d-3690-6756-8001-6678f20f6333", "thread_id": "ticket-42", "tool_call_id": "c2" }
  },
  "id": "01M5D18YJGSE2WFVT4BVC1B88Q",
  "inputs": [
    { "digest": "hmac-sha256:eb0e8fcf…", "ref": "langchain:tool_call/c2", "version": "args" }
  ],
  "kind": "approval",
  "prev": "sha256:569e8036…",
  "principal": { "id": "alice", "kind": "person" },
  "rule": { "ref": "policy:refund", "result": "allow", "version": "14" },
  "seq": 3,
  "sig": "ed25519:b1b356f7…",
  "signer": "sess:2026-10-20:0e9ebfd2bddb",
  "stream": "tnb1:00b30bb2fd940776b7d49a27e86a8173",
  "ts": "2026-10-20T16:10:53.907Z",
  "v": 1
}
{
  "action": { "target": "stripe:ch_2", "type": "issue_refund" },
  "approval": {
    "by": { "id": "alice", "kind": "person" },
    "entry": "sha256:cafedce0…"
  },
  "durability": "local",
  "ext": {
    "langchain": { "run_id": "01a15a14-7a65-72bc-af30-cde5e2188b16", "thread_id": "ticket-42", "tool_call_id": "c2" }
  },
  "id": "01M5D18YK5JFD9YB5KY1Q1XR31",
  "inputs": [
    { "digest": "hmac-sha256:eb0e8fcf…", "ref": "langchain:tool_call/c2", "version": "args" }
  ],
  "kind": "decision",
  "phase": "intent",
  "prev": "sha256:cafedce0…",
  "principal": { "id": "support", "kind": "agent" },
  "rule": { "ref": "policy:refund", "result": "allow", "version": "14" },
  "seq": 4,
  "sig": "ed25519:383aa576…",
  "signer": "sess:2026-10-20:0e9ebfd2bddb",
  "source": "sdk",
  "stream": "tnb1:00b30bb2fd940776b7d49a27e86a8173",
  "ts": "2026-10-20T16:10:53.929Z",
  "v": 1
}

The rejected refund’s intent, written as the graph resumed, with the reviewer:

{
  "action": { "target": "stripe:ch_3", "type": "issue_refund" },
  "durability": "local",
  "ext": {
    "langchain": { "checkpoint_id": "1f1cca0d-374c-6316-8008-18783965d14f", "decision": "reject", "reviewer": "person:alice", "thread_id": "ticket-42", "tool_call_id": "c3" }
  },
  "id": "01M5D18YMGTRWPM9P914DJN0RS",
  "inputs": [
    { "digest": "hmac-sha256:1c72c83f…", "ref": "langchain:tool_call/c3", "version": "args" }
  ],
  "kind": "decision",
  "phase": "intent",
  "prev": "sha256:0a1af256…",
  "principal": { "id": "support", "kind": "agent" },
  "rule": { "ref": "policy:refund", "result": "allow", "version": "14" },
  "seq": 6,
  "sig": "ed25519:a1b1d13c…",
  "signer": "sess:2026-10-20:0e9ebfd2bddb",
  "source": "sdk",
  "stream": "tnb1:00b30bb2fd940776b7d49a27e86a8173",
  "ts": "2026-10-20T16:10:53.971Z",
  "v": 1
}

Its outcome (seq 7) has "status": "refused" and the same principal, agent:support. See the entry format for every field.

Advanced: lower-level API

Section titled “Advanced: lower-level API”

The instrumentation is built on TenbriCallbackHandler, a LangChain callback handler on a stream you opened yourself, passed in callbacks, with no default stream and no environment. Use it to hold several streams, or to record some runs and not others. It writes the same entries.

handler.py
from langchain.agents import create_agent
from langchain_core.tools import tool
from tenbri import Tenbri
from tenbri_langgraph import TenbriCallbackHandler




@tool
def issue_refund(charge: str, amount: int) -> str:
    """Refund `amount` EUR of a charge."""
    return f"refunded {amount} EUR of {charge}"




log = Tenbri.open("./audit", create=True)  # synchronous: Tenbri, not AsyncTenbri
handler = TenbriCallbackHandler(log, target=lambda tool, args: f"stripe:{args['charge']}")


agent = create_agent("anthropic:claude-sonnet-5-5", tools=[issue_refund], name="support")
agent.invoke({"messages": [("user", "Refund 20 EUR of ch_1")]}, {"callbacks": [handler]})
print([(m.entry["phase"], m.entry["action"]) for m in log.query(action="issue_refund")])
log.close()

Both print the intent and the ok outcome of issue_refund on stripe:ch_1. The handler takes principal, tools, target and inputs (functions of the tool and its arguments), describe (any of action, target, rule, subject at once), approver and fail_open (failOpen). Without the instrumentation, record a person’s answer as you resume: agent.invoke(handler.resume(agent, config, answer, by="person:alice"), config) (await handler.resume(agent, config, answer, { by: 'person:alice' }) in TypeScript) records it and returns the Command. For an interrupt() of your own, call handler.approve(tool, args, by=..., tool_call_id=...) (await handler.approve(tool, args, { by, toolCallId })) once the person has answered.