Python and TypeScript SDK

The tenbri package for Python and Node.js writes through the same Rust core as the tenbri CLI and tenbrid. This page covers version 0.1.0-beta.3.

pip install --pre tenbri

It has two levels:

• The default stream: init() opens one stream for the process, decision records each call of a function, agent() says who acts, and one line instruments an agent framework. Most applications need only this.
• The stream API: Tenbri.open and the methods on a stream (run, decision, read, intent, outcome, verify, query). The default stream is one of these, so all of them work on it too.

initCopy Anchor Link

Opens the process's default stream, creating the stream and its keys if there is none, and closes it (flushing) when the process exits.

def init(
    stream: str | os.PathLike[str] | None = None,
    *,
    keys: str | os.PathLike[str] | None = None,
    agent: Principal | None = None,
    approver: Principal | None = None,
    schema: str | os.PathLike[str] | None = None,
    key: str | None = None,
    digest_key: bytes | None = None,
    fail_open: bool = False,
) -> Default

An argument given wins over its environment variable:

• The stream is opened with create: the first init() on an empty location makes the stream and its keys, like tenbri init. Later calls open it.
• The schema file is checked as tenbri schema check checks it; an invalid one fails init() with the same message. The SDKs do not check entries against it.
• Calling init() again closes the previous default and opens the new one.
• Python: init() returns the open stream, and closes it at interpreter exit (atexit). with tenbri.init(...) as log: closes it at the end of the block instead.
• TypeScript: init() returns a Promise; awaiting it is optional, because a decision waits for it (and throws what it threw). It is closed on the process's beforeExit. await using log = await init(...) closes it at the end of the scope.
• Without a stream: InvalidRequestError, no stream: pass tenbri.init(stream) or set TENBRI_STREAM to a directory or s3://bucket/prefix (TypeScript: pass init({ stream })).

The value returned is a Default: a Tenbri stream with the settings the rest of the process uses, agent, approver, actions (the action schemas of tenbri.toml, by type) and fail_open (failOpen). In Python, .aio is the same stream as an AsyncTenbri.

tenbri.default() (Python) and defaultStream() (TypeScript, a Promise) return it; before init(), they fail with InvalidRequestError: Tenbri is not initialized: call tenbri.init() first (call init() first).

import tenbri

tenbri.init()  # TENBRI_STREAM, TENBRI_AGENT...

with tenbri.init("audit/support", agent="agent:support") as log:  # scoped to the block
    print(log.fingerprint)

decisionCopy Anchor Link

Records each call of a function as a decision on the default stream: the intent is durable before the function runs, then the outcome follows.

def decision(
    action: str,
    target: str | Callable[..., str] | None = None,
    *,
    policy: str | None = None,
    inputs: Iterable[Pin] | Callable[..., Iterable[Pin]] | None = None,
    subject: str | Callable[..., str] | None = None,
    principal: Principal | None = None,
    fail_open: bool | None = None,
    **fields: Any,
) -> Callable[[F], F]

OutcomesCopy Anchor Link

If the intent cannot be written, the function does not run and the TenbriError propagates (fail closed), unless fail_open is set. If the outcome cannot be written, the function's result stands: a warning is logged, and the open intent is reconciled as unknown by the stream's next writer.

TemplatesCopy Anchor Link

target and subject strings are templates: each {name} is replaced with an argument, and {name.key} reads into it (a key of a mapping, or an attribute in Python). A missing argument fails the call, before the function runs, with InvalidRequestError: 'x:{missing}': no argument 'missing' (TypeScript: "x:{missing}": no argument missing).

• Python binds the template to the function's parameters by name, defaults included: "stripe:{charge}" is the charge argument, however it was passed.
• TypeScript has no parameter names: {charge} reads the first argument's charge member, and {0}, {1} are positional arguments ({0.id} reads the first argument's id).

Who actsCopy Anchor Link

The principal is, in order: principal=; else the innermost agent() scope; else init(agent=...) or TENBRI_AGENT. With none, the call fails with InvalidRequestError: no principal: pass agent= to tenbri.init(), set TENBRI_AGENT, or use with tenbri.agent("agent:...") (TypeScript: no principal: pass agent to init(), set TENBRI_AGENT, or run in agent('agent:...', fn)). Nothing is guessed.

Inputs and digestsCopy Anchor Link

