Worked examples

Each section shows one thing the product does, as you would do it. The agent examples assume the agent.yaml and tools of Build an agent; the verification examples assume a Client() with QLS_API_KEY set. Every example needs your key and an activated licence (the licence and the runtime).

A turn read with no model call

Your agent reads each turn against the words each tool declares -- its name, its description, its arguments -- and picks the tool itself. No model is asked to choose, the same words get the same tool, and every step names the rule that produced it:

from stalwart_agents.client import Client

client = Client()
version = client.register_catalogue(mcp_tools_list, agent="desk")
chat = client.session("desk", catalogue=version)
plan = chat.read("show me the stock in depot HB-7")
for step in plan.steps:
    print(step.kind, step.selection.tool, step.decided_by)   # tool ListStock head-parts

An Agent does the same on every turn before anything else runs. The plan is made on your own machine, by the licensed runtime, or by the qls-router you run in your environment.

Calculations no tool covers, handed to the model

A turn no tool covers -- a calculation, say -- goes to your model only where agent.yaml says so. Here it does:

agents:
  desk:
    model: "anthropic:claude-sonnet-5-5"   # your provider and key
    tools: [get_order, refund_card]
    on_unread: model                       # a turn no tool covers, a calculation say, goes to the model

from stalwart_agents import Agent

desk = Agent.from_file("agent.yaml", name="desk")
reply = desk.chat("what is 17.5% VAT on 240 pounds?", user="u-17")
print(reply.text)            # the model's answer
print(reply.rated)           # the turn's record: the verdict `unread`, so the model answered it

on_unread: ask asks the person first, and the model has the turn only once they say yes (desk.confirm("u-17", "approve"), or "decline").

Where a plan can run the calculation itself, it does, on your machine: compute steps run in the qls-sandbox container image on your own host, when a container engine and the image are present. Every plan includes some -- 100 runs and 300 seconds a month on the free plan, 10,000 runs and 30,000 seconds on the paid plan, no limit on enterprise. Where the image is not there, a plan that needs a calculation says compute off: in its reason, and on_unread decides what happens to the turn.

No model call at all

Turn the model off and it is never called, for anything:

agents:
  desk:
    tools: [get_order, refund_card]
    on_unread: refuse                      # a turn no tool covers is refused, by name
replies:
  phrasings: phrasings.yaml
  composer: template                       # replies are your phrasings; no model writes them
answers:
  index: sources.json
  composer: extract                        # answers are your documents' own sentences

With no model, on_unread: refuse, no hands_off_to, replies from your phrasings and answers in your documents' own sentences, there is nothing left for a model to do. A turn no tool covers is refused with the reason, and the person is asked to say it another way once first.

Every tool call checked against your rules

Every call is checked before it runs, whoever chose it -- the reader or the model -- against the catalogue, the argument schema, the session's history and your rules. The answer is one of three: allow, confirm (a person approves the exact arguments shown) or refuse, and no model or network is involved in it.

# rules.yaml
rules:
  - action: issue_refund
    requires: [supervisor_approved]
    when: { amount_over: 200 }
  - action: close_account
    requires_confirmation: true
  - action: delete_customer
    never: true
  - action: send_statement
    after: [verify_identity]

$ stalwart rules compile rules.yaml --save

check = chat.check_tool_call("issue_refund", {"order_id": "4411", "amount": 450})
print(check.decision, check.reasons)   # "refuse", and the rule that decided

A refused call runs nothing, and the agent's reply says why. Your existing Colang 1.0 or 2.x policy can be the rulebook instead (rulebook: ./policy.co).

One catalogue from the tools you have

Your tools can come from Python functions, MCP servers, OpenAPI documents, LangChain tools and a TypeScript project, into one versioned catalogue:

$ stalwart catalogue import openapi.yaml
$ stalwart catalogue import ./tools-ts            # exported functions tagged @read or @write

Each registration is a version, linted before it routes: a write with no undo, two tools the reader cannot tell apart, a name the reader takes for grammar. The tools' own code -- a TypeScript project's sources on import, and your Python, JavaScript, TypeScript and Go under stalwart verify -- is first put through the code checks for network-security and credential problems, on your machine: a finding fails, nothing can suppress it, and only the counts are sent to Stalwart. The version before can always be restored (client.restore_catalogue(version)).

