LangChain

Add the HESO callback handler to a LangChain or LangGraph agent. From then on every tool the agent invokes is gated against your policy and signed into an Action Receipt. Tool definitions do not change; you add one callback.

A LangChain agent picks which tools to run and with what arguments. HESO gates that boundary. Attach HesoCallbackHandler as a callback and every tool the agent invokes is evaluated against your policy before it executes, then recorded as a signed Action Receipt. The decision is one of allow, block, redact, or route to a human.

The callback handler#

HesoCallbackHandler is a LangChain BaseCallbackHandler. LangChain calls on_tool_start before a tool runs and on_tool_end after, so the handler gates the call on the way in and records the result on the way out. When the agent is about to run a tool, HESO:

• captures the call as a tool_call action, with the tool name in tool_name and the arguments as fields;
• evaluates it against your policy and resolves a decision;
• signs the result into a receipt and appends it to the tamper-evident audit chain;
• lets the tool run, or refuses it, per that decision.

After the tool returns, on_tool_end binds the result to the gated action as a follow-up receipt, so the trail records what happened, not only what was allowed. This is best-effort: a recording failure never reverses the allow decision the gate already made.

LangChain is an optional dependency. A plain import heso does not require it; the handler loads the real BaseCallbackHandler only when you reference HesoCallbackHandler.

Wire it up#

Call heso.init() once at startup, then pass the handler in the callbacks= list your agent or executor already accepts. Which LangChain classes you use is up to you; the hook is always the standard callbacks= argument.

import heso
from heso import HesoCallbackHandler
from langchain.agents import AgentExecutor  # your agent setup, unchanged

heso.init()

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[HesoCallbackHandler()],
)

executor.invoke({"input": "refund order 4821"})

Construct the handler once and reuse it across runs. It reads the active heso.init() config on every call, so the same workflow and account labels apply as for the decorators and the proxy. To label the actions a run produces, wrap the call in heso.step(...). Every action gated inside the block carries that label, which makes a run easy to find later.

import heso

with heso.step(workflow="support-refunds", account="acct_42"):
    executor.invoke({"input": "refund order 4821"})

For a destructive-tool agent, pass HesoCallbackHandler(verb=Verb.DELETE) to record every gated call under the delete verb. This is an escalate-only hint; the engine still classifies each call structurally.

LangGraph#

LangGraph reuses LangChain’s callback system, so the same handler gates LangGraph tool nodes. Pass it per-invocation in config={"callbacks": [...]}.

from heso import HesoCallbackHandler

# Same handler. LangGraph shares LangChain's callback system,
# so pass it per-invocation in config.
graph.invoke(
    {"messages": [...]},
    config={"callbacks": [HesoCallbackHandler()]},
)

LangGraph also has a durable human-in-the-loop primitive. The adapter bridges it to HESO’s suspend and resume so a high-stakes node parks behind a signed approval and resumes once. See Framework adapters for heso_interrupt_for_approval and the approval bridge.

What gets captured#

Each tool the agent invokes becomes one tool_call action. The tool name lands in tool_name, and the call arguments become the action’s fields. Your policy then resolves one of the four decisions:

• allow — the tool runs and a receipt is signed.
• block — the tool is refused and never executes.
• redact — named fields are redacted before the receipt is signed.
• require_approval — the call is routed to a human approver before it can proceed.

See Actions & verbs for the full shape each tool call records, and Policy & decisions for how first-match-wins rules pick the decision.

Combine with heso.wrap#

The callback gates the agent’s tools. To gate the model calls as well, wrap the underlying LLM client with heso.wrap(...). The two run side by side: the wrapped client gates each completion as an llm_call, and the callback gates each tool_call.

import heso
from heso import HesoCallbackHandler
from langchain_openai import ChatOpenAI

heso.init()

# Gate the model calls.
llm = heso.wrap(ChatOpenAI(model="gpt-4o"))

# Gate the tools the agent runs.
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[HesoCallbackHandler()],
)

For wrapping an LLM client directly, including the OpenAI and Anthropic clients, see OpenAI & Anthropic.

Blocked tools#

When policy blocks a tool, the handler raises BlockedError instead of running it (a suspension raises SuspendedError). The handler sets raise_error = True so LangChain propagates the refusal rather than swallowing it and running the tool ungated. A receipt with decision set to block is still signed and appended to the audit chain, so the refusal is recorded as evidence.

from heso import BlockedError

try:
    executor.invoke({"input": "delete every customer record"})
except BlockedError as e:
    # Policy refused the tool before it ran. The tool never executed;
    # a receipt with decision="block" was still signed and audited.
    print("tool refused:", e)

Next steps#