Add Soterias to an agent

Keep your model, tools and framework. Start by measuring one function, then add an optional framework adapter. Use a configured native broker when an operation must be restricted. These are separate entry points: measurement requires no warrant; an observed callback does not acquire permissions.

This guide uses Python soterias==0.1.2 (native 0.1.0rc3) and TypeScript @soterias/core@0.1.0-rc.3. The earlier local and invoke APIs remain supported. Install the matching version before using the constructor, run, summary, or optional middleware.

Your first function

Save this complete program as first.py:

from soterias import Soterias

def invoice_total(amounts):
    return sum(amounts)

with Soterias(workflow="invoice-total") as sdk:
    total = sdk.run(invoice_total, [1200, 300])
    print(total)
    print(sdk.summary()["counts"]["raw"])

The first line of output is 1500. The second is the number of observations in this local workflow store, including previous runs. The self-contained start.py uses a new private store so its count is exactly one. No arguments or result bodies are stored.

The same full program in JavaScript (also valid TypeScript):

import {Soterias} from '@soterias/core';

const sdk = new Soterias({workflow: 'invoice-total'});
try {
  const total = sdk.run(amounts => amounts.reduce((sum, value) => sum + value, 0), [1200, 300]);
  console.log(total);
  console.log(sdk.summary().counts.raw);
} finally {
  sdk.close();
}

For async functions use await sdk.run_async(...) in Python or await sdk.runAsync(...) in TypeScript. A synchronous run returns immediately if your function returns a coroutine/Promise; it does not pretend to measure awaited work. Choose the async method for an async operation.

Components and names

What you need	Python	TypeScript	What it provides
A local client	`Soterias(workflow="name")`	`new Soterias({workflow: "name"})`	Private native observation store
Run your function	`sdk.run(function, ...)`	`sdk.run(function, ...)`	One call, original result/error, measured duration
Run an async function	`await sdk.run_async(function, ...)`	`await sdk.runAsync(function, ...)`	Awaited completion and native recording
Record reported model usage	`sdk.run_model(call, usage=extract)`	`sdk.runModel(call, {usage: extract})`	Explicit input/output/cache counts, with provenance
Read metrics	`sdk.summary()`	`sdk.summary()`	Native snapshot as an ordinary dictionary/object
Async storage access	`await sdk.summary_async()`	`await sdk.summaryAsync()`	Same snapshot through async storage access
Close and drain	`with` / `async with`, or `close` / `close_async`	`close` / `closeAsync`	Finish accepted recording work and close
Reusable verified graph	`Procedure`, `ProcedureMemory`	`Procedure`, `ProcedureMemory`	Native graph interpretation and evidence qualification
Protected effects	`TrustedSqliteExecutor`	`TrustedSqliteExecutor`	Current native admission and fixed broker routes
Independent approval	`TrustedOperationReviewer`	`TrustedOperationReviewer`	One-shot signed approval of an exact operation

local and invoke remain compatibility names. The beginner examples teach one path; advanced native components are introduced when they are needed.

An existing model or chain

run_model accepts any synchronous provider callback. The async variant accepts an async callback. Your callback owns provider configuration and credentials. The extractor maps explicit provider metadata, after native timing has stopped:

from soterias import ModelUsage, Soterias

def ask_with_metrics(chain, question):
    with Soterias(workflow="assistant") as sdk:
        answer = sdk.run_model(
            lambda: chain.invoke(question),
            usage=lambda message: ModelUsage(
                "my-provider",
                input_tokens=(message.usage_metadata or {}).get("input_tokens"),
                output_tokens=(message.usage_metadata or {}).get("output_tokens"),
            ),
        )
        return answer, sdk.summary()

This is a complete integration function; pass your real chain and question. The full local-model example also constructs the model and agent. Unknown usage stays unknown, and a provider count is not authenticated proof. Soterias does not infer model calls, prices or quality from the presence of a returned message.

LangChain agents: one middleware entry point