Each input is pinned with the keyed digest of its content: under digest_key if init() was given one, else under the decision's subject's data key (the tenant's when there is no subject) from the stream's local key store, made on first use. Erasing the subject with tenbri erase leaves its digests unconfirmable (Privacy).

In Python, an intent also gets trace from the active OpenTelemetry span when OpenTelemetry is installed and tracing.

ExampleCopy Anchor Link

import asyncio
import tenbri

tenbri.init("audit/support", agent="agent:support")

@tenbri.decision(
    "issue_refund",
    target="stripe:{charge}",
    policy="policy:refund@14",
    subject="{customer}",
    inputs=[("policy:refund", "14", b"refunds under 500 are automatic")],
)
def refund(charge: str, amount: int, customer: str) -> str:
    if amount > 500:
        raise tenbri.Refused("over the ceiling")
    return "re_1"

@tenbri.decision("reply_ticket", target="zendesk:ticket/{ticket.id}")
async def reply(ticket: dict) -> None: ...

print(refund("ch_123", 20, "subj_7f3c91d2"))
try:
    refund("ch_124", 900, "subj_7f3c91d2")
except tenbri.Refused as e:
    print("Refused:", e)
with tenbri.agent("agent:billing"):
    asyncio.run(reply({"id": 42}))

for m in tenbri.default().query(action="issue_refund"):
    e = m.entry
    print(e["seq"], e["phase"], e.get("status"), e["action"]["target"], e.get("rule"))

re_1
Refused: over the ceiling
3 intent None stripe:ch_123 {'ref': 'policy:refund', 'result': 'allow', 'version': '14'}
4 outcome ok stripe:ch_123 None
5 intent None stripe:ch_124 {'ref': 'policy:refund', 'result': 'allow', 'version': '14'}
6 outcome refused stripe:ch_124 None

agentCopy Anchor Link

Sets the principal for a scope: decisions in it, and in the tasks, threads and async work it starts, are by that principal unless they name their own.

@contextlib.contextmanager
def agent(principal: Principal) -> Iterator[None]

with tenbri.agent("agent:billing"):
    refund("ch_123", 20, "subj_7f3c91d2")

A Principal is "agent:support", "person:alice", "service:billing", a bare id ("support", an agent), or {kind, id} with kind one of agent, person, service.

RefusedCopy Anchor Link

Raise (throw) it inside a decision when your own policy refuses the action: the outcome is recorded as refused, and it propagates.

class Refused(Exception): ...

raise tenbri.Refused("over the ceiling")

It works the same in the stream API's decisions (log.decision, run.decision).

Framework instrumentationCopy Anchor Link

One call records every tool call of an agent framework on the default stream: an intent before the tool runs, its outcome after, and approvals through the framework's own approval flow. Call it once, after init().

In Python the functions are in tenbri and import the adapter package; without it they raise ImportError naming the pip install --pre command. In TypeScript they are exported by the adapter packages. instrumentClaudeAgentSdk wraps the Options of one query() and returns them; the others patch the framework for every run from then on.

Options, all optional:

import tenbri

tenbri.init()
tenbri.instrument_claude_agent_sdk()  # or instrument_openai_agents(), instrument_langchain()

What each framework records, and its lower-level hooks, are in Adapters.

describeCopy Anchor Link

How a tool call becomes intent fields, exposed for your own integrations:

def describe(tool: str, args: Mapping[str, Any], actions: Actions | None = None) -> dict[str, Any]

It returns action (when it differs from the tool's name), target, subject and rule, from actions over the default stream's action schemas. The rules and an example are in Action schemas.

Tenbri.openCopy Anchor Link

The stream API: open a stream yourself, to hold several, or to see every call.

class Tenbri:
    @classmethod
    def open(
        cls,
        url: str | os.PathLike[str],
        *,
        keys: str | os.PathLike[str] | None = None,
        key: str | None = None,
        create: bool = False,
        digest_key: bytes | None = None,
    ) -> Tenbri

class AsyncTenbri:
    @classmethod
    async def open(...) -> AsyncTenbri  # the same arguments

A stream has id (tnb1:...), fingerprint (what a verifier pins) and writable.

MethodsCopy Anchor Link

run.read digests under the tenant's data key (it has no subject); to digest under a subject's key, pass digest=log.digest(content, subject=...), or use inputs on the high-level decision.

from tenbri import Tenbri

log = Tenbri.open("audit/support", create=True)
print("pin this:", log.fingerprint)

with log.run("agent:support-assistant") as run:
    run.read("policy:refund", "14", b"refunds under 50 EUR are automatic")
    with run.decision("issue_refund", "stripe:ch_123") as d:
        pass  # issue the refund here; raise Refused(...) when your policy says no
print(d.intent["inputs"])

report = log.verify(log.fingerprint)
print(report.ok, report.entries)
for m in log.query(inputs=["policy:refund@14"]):
    print(m.verification, m.entry["action"])
log.close()

Python prints (the digest differs per stream, its data key being random):

pin this: sha256:026cfe12eeed02fe2fcde877832504ebe6289de02ac812f0fdbf07f0a4a1397b
[{'digest': 'hmac-sha256:9e7aa6048ae664013cd18b24200969866b3db8fba53fb1f4a71a8f8a05c043a7', 'ref': 'policy:refund', 'version': '14'}]
True 4
unpinned {'target': 'stripe:ch_123', 'type': 'issue_refund'}

query says unpinned because the stream was opened without key; verify was given the fingerprint, so report.ok is true.

ErrorsCopy Anchor Link

Every failure is a TenbriError with a stable code, shared with tenbrid (API), reason (the format v1 code, where there is one, Errors) and id (the entry being written: retry with it).

With fail_open, a decorated function whose intent fails with any of these runs anyway, and a warning is logged:

tenbri: noop runs unrecorded (fail-open): read-only: no stream key for tnb1:7975387dece6df1938db869d9f2844d2 in otherkeys