A team of agents verified before it runs

Agents hand off to each other with hands_off_to. Before the first turn, and in CI with stalwart verify, every route through the team is searched for a way to reach a risky action without its safeguard:

agents:
  desk:
    tools: [get_order, read_inbox]
    hands_off_to: [billing]
  billing:
    privilege: high
    tools: [issue_refund]

$ stalwart verify agent.yaml --keep findings/
$ stalwart verify agent.yaml --replay      # each finding's witness replayed on the agents, every tool a stub

A privileged tool reachable after untrusted input -- read_inbox here, then a hand-off to billing -- is a finding, with the path that reaches it and the line of agent.yaml it comes from. A dangerous path stops the agent: Agent(...) raises NotVerified rather than run it.

Validating a LangGraph graph

A graph you built with LangGraph is checked the same way. Export the compiled graph with the tool catalogue it binds, and send both to the graph mode:

import json
from stalwart_agents.client import Client
from stalwart_agents.langchain.graph import export_graph

app = builder.compile()                    # your LangGraph StateGraph
with open("graph.json", "w") as f:
    json.dump(export_graph(app, catalogue, model_nodes=["agent"]), f)
result = Client().verify_graph("graph.json", catalogue="catalogue.json")
for finding in result.findings:
    print(finding.severity, finding.rule, finding.location)

Each tool the graph can reach is held to its check, its confirmation, its order rules and what an undoable write needs, over the session depth you set (session_steps=). A finding comes with the exact path, in SARIF too, and replay_witness replays its witness on your graph with every tool stubbed. Nodes that are not Stalwart's own are read as opaque, and a clean result holds up to the depth checked.

Controlled rollbacks

Declare how each write is undone. When a run fails, its writes are undone last first, each undo checked against your rules like any call; a run waiting at a confirmation is cancelled there:

from decimal import Decimal
from stalwart_agents import Agent, tool

@tool(writes=True, undo="refund_card")     # how this write is undone
def charge_card(order_id: str, amount: Decimal) -> dict: ...

@tool(writes=True)
def refund_card(order_id: str, amount: Decimal) -> dict: ...

@tool(writes=True)
def provision_account(order_id: str) -> dict: ...

desk = Agent("desk", tools=[charge_card, refund_card, provision_account], rulebook="policy.co")
reply = desk.chat("charge order 4411 and open the account", user="u-17")
# provision_account fails -> the charge is refunded, last write first

runs: atomic                               # the default; `stop` leaves a failed run as it is

Under stalwart serve, an operator rolls a thread's run back the same way. The route needs the operator token you started the server with (--operator-token, or QLS_OPERATOR_TOKEN):

$ curl -X POST -H "Authorization: Bearer $QLS_OPERATOR_TOKEN" http://127.0.0.1:8700/threads/t-17/rollback

With no token set, the route answers 403 and says how to set one, so nobody who merely reaches the server can undo a run. In LangGraph, stalwart_rollback_node does the same, and ResumeServer(operator_token=...) guards its route the same way.

Receipts behind Kong

Point the agent's MCP source at your Kong route; the agent's catalogue is that server's tools/list. With receipts: true, each call carries the agent's own check -- the session's history, ending on the check that allowed it -- and Kong's route admits it by asking your qls-router's POST /v1/tool-receipts, so a write the person approved in the agent is not asked about twice:

tools:
  - mcp: https://kong.internal/mcp         # your MCP servers, behind Kong
    receipts: true                         # each call carries the agent's own check

A receipt admits one call, once: presented again for other arguments, another tool or a call already reported, it is a 403. A call without one is checked in full at your router's POST /v1/tool-checks. The Kong recipe has the route, the plugin settings and the network policy that make Kong the only way to the tools.

Replies true to what was done

replies:
  rules: reply-rules.yaml        # personal data by class, phrases never to say, allowed offers
  phrasings: phrasings.yaml      # how each write's outcome is told: done, failed, refused, ...
  composer: model                # or template: the reply is the phrasings, and no model writes it

