Python — heso

Reference for the heso Python package. Decorators and a client proxy capture each action in-process, run it through the Rust core, and raise before an unauthorized side effect fires.

heso gates your agent from inside your process. It bundles the Rust core as the heso._core wheel, so every gate runs in-process: no subprocess, no crypto rewritten in Python. The same core powers the Node and browser surfaces, so a verdict is byte-identical wherever it runs.

New to HESO? Start with the Python quickstart, then keep this open. For the command-line tools that ship with the package, see the CLI reference.

Common tasks#

The four things most people come here to do, and where each one lives:

• Gate a tool — decorate the function with @heso.tool. The call is captured as a tool_call and checked against policy before the body runs.
• Gate an LLM client — wrap it with heso.wrap(client). Every completion is gated as an llm_call; the rest of the client API is untouched.
• Redact a field — pass @heso.tool(redact=["field"]). The value is commit-and-reveal redacted before signing, so cleartext stays local.
• Route to a human — use @heso.gated for durable suspend/resume. A first call parks the action; a later call resumes it once an approver decides.

Install#

Install heso from PyPI. It needs Python 3.10 or newer and bundles the Rust core as the heso._core wheel, so there is no separate binary.

pip install heso   # or: uv add heso

Installing the package also installs the heso console script (see the CLI reference). To scaffold a project — mint an operator identity and write a starter heso.toml — run heso init.

Initialize#

Call heso.init() once at process start. It resolves and installs the active config that every decorator, the proxy, and heso.process read from. It fails fast if no heso.toml can be found.

heso.init(*, project_root=None, binary=None, workflow=None,
          account=None, clock_override=None, timeout=None,
          blocking=None) -> Config

import heso

# explicit args > env vars > heso.toml > defaults
cfg = heso.init(
    workflow="vendor-payouts",
    account="acct_19",
    blocking=True,  # raise on a blocked/suspended action (the default)
)

Decorators#

Decorators are the common way to gate. Each wraps a function so the call is captured as an action, checked against policy, and signed into a receipt before the body runs. With blocking on (the default), a blocked or routed action raises instead of running.

@heso.tool                      # verb = tool_call
@heso.tool(redact=["api_key"])  # redact fields before signing

import heso

heso.init()

@heso.tool
def transfer(payee: str, amount_usd: int) -> str:
    # Captured as a tool_call, checked against policy, and signed
    # into a receipt before this body runs.
    return stripe.transfers.create(destination=payee, amount=amount_usd)

# Redact named fields before signing (commit-and-reveal).
@heso.tool(redact=["api_key"])
def call_vendor(api_key: str, endpoint: str) -> dict:
    return vendor.request(endpoint, key=api_key)

@heso.destructive   # verb = delete

import heso

heso.init()

@heso.destructive
def delete_member(member_id: str) -> None:
    # verb = delete. delete carries a pinned floor, so an unmatched
    # delete routes to a human approver by default.
    db.members.delete(member_id)

@heso.action("domain.action_id")   # declare a catalog action

import heso

heso.init()

# Declare a catalog action ("domain.action_id"). It gates with the
# catalog's coarse verb (here, payment) and stamps action.domain /
# action.action on the receipt. An unknown id raises ValueError at
# decoration time.
@heso.action("payment.authorize_payment")
def charge_card(amount: int, to: str) -> str:
    return stripe.charges.create(amount=amount, destination=to)

Scoping#

Use heso.step to tag a block of actions with a workflow or account label — for example a single agent run — without threading the label through every call. Steps nest: an inner override wins, an unset field falls through to the outer step, then to the config default. The scope is task- and thread-local.

with heso.step(workflow="run-42", account="acct_19"):
    ...

import heso

heso.init()

with heso.step(workflow="run-42", account="acct_19"):
    transfer("Globex LLC", 4200)   # scoped to workflow "run-42"
    notify_finance()               # same workflow + account labels

Wrapping clients#

heso.wrapreturns a transparent proxy around a client. It gates the calls that produce side effects and forwards everything else unchanged, so you keep using the client’s own API.

heso.wrap(client) -> proxy

import heso
from openai import OpenAI

heso.init()

client = heso.wrap(OpenAI())

# .create at the leaf is gated as an llm_call; the call kwargs
# become the action fields. Everything else forwards verbatim.
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "summarize Q3"}],
)

