Exemplos completos
Cada seção mostra uma coisa que o produto faz, do jeito que você a faria. Os exemplos de agente
partem do agent.yaml e das ferramentas de Construa um agente;
os exemplos de verificação partem de um Client() com QLS_API_KEY definida. Todo exemplo
precisa da sua chave e de uma licença ativada (a licença e o runtime).
Um turno lido sem chamada de modelo
O seu agente lê cada turno pelas palavras que cada ferramenta declara -- o seu nome, a sua descrição, os seus argumentos -- e escolhe a ferramenta sozinho. Nenhum modelo é consultado para escolher, as mesmas palavras levam à mesma ferramenta, e cada passo nomeia a regra que o produziu:
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
Um Agent faz o mesmo em todo turno, antes de qualquer outra coisa rodar. O plano é feito na sua
própria máquina, pelo runtime licenciado, ou pelo qls-router que você executa no seu ambiente.
Cálculos que nenhuma ferramenta cobre, entregues ao modelo
Um turno que nenhuma ferramenta cobre -- um cálculo, por exemplo -- só vai para o seu modelo onde
o agent.yaml diz. Aqui ele diz:
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 pergunta primeiro à pessoa, e o modelo só recebe o turno depois que ela diz sim
(desk.confirm("u-17", "approve"), ou "decline").
Onde um plano pode executar o cálculo por conta própria, ele o executa, na sua máquina: os passos
de cálculo rodam na imagem de contêiner qls-sandbox no seu próprio host, quando um mecanismo de
contêineres e a imagem estão presentes. Todo plano inclui alguns -- 100 execuções e 300 segundos
por mês no plano gratuito, 10.000 execuções e 30.000 segundos no plano pago, sem limite no
enterprise. Onde a imagem não está, um plano que precisa de um cálculo traz compute off: no seu
motivo, e on_unread decide o que acontece com o turno.
Nenhuma chamada de modelo
Desligue o modelo e ele nunca é chamado, para nada:
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
Sem model, com on_unread: refuse, sem hands_off_to, com respostas a partir das suas frases
e respostas nas próprias frases dos seus documentos, não sobra nada para um modelo fazer. Um turno
que nenhuma ferramenta cobre é recusado com o motivo, e antes a pessoa é convidada, uma vez, a
dizê-lo de outra forma.
Toda chamada de ferramenta verificada pelas suas regras
Toda chamada é verificada antes de rodar, quem quer que a tenha escolhido -- o leitor ou o
modelo --, contra o catálogo, o esquema de argumentos, o histórico da sessão e as suas regras. A
resposta é uma de três: allow, confirm (uma pessoa aprova exatamente os argumentos mostrados)
ou refuse, e nenhum modelo nem rede participa dela.
# 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
Uma chamada recusada não executa nada, e a resposta do agente diz por quê. A sua política Colang
1.0 ou 2.x existente pode ser o rulebook no lugar da planilha (rulebook: ./policy.co).
Um catálogo a partir das ferramentas que você tem
As suas ferramentas podem vir de funções Python, servidores MCP, documentos OpenAPI, ferramentas LangChain e um projeto TypeScript, para um único catálogo versionado:
$ stalwart catalogue import openapi.yaml
$ stalwart catalogue import ./tools-ts # exported functions tagged @read or @write
Cada registro é uma versão, passada por lints antes de rotear: uma escrita sem reversão, duas
ferramentas que o leitor não consegue distinguir, um nome que o leitor toma por gramática. O
próprio código das ferramentas -- as fontes de um projeto TypeScript na importação, e o seu
Python, JavaScript, TypeScript e Go no stalwart verify -- passa antes pelas verificações de
código em busca de problemas de segurança de rede e de credenciais, na sua máquina: um achado
reprova, nada pode suprimi-lo, e só as contagens são enviadas à Stalwart. A versão anterior
sempre pode ser restaurada (client.restore_catalogue(version)).
Uma equipe de agentes verificada antes de rodar
Os agentes transferem a conversa entre si com hands_off_to. Antes do primeiro turno, e na CI com
stalwart verify, cada caminho pela equipe é examinado em busca de um jeito de chegar a uma ação
arriscada sem a sua salvaguarda:
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
Uma ferramenta privilegiada alcançável depois de uma entrada não confiável -- aqui read_inbox,
depois uma transferência para billing -- é um achado, com o caminho que a alcança e a linha do
agent.yaml de onde ela vem. Um caminho perigoso impede o agente: Agent(...) levanta
NotVerified em vez de executá-lo.
Validar um grafo LangGraph
Um grafo que você construiu com LangGraph é verificado da mesma forma. Exporte o grafo compilado
com o catálogo de ferramentas que ele vincula e envie os dois ao modo graph:
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)
Cada ferramenta que o grafo alcança é confrontada com a sua verificação, a sua confirmação, as
suas regras de ordem e o que uma escrita reversível exige, até a profundidade de sessão que você
definir (session_steps=). Um achado vem com o caminho exato, também em SARIF, e replay_witness
reproduz a sua testemunha no seu grafo com todas as ferramentas simuladas. Os nós que não são da
Stalwart são lidos como opacos, e um resultado limpo vale até a profundidade verificada.
Reversões controladas
Declare como cada escrita é desfeita. Quando uma execução falha, as suas escritas são desfeitas da última para a primeira, cada reversão verificada pelas suas regras como qualquer chamada; uma execução que espera uma confirmação é cancelada ali:
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
Com stalwart serve, um operador reverte a execução de uma conversa da mesma forma. A rota exige
o token de operador com que você iniciou o servidor (--operator-token, ou QLS_OPERATOR_TOKEN):
$ curl -X POST -H "Authorization: Bearer $QLS_OPERATOR_TOKEN" http://127.0.0.1:8700/threads/t-17/rollback
Sem token definido, a rota responde 403 e diz como definir um, de modo que ninguém que apenas
alcance o servidor consegue desfazer uma execução. No LangGraph, stalwart_rollback_node faz o
mesmo, e ResumeServer(operator_token=...) protege a sua rota da mesma forma.
Recibos atrás do Kong
Aponte a fonte MCP do agente para a sua rota do Kong; o catálogo do agente é o tools/list desse
servidor. Com receipts: true, cada chamada leva a verificação do próprio agente -- o histórico
da sessão, terminando na verificação que a permitiu --, e a rota do Kong a admite perguntando ao
POST /v1/tool-receipts do seu qls-router, de modo que uma escrita que a pessoa aprovou no
agente não é perguntada duas vezes:
tools:
- mcp: https://kong.internal/mcp # your MCP servers, behind Kong
receipts: true # each call carries the agent's own check
Um recibo admite uma chamada, uma vez: apresentado de novo para outros argumentos, outra
ferramenta ou uma chamada já relatada, ele é um 403. Uma chamada sem recibo é verificada por
completo no POST /v1/tool-checks do seu roteador. A receita do Kong
traz a rota, as configurações do plugin e a política de rede que fazem do Kong o único caminho até
as ferramentas.
Respostas fiéis ao que foi feito
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
Cada frase que fala de uma escrita se atém ao resultado dessa escrita: feita só se ela rodou, uma falhada ou recusada nunca contada como feita, e todo valor, data ou identificador nela a um valor que a chamada enviou ou recebeu. Uma frase que distorce é substituída pela frase do resultado. Números de cartão, IBANs, identificadores nacionais, endereços de e-mail e números de telefone que as regras retêm são mascarados, ou a resposta é retida; frases proibidas são removidas e avisos obrigatórios, acrescentados. Isso roda no runtime local, e só contagens saem da máquina.
Respostas a partir dos seus documentos
$ 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
Os documentos e o seu índice nunca saem do seu host. Cada frase de uma resposta cita a sua
passagem, e uma frase composta que afirma um número, data, valor, nome ou negação que as suas
passagens não afirmam é removida. compose exige um plano que inclua respostas compostas.
Aprender as palavras dos seus usuários
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
Polegares para cima e para baixo são registros de feedback com que o minerador hospedado aprende
as palavras dos seus usuários. Um vocabulário novo só é publicado quando não piora nada em cinco
verificações, toda versão pode ser revertida (client.revert_vocabulary()), e o seu feedback é
usado só para o seu tenant, nunca para treinar modelos.
Um certificado para cada regra
Peça uma linha por regra da sua política, dê a cada linha que você mantiver um id, um expect e
um gate, e execute a tabela: no perfil patch, cada linha recebe um certificado próprio.
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)
Os perfis: fast, só o veredito, em cerca de um quarto de segundo, sem certificado e sem reparo;
patch, um reparo com um certificado que você pode reproduzir; e proof. Um certificado é
verificado sem chave, sem rede e sem nenhum software nosso
(verificar um certificado por conta própria).
Todo rascunho verificado
Descreva uma regra em linguagem simples e indique o seu próprio modelo. Cada rascunho é compilado e verificado contra a ação e o contexto que você indicar, e um rascunho que falha volta ao seu modelo com o reparo do solver, por até três rodadas onde o seu plano permitir:
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
A chave do seu modelo viaja só com esta solicitação: o SDK não a armazena nem a registra, e o serviço não guarda a chave, o prompt nem a resposta bruta do modelo.
Caçar bugs em uma política
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
Todo achado tem uma severidade, uma reprodução -- a conversa que o dispara, reproduzida -- e uma
correção verificada por uma nova análise. gate=True executa o portão de qualidade, que pode
bloquear uma versão, e os achados vêm também em SARIF. Análises agrupadas em projetos, cada uma
comparada com a anterior, e a triagem de cada achado ficam no console e na API
(projetos e análises).
Verificar uma 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")
Uma floresta é lida em ONNX-ML ou Treelite JSON, e um modelo scikit-learn por meio de uma exportação ONNX. Cada violação vem com um contraexemplo verificado de novo no próprio modelo; a busca em torno de registros reais é exaustiva, não amostrada. Uma região ampla de equidade ou de monotonicidade pode travar: uma trava não conclui nada, e uma região mais estreita obtém uma resposta.