Each sentence that speaks of a write is held to that write's outcome: done only if it ran, a failed or refused one never told as done, and every amount, date or identifier in it to a value the call was sent or returned. A sentence that misstates is replaced by the outcome's phrasing. Card numbers, IBANs, national identifiers, email addresses and telephone numbers the rules withhold are masked or the reply held back, banned phrases are removed and required disclaimers added. It runs in the local runtime, and only counts leave the machine.

Answers from your documents

$ stalwart sources build docs/*.md --out sources.json     # the index, made here, from local files
$ stalwart sources ask --index sources.json "how long does a refund take?"

answers:
  index: sources.json
  composer: extract        # the documents' own sentences; compose: your model writes from them
agents:
  desk:
    on_unread: answer

The documents and their index never leave your host. Each sentence of an answer cites its passage, and a composed sentence that states a number, date, amount, name or negation its passages do not is removed. compose needs a plan that includes composed answers.

Learning your users' words

reply = desk.chat("what is in order 4411?", user="u-17")
desk.thumbs(reply, up=False)        # one record: the turn's words, the tool, the reader's verdict and rule

Thumbs up and down are feedback records the hosted miner learns your users' words from. A new vocabulary is published only when it makes nothing worse on five checks, every version can be rolled back (client.revert_vocabulary()), and your feedback is used for your tenant alone, never to train models.

A certificate for every rule

Ask for a row per rule of your policy, give each row you keep an id, an expect and a gate, and run the table: under the patch profile, every row gets a certificate of its own.

import json
from pathlib import Path

from stalwart_agents.client import Client

client = Client()
proposed = client.enumerate_properties("policy.co")          # step 1: one candidate row per rule
rows = [{**row, "id": f"RULE-{n:02}", "expect": "UNSAT", "gate": "hard"}
        for n, row in enumerate(proposed.records, 1)]
Path("rules.json").write_text(json.dumps({"policy": "policy.co", "properties": rows}))

result = client.verify_batch("rules.json", "policy.co", profile="patch")   # step 2: run the table
for record in result.records:
    print(record["id"], record["status"], record["certificate_artifact_id"])
print(result.all_passed, result.hard_failures)

The profiles: fast, a verdict only, in about a quarter of a second, with no certificate and no repair; patch, a repair with a certificate you can replay; and proof. A certificate is checked with no key, no network and none of our software (checking a certificate yourself).

Every draft verified

Describe a rule in plain language and name your own model. Each draft is compiled and checked against the action and context you name, and a draft that fails is sent back to your model with the solver's repair, for up to three rounds where your plan allows:

result = client.generate_and_verify(
    "Funds may move only for a signed-in user whose account is not frozen.",
    "transfer_funds", "is_authenticated",
    provider="anthropic", model="claude-sonnet-5-5", api_key_env="ANTHROPIC_API_KEY", rounds=3,
)
print(result.status)                     # the verdict on the draft that ships
print(result.shipping_candidate)         # the draft, verified

Your model's key travels with this one request: the SDK neither stores nor logs it, and the service keeps neither the key, the prompt nor the model's raw answer.

Hunting a policy for bugs

result = client.hunt("policy.co", gate=True)
for finding in result.findings:
    print(finding.severity, finding.rule, finding.reproduction)
    print(finding.fix)                    # a corrective guard, checked by scanning again

Every finding has a severity, a reproduction -- the conversation that triggers it, replayed -- and a fix verified by scanning again. gate=True runs the quality gate, which can block a release, and the findings come as SARIF too. Scans grouped into projects, each compared with the last, and triage of each finding are on the console and the API (projects and scans).

Checking a random forest

result = client.verify_forest("model.onnx", "properties.json")
for record in result.records:
    print(record)                        # each property: holds, a counterexample re-checked on the model, or undecided
counter = client.enumerate_counterexamples("model.onnx", "search.json", anchors="records.json")

A forest is read as ONNX-ML or Treelite JSON, and a scikit-learn model through an ONNX export. Each violation comes with a counterexample checked again on the model itself; the search around real records is exhaustive, not sampled. A wide fairness or monotonicity region may stall: a stall concludes nothing, and a narrower region gets an answer.