Python — heso

heso is the gating SDK — decorators and a proxy that sit between your agent and the side effects it wants to take. It bundles the Rust core as the heso._core wheel, so every gate runs in-process: no subprocess, and no crypto rewritten in Python. The same core powers the Node and browser surfaces, so a verdict is byte-identical wherever it runs.

This page is a reference. If you are starting from scratch, read the Python quickstart first, then keep this open. For the command-line tools shipped alongside the package, see the CLI reference.

Install#

The package is on PyPI and requires Python 3.10 or newer. It bundles the Rust core as the heso._core wheel, so there is no separate binary to install.

pip install heso

Installing heso also installs the heso console script — documented in 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 configuration; every decorator, the proxy, and heso.process read from it. Gating needs no binary, since the Rust core ships as an in-process wheel.

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

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

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"])  # commit-and-reveal redaction first

import heso heso.init() @heso.tooldef transfer(payee: str, amount_usd: int) -> str:    # The call is captured as a tool_call, evaluated against policy,    # and signed into a receipt before this body runs.    return stripe.transfers.create(destination=payee, amount=amount_usd) # Redact named fields (commit-and-reveal) before signing.@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.destructivedef delete_member(member_id: str) -> None:    # verb = delete. Dangerous lanes carry a pinned floor, so an    # unmatched delete routes to a human approver by default.    db.members.delete(member_id)

@heso.action   # assemble an Action yourself

import hesofrom heso import Action, Verb heso.init() @heso.actiondef export_report(rows: list[dict]) -> str:    # Assemble the Action yourself when a decorator's defaults    # don't fit — this is the lower-level building block.    action = Action(verb=Verb.data_export, tool_name="reports.export",                    fields={"row_count": str(len(rows))})    outcome = heso.process(action)    return write_csv(rows)

Scoping#

Use heso.step to tag a block of actions with a specific workflow label — for example, a single agent run — without threading the label through every call.

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

import heso heso.init() with heso.step(workflow="run-42"):    transfer("Globex LLC", 4200)   # scoped to workflow "run-42"    notify_finance()               # same workflow label

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 can keep using the client’s own API.

heso.wrap(client) -> proxy

import hesofrom 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 hesofrom 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 lets you capture an action, save its action hash, wait for a decision elsewhere, and then resume. These are the values and callables it exposes:

• configure — set up the suspend/resume behavior for the active config.
• gated — wrap a call so it participates in the suspend/resume flow.
• gate and gate_async — gate an action, synchronously or in an async context.
• resume — continue a previously suspended action once a decision exists.
• decision and append_decision — read the decision for a suspended action, or record one.
• current_action_hash — the action hash of the action currently in flight, used to correlate a suspension with its later resume.
• The values Gate, ResumeOutcome, SUSPENDED, DENIED, Paused, and ContextLost — the result types and sentinels the layer returns.

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’ll import alongside these include Config, Action, Outcome, OutcomeKind, Verb, and RedactStrategy.

LangChain#

To gate tool calls inside a LangChain agent, pass heso.HesoCallbackHandler (lazily imported) as a callback. The handler intercepts tool invocations and runs them through the same gate as the decorators. The full setup is in the LangChain guide.