Install the optional extra and the provider you choose. LangChain is not a core SDK dependency. This follows LangChain's documented Python agent, TypeScript agent and middleware interfaces; Soterias does not replace the agent loop.

In Python, add this to your own create_agent call:

from langchain.agents import create_agent
from soterias import Soterias
from soterias.integrations.langchain import SoteriasMiddleware

def run_existing_agent(model, tools, messages):
    with Soterias(workflow="assistant") as sdk:
        agent = create_agent(model=model, tools=tools,
                             middleware=[SoteriasMiddleware(sdk, provider="my-provider")])
        response = agent.invoke({"messages": messages})
        return response, sdk.summary()

The TypeScript integration uses the framework's normal middleware list:

import {createAgent} from 'langchain';
import {Soterias} from '@soterias/core';
import {soteriasMiddleware} from '@soterias/core/integrations/langchain';

// In your application, model/tools/messages are your existing configured objects.
const sdk = new Soterias({workflow: 'assistant'});
try {
  const agent = createAgent({model, tools,
    middleware: [soteriasMiddleware(sdk, {provider: 'my-provider'})]});
  const response = await agent.invoke({messages});
  console.log(response);
  console.log(await sdk.summaryAsync());
} finally {
  await sdk.closeAsync();
}

For a fully defined runnable program, including typed tool, local provider, invocation, assertions and cleanup, read langchain_agent.py or langchain_agent.mts. The accompanying README has exact dependency installation and expected results. The docs-only Gmail recipe is a separate application example; no Gmail functionality is built into the SDK.

Read the metrics correctly

The adapter records each actual middleware model/tool handler, once. Native monotonic timing and SQLite persistence are shared by Python and TypeScript. The model message's explicit usage_metadata supplies token counts; a tool run without model metadata remains unmeasured for tokens. Missing usage is not zero.

summary()["counts"]["raw"] / summary().counts.raw counts observations. all_runs.duration_ns holds the native distribution. all_runs.input_tokens and all_runs.output_tokens retain separate provenance buckets and unmeasured counts. Quantities are decimal strings in the native schema, preserving large integers across Python and JavaScript. Read distribution.samples and sum with their basis. Convert nanoseconds to milliseconds by dividing by 1,000,000.

The native recorder does not retain prompts, tool arguments, message bodies or credentials automatically. Keep workflow/provider labels non-sensitive; chosen labels are metadata stored locally. Explicit exports remain local unless your application uploads them. LangChain/provider tracing and retention settings are separately yours. The examples disable no controls on your behalf; the test harness explicitly disables optional external tracing when running public synthetic samples.

Place observability middleware inside retry/cache middleware if you want the handlers that really execute. Wrapping the whole agent as well produces nested, overlapping timing; do not sum those durations as end-to-end latency. A returned tool call is not automatically a verified business outcome: unverified is the correct label until a real independent verification boundary exists. There is no general speed, token, price or quality guarantee.

Authority is outside the agent's context

Messages, scratchpads, tool results and another agent's handoff are untrusted proposal data. A message saying “approved”, a poisoned context saying “next run the payment”, or a model-generated warrant cannot grant authority. A poisoned context can still produce bad answers/proposals inside an already authorized scope; permission checks are not a general factual-content verifier.

The current protected path uses signed canonical objects and a native host:

Agent/framework: messages, proposed arguments, tool choice (untrusted)
                            |
                            v
Constrained MCP or trusted SDK host: fixed graph/routes + actual principal
                            |
                            v
Rust: signature/scope/currentness -> policy/budget -> exact signed approval
                            |
                            v
Credential-owning broker -> real effect -> independent verification -> receipts

TrustedWarrantIssuer and TrustedDelegationOwner belong to trusted operator setup. Signed warrant content is pinned by canonical bytes/signature; changing fields is a different object and cannot retain the original valid signature. Narrower delegation is a newly signed authority object, not a mutable update in an agent message. Immutable content does not mean permanent authorization: the host still checks current state, scope, validity and policy for each operation.