Imperative API#

Beneath the decorators sits heso.process — the escape hatch for building and gating an action by hand. It requires that heso.init() has run first.

heso.process(action: Action) -> Outcome

import heso
from heso import Action, Verb, OutcomeKind

heso.init()

action = Action(verb=Verb.payment, tool_name="stripe.transfers.create",
                fields={"amount_usd": "4200", "payee": "Globex LLC"})

outcome = heso.process(action)
if outcome.kind is OutcomeKind.BLOCKED:
    raise RuntimeError("policy blocked this payment")

Suspend / resume#

For long-running agents, an action routed to a human pauses instead of failing. The suspend/resume layer captures an action, parks it under a session, and runs the body once an approver decides. A cleared action re-mints to an L1 receipt. See Human approval for the full flow.

@heso.gated#

The primary suspend/resume surface. The decorated body runs only after the action is approved and this caller wins the local exactly-once claim. Thread a session with _heso_session= (auto-generated when omitted); calling the function again with the same session is the resume.

@heso.gated(*, sla="1d", on_timeout="deny",
            tool_version=None, expires_at=None,
            raise_on_suspend=False)

import heso

heso.init()

@heso.gated
def wire_funds(payee: str, amount_usd: int) -> str:
    # Runs ONLY after an approver approves. First call parks and
    # returns heso.SUSPENDED; a later call with the same session resumes.
    return bank.wire(payee, amount_usd)

out = wire_funds("Globex LLC", 50_000, _heso_session="payout-7")
if out is heso.SUSPENDED:
    ...  # waiting on a human; call again later with the same session

resume, decision, append_decision#

The top-level helpers for advancing a session out of band:

• heso.resume(session) — re-read the head, apply the decision, fire-once-or-replay, and return a ResumeOutcome. A bare resume drives the engine lifecycle but cannot run the body; re-call the @gated function with the same _heso_session to run it.
• heso.decision(session) — read the current lifecycle state off the chain head (approved / denied / expired / escalated / completed / suspended), or None if there is no chain yet.
• heso.append_decision(session, kind, *, reason=None, decided_at=None) — append a decision link (approved / denied / escalated / expired) to a session’s chain.
• heso.current_action_hash() — the action hash in flight, exposed while a gated body runs, to correlate a suspension with its later resume.

import heso

heso.init()

# Record an approver decision out of band...
heso.append_decision("payout-7", "approved", reason="cfo ok")

# ...then advance the lifecycle. Re-call the @gated fn with the same
# session to actually run the body; bare resume() only drives the engine.
outcome = heso.resume("payout-7")

finalize_l1, finalize_quorum#

The out-of-band finalize path: when an approver co-signs through the cloud relay, the operator stitches the L1 together locally. Both verify the assembled receipt and push it to the outbox, superseding the suspended action.

• heso.finalize_l1(action_hash, suspended_content, relayed_parts) — assemble and finalize a single-approver L1receipt. Refuses unless the relayed record’s decision is approved. Raises OperatorKeyMismatchError if the operator key rotated since suspend.
• heso.finalize_quorum(action_hash, suspended_content, relayed_parts) — the k-of-n sibling. Every leg must be approved. A quorum is still L1, carrying a multi_approval block — not a higher level.

Errors#

The package raises a small, typed set of exceptions. The first two are how a refusal surfaces when blocking is on; the last two are setup and bridge failures.

Public types you import alongside these include Config, Action, Outcome, OutcomeKind, Verb, and RedactStrategy.

Framework adapters#

Beyond the decorators, heso ships drop-in adapters for the major agent frameworks. Each is exposed lazily, so a bare import heso never imports a framework you are not using. Every adapter runs each call through the same gate as the decorators, so the receipt is identical no matter which framework produced the action.

• LangChain & LangGraph — pass heso.HesoCallbackHandler() as a callback to gate every tool invocation. See the LangChain guide.
• CrewAI — heso.crewai (for example heso.crewai.heso_before_hook).
• OpenAI Agents SDK — heso.openai_agents.
• Claude Agent SDK — heso.claude_agent.
• Pydantic AI — heso.pydantic_ai.
• LangGraph — heso.langgraph for graph-node gating.
• MCP — heso.mcp (for example heso.mcp.wrap_call_tool) to gate Model Context Protocol tool calls.

Next steps#