O cliente

stalwart_agents.client é o cliente da API Stalwart: escolha a ferramenta certa para cada turno de uma conversa, verifique cada chamada de ferramenta que o seu agente faz e verifique guardrails, grafos de agentes e random forests.

pip install stalwart-agents

Python 3.10 a 3.14. Defina QLS_API_KEY, e o cliente a lê. O cliente precisa da chave e de uma licença ativada (stalwart licence activate, uma vez em cada máquina) para tudo, exceto verificar um certificado (a licença e o runtime).

Planos e verificações de ferramentas leem os turnos dos seus usuários e as chamadas dos seus agentes, por isso são respondidos no seu próprio host: pelo runtime licenciado que o wheel traz (stalwart licence activate, uma vez) ou pelo qls-router enterprise que você executa no seu próprio ambiente (router_url=, ou QLS_ROUTER_URL). A API hospedada da Stalwart não responde a nenhum dos dois. Sem nenhum deles configurado, um plano ou uma verificação levanta SelectorUnavailableError, que diz qual configurar.

Roteando uma conversa

Registre as suas ferramentas uma vez e, depois, leia cada turno de uma conversa em um plano. Nenhum modelo é consultado: o plano nomeia a ferramenta, ou a pergunta a fazer ao usuário, e a regra que decidiu.

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")
print(plan.outcome)                    # "run", "ask", "declined" or "unread"
for step in plan.steps:
    print(step.kind, step.selection.tool, step.selection.missing)

A sessão carrega o contexto da conversa -- a pergunta ainda em aberto, os valores digitados, o que uma chamada devolveu -- de um turno para o seguinte. Ela não interpreta nada disso: o próximo contexto volta com cada plano e é enviado com o próximo turno.

plan = chat.read("show me the stock in depot HB-7")
if plan.outcome == "ask" and plan.chain:
    # The step can be completed by listing first: make those calls, and report them.
    results = [{"tool": c["tool"], "arguments": c["arguments"], "result": my_tools(c)} for c in plan.chain]
    plan = chat.answered(plan, results)
stored = chat.to_json()                # for a stateless backend; session_from_json() restores it

Um plano não tem valor de verdade: if plan: levanta um erro, porque "funcionou?" significa quatro coisas aqui. Um plano recusado nunca é entregue a um modelo, diga o on_unread o que disser. Um turno que precisa de um cálculo que nenhuma ferramenta responde é executado na imagem de contêiner qls-sandbox no seu próprio host, quando há um mecanismo de contêineres e a imagem (client.compute); quando não há, o plano é recusado, com um motivo que começa com compute off:.

O que é enviado aos servidores da Stalwart

Uma resolução e as suas entradas (uploads incluídos), um prompt de geração de guardrails, um catálogo, uma planilha de regras e o modelo do minerador de vocabulário são analisados nesta máquina antes de qualquer envio, em busca de endereços de e-mail, números de telefone, números de cartão de pagamento, IBANs, números do National Insurance do Reino Unido, números de CPF e os identificadores que você lista como seus:

import re

from stalwart_agents.client import PersonalDataError

guarded = Client(own_identifiers=["Acme Holdings", re.compile(r"ACME-\d{6}")])
try:
    guarded.register_catalogue({"tools": [{"name": "get_order", "description": "Ask jo@acme.co.uk"}]})
except PersonalDataError as refused:
    print(refused.findings[0].kind, refused.findings[0].path)   # email document.tools[0].description

Uma ocorrência não envia nada. O erro nomeia o tipo de cada ocorrência e onde ela está, nunca o seu valor. Um endereço em example.com, .example, .invalid ou .test é um exemplo, e não o de uma pessoa.

Verificando uma chamada de ferramenta

check = chat.check_tool_call("send_payment", {"payee": "ACME", "amount": 120})
print(check.decision, check.reasons)   # "allow", "confirm" or "refuse"
chat.ran(check, outcome="ran")         # what became of the call, for the next check

Verificando um guardrail

result = client.verify_guardrail("policy.co", "execute_transfer_funds", "is_authenticated")
print(result.status)                   # "UNSAT", "SAT", "STALL", or None for a void verdict
if result.certificate:
    result.certificate.save("certificate.json")   # verified against its digest as it is written

Um veredito SAT não levanta nada: ele é a resposta à pergunta, com o reparo em result.corrective_action e result.patched_policy.

Verificando um certificado você mesmo

Um certificado pode ser verificado sem chave, sem rede e sem nenhum software nosso.

1. O verificador independente: um único arquivo, só com a biblioteca padrão, sob licença MIT: https://api.stalwart.it/downloads/qls-check.py, com o seu digest em https://api.stalwart.it/downloads/qls-check.py.sha256. O formato que ele verifica está especificado em https://api.stalwart.it/certificate-spec/.
2. O mesmo arquivo, carregado neste pacote, por conveniência:

from stalwart_agents.client import check_certificate

outcome = check_certificate(certificate_path)
print(outcome.outcome, outcome.reason)  # "verified", "not_self_contained" or "failed"

ou stalwart check certificate.json. stalwart check --print-source escreve o arquivo carregado, para compará-lo com o publicado. Uma verificação estabelece que uma derivação é válida, nunca que uma política é segura.

Erros

Novas tentativas

Leituras, planos e verificações de ferramentas são repetidos em uma conexão interrompida, um tempo esgotado, um 5xx e um 429 que se resolve sozinho (respeitando o Retry-After). Uma resolução é enviada sob uma chave de idempotência que o cliente deriva, de modo que um envio repetido é um único job e um segundo envio deliberado são dois. O registro de um catálogo, um registro de feedback e uma tentativa sobre um achado são repetidos sob uma chave por chamada, de modo que uma nova tentativa registra uma vez e uma segunda chamada registra de novo. Os efeitos de um cancelamento nunca são repetidos.

O que o cliente não faz

Ele verifica a forma do que você envia -- uma chave de configuração que o serviço não tem, um perfil que um modo recusa -- e nada mais (o que o cliente não valida). Se uma solicitação pode responder à sua pergunta é o serviço que diz, uma vez. Ele não carrega código de seletor: um turno é lido pelo runtime local licenciado que o wheel do stalwart-agents traz, ou pelo seu qls-router, nunca por este módulo.

Compatibilidade

Licença

© Stalwart Technologies Ltd. Proprietário; veja LICENSE. O verificador carregado em stalwart_agents/client/_vendor/ é MIT, com o seu próprio LICENSE. Suporte: crm@stalwart.it.