Keep signing keys, broker credentials, trust configuration and reviewer custody outside agent state. The constrained procedure MCP server has five host-selected tools; it offers no tool for enrolling arbitrary graphs, importing keys or changing host configuration. A caller credential is not a reviewer key. Separate transport identities prevent one caller from resuming or reading another caller's operations in the tested profile.

Use the complete protected Python/TypeScript/MCP examples for the actual configured host, signed approval pause, effect, public-only receipt verification and operation binding checks. Observability middleware alone grants none of these guarantees. A LangChain interrupt/approval checkbox is not a native signed approval and cannot substitute for it.

Isolation and learned execution

The embedded SDK trusts the application and OS owner. An arbitrary callback with filesystem access or a provider credential retains that access. A separate MCP host and narrow tools keep broker/reviewer secrets outside model context, but same-user processes are not automatically an OS sandbox. The bounded Wasmi guest has its own narrower no-WASI boundary. Complete arbitrary-agent OS isolation and named model-driven protected-client acceptance remain separate original gates.

Learned procedures are bounded typed graphs, not copied chat history. An application supplies a separate trusted verifier. Three distinct training inputs, one held-out input and a canary must pass before qualification. Changing pinned assumptions/verifier invalidates reuse. READ/EFFECT nodes still cross the current broker, and effects/approvals are never cached. The native scheduler/operators are deterministic under frozen inputs and state; an external model's generated text is not made deterministic by wrapping it. See the runnable learning program and implementation guide.

On refusal, stop or explicitly replan within the existing scope. Cancellation is not rollback, and an uncertain external write is not permission to retry. Task16 recovery and the complete production/security evaluation remain open; the pilot's documented verified components are not certification.

Complete local-model agent programs

These are the full executable sources. Their adjacent learned calculation module uses existing native procedure memory, not arbitrary code evaluation or chat history. The tool accepts only numeric inputs; trusted application code selects the graph and independent verifier before the agent runs.

Python agent

"""Real optional LangChain/Ollama agent. Public sample; ordinary observations only."""
import json
import os
from pathlib import Path
import tempfile
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from soterias import Soterias
from soterias.integrations.langchain import SoteriasMiddleware
from learned_total import learn_total, calculate

tool_calls = []
directory = Path(tempfile.mkdtemp(prefix="soterias-langchain-"))


@tool
def invoice_total(quantity: int, unit_price: int) -> str:
    """Multiply a quantity by a unit price in integer minor currency units."""
    result, report = calculate(qualified, environment, quantity, unit_price)
    tool_calls.append({"quantity": quantity, "unit_price": unit_price, "total": int(result),
                       "learned_execution": report})
    return result


with Soterias(workflow="invoice-assistant", directory=directory / "metrics") as sdk:
    # Include actual qualification/setup time as its own native observation.
    qualified, environment = sdk.run(learn_total, directory / "memory")
    agent = create_agent(
        model=ChatOllama(model=os.environ.get("OLLAMA_MODEL", "qwen3.5:4b"),
                         base_url=os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434"),
                         temperature=0, seed=7, reasoning=False, num_predict=384),
        tools=[invoice_total],
        middleware=[SoteriasMiddleware(sdk, provider="ollama")],
        system_prompt="Use invoice_total for the requested calculation. Report the actual tool result.",
    )
    response = agent.invoke({"messages": [{"role": "user", "content":
        "Use invoice_total for quantity 6 and unit_price 1200. Then state the result."}]},
        config={"recursion_limit": 6})
    assert len(tool_calls) == 1
    assert {key: tool_calls[0][key] for key in ("quantity", "unit_price", "total")} == {
        "quantity": 6, "unit_price": 1200, "total": 7200}
    assert "7200" in str(response["messages"][-1].content)
    print(json.dumps({"actual_tool_calls": tool_calls, "metrics": sdk.summary(),
                      "security_mode": "ordinary-observation", "output_quality": "unscored",
                      "metrics_directory": str(directory)}))

Python learned calculation

"""An explicit, independently verified pure graph. No authority or auto-mining."""
from soterias import Procedure, ProcedureMemory


def learn_total(directory):
    # Ordinary graph-assumption identifiers, not signed policy or credentials.
    environment = dict(policy="1" * 64, tools="2" * 64, credentials="3" * 64,
                       evidence="4" * 64, host_configuration=None)
    def slot(name):
        return {"op": "input", "name": name, "pointer": ""}
    calculation = {"op": "multiply", "left": slot("quantity"), "right": slot("price")}
    graph = {"version": 1, "id": "invoice-total", "environment": environment,
             "inputs": {"quantity": "integer", "price": "integer"}, "terminal": "result",
             "nodes": [{"id": name, "after": [], "output": "integer",
                        "step": {"kind": "pure", "value": calculation}}
                       for name in ("calculate", "duplicate")] + [
                           {"id": "result", "after": ["calculate", "duplicate"],
                            "output": "integer", "step": {"kind": "verify",
                                "condition": {"op": "literal", "value": True},
                                "value": {"op": "node", "id": "calculate", "pointer": ""}}}]}
    # Selected by trusted application code, separately from the candidate graph.
    verifier = {"op": "equal", "left": {"op": "input", "name": "context", "pointer": "/result"},
                "right": {"op": "multiply",
                          "left": {"op": "input", "name": "context", "pointer": "/inputs/quantity"},
                          "right": {"op": "input", "name": "context", "pointer": "/inputs/price"}}}
    procedure = Procedure(graph, verifier=verifier)
    memory = ProcedureMemory.create(directory)
    for index in range(5):
        run = procedure.start({"quantity": str(index + 1), "price": "1200"}, environment=environment)
        try:
            assert run.advance_pure(environment=environment) == "verified"
            memory.observe(run, phase="training" if index < 3 else "shadow" if index == 3 else "canary")
        finally:
            run.close()
    qualified = ProcedureMemory.open(directory).qualify(procedure, environment=environment)
    return qualified, environment


def calculate(qualified, environment, quantity, price):
    run = qualified.start({"quantity": str(quantity), "price": str(price)}, environment=environment)
    try:
        assert run.advance_pure(environment=environment) == "verified"
        assert run.report["independently_verified"] is True
        assert sum(item["reused_pure"] for item in run.report["trace"]) == 1
        return run.result, run.report
    finally:
        run.close()

TypeScript agent

// Real optional LangChain/Ollama agent. Synthetic data; no broker/security claim.
import assert from 'node:assert/strict';
import fs from 'node:fs';import os from 'node:os';import path from 'node:path';
import {createAgent,tool} from 'langchain';
import {ChatOllama} from '@langchain/ollama';
import * as z from 'zod';
import {Soterias} from '@soterias/core';
import {soteriasMiddleware} from '@soterias/core/integrations/langchain';
import {learnTotal,calculate} from './learned_total.mjs';

const directory=fs.mkdtempSync(path.join(fs.realpathSync(os.tmpdir()),'soterias-langchain-'));
const calls:{quantity:number;unitPrice:number;total:number;learnedExecution:unknown}[]=[];
const sdk=new Soterias({workflow:'invoice-assistant',directory:path.join(directory,'metrics')});
try {
  const {qualified,environment}=sdk.run(learnTotal,path.join(directory,'memory'));
  const invoiceTotal=tool(({quantity,unitPrice})=>{
    const {result,report}=calculate(qualified,environment,quantity,unitPrice);
    calls.push({quantity,unitPrice,total:Number(result),learnedExecution:report});return String(result);
  },{name:'invoice_total',description:'Multiply quantity by an integer unit price in minor currency units.',
    schema:z.object({quantity:z.number().int().min(0).max(1000),unitPrice:z.number().int().min(0).max(1000000)})});
  const agent=createAgent({
    model:new ChatOllama({model:process.env.OLLAMA_MODEL??'qwen3.5:4b',
      baseUrl:process.env.OLLAMA_BASE_URL??'http://127.0.0.1:11434',
      temperature:0,seed:7,think:false,numPredict:384,checkOrPullModel:false}),
    tools:[invoiceTotal],middleware:[soteriasMiddleware(sdk,{provider:'ollama'})],
    systemPrompt:'Use invoice_total for the requested calculation. Report the actual tool result.',
  });
  const response=await agent.invoke({messages:[{role:'user',content:'Use invoice_total for quantity 6 and unitPrice 1200. Then state the result.'}]},{recursionLimit:6});
  assert.equal(calls.length,1);assert.equal(calls[0]!.quantity,6);assert.equal(calls[0]!.unitPrice,1200);assert.equal(calls[0]!.total,7200);
  assert(String(response.messages.at(-1)!.content).includes('7200'));
  console.log(JSON.stringify({actual_tool_calls:calls,metrics:await sdk.summaryAsync(),
    security_mode:'ordinary-observation',output_quality:'unscored',metrics_directory:directory}));
} finally {await sdk.closeAsync();}

TypeScript learned calculation

// Explicit qualified graph DATA; no model-context authority or automatic mining.
import assert from 'node:assert/strict';
import {Procedure,ProcedureMemory,type ProcedureEnvironment,type ProcedureExpr,type ProcedureGraph,type QualifiedProcedure} from '@soterias/core';

export function learnTotal(directory:string){
  const environment:ProcedureEnvironment={policy:'1'.repeat(64),tools:'2'.repeat(64),credentials:'3'.repeat(64),evidence:'4'.repeat(64),host_configuration:null};
  const slot=(name:string):ProcedureExpr=>({op:'input',name,pointer:''});
  const calculation:ProcedureExpr={op:'multiply',left:slot('quantity'),right:slot('price')};
  const graph:ProcedureGraph={version:1,id:'invoice-total',environment,inputs:{quantity:'integer',price:'integer'},terminal:'result',nodes:[
    ...['calculate','duplicate'].map(id=>({id,after:[],output:'integer' as const,step:{kind:'pure' as const,value:calculation}})),
    {id:'result',after:['calculate','duplicate'],output:'integer',step:{kind:'verify',condition:{op:'literal',value:true},value:{op:'node',id:'calculate',pointer:''}}},
  ]};
  const verifier:ProcedureExpr={op:'equal',left:{op:'input',name:'context',pointer:'/result'},right:{op:'multiply',left:{op:'input',name:'context',pointer:'/inputs/quantity'},right:{op:'input',name:'context',pointer:'/inputs/price'}}};
  const procedure=new Procedure(graph,{verifier});const memory=ProcedureMemory.create(directory);
  for(let index=0;index<5;index++){
    const run=procedure.start({quantity:String(index+1),price:'1200'},{environment});
    try{assert.equal(run.advancePure({environment}),'verified');memory.observe(run,{phase:index<3?'training':index===3?'shadow':'canary'});}finally{run.close();}
  }
  return{qualified:ProcedureMemory.open(directory).qualify(procedure,{environment}),environment};
}

export function calculate(qualified:QualifiedProcedure,environment:ProcedureEnvironment,quantity:number,price:number){
  const run=qualified.start({quantity:String(quantity),price:String(price)},{environment});
  try{
    assert.equal(run.advancePure({environment}),'verified');assert.equal(run.report.independently_verified,true);
    assert.equal(run.report.trace.filter(item=>item.reused_pure).length,1);
    return{result:run.result,report:run.report};
  }finally{run.close();}
}

Qualification/setup is recorded as its own native observation. The graph report shows native time and one reused duplicate pure expression. The example still invokes the model normally: no avoided model calls or token savings are claimed. Middleware does not automatically extract a trusted graph from arbitrary code or a conversation.