stalwart_agents.client

stalwart_agents.client: the client for the Stalwart API.

from stalwart_agents.client import Client
client = Client()                                   # QLS_API_KEY
chat = client.session("desk", catalogue=catalogue)
plan = chat.read("show me the stock in depot HB-7")

The selector first -- catalogues, plans and sessions, computations, tool checks, feedback and the hosted vocabulary -- then the guardrail solver, the bug hunter and the random forest verifier. The Python carries no selector code: a turn is read by the licensed local runtime, or by an enterprise qls-router the customer runs, never here and never by Stalwart's hosted API. A plan's compute step runs on the customer's host as well, in the sandbox image (Client.compute). A request bound for Stalwart's servers is scanned for personal data first, and not sent on a match (PersonalDataError).

check_certificate checks a certificate on this machine, offline, with no key; it is importable where nothing but the standard library is installed.

STREAM_DEGRADED = 'stream_degraded'

The event the iterator yields once when it falls back to polling.

class ApiError(stalwart_agents.client.QlsError):

An error the service answered with.

code is the service's own code, status the HTTP status, and body the error body as sent; message is the service's sentence.

ApiError(status: int, body: Mapping[str, Any], message: str | None = None)

status

body

code

message

class Artifact(stalwart_agents.client.artifacts._Fields):

One artefact of a job, fetched through the client that read the job.

Artifact( record: Mapping[str, Any], fetch: Callable[[str], stalwart_agents.Reply])

def bytes(self) -> bytes:

The artefact's bytes, verified against its digest.

def text(self, encoding: str = 'utf-8') -> str:

The artefact's bytes, verified against its digest, decoded as encoding.

def save(self, path: str | os.PathLike[str]) -> pathlib.Path:

Writes the verified bytes to @p path; on a mismatch, nothing is written.

@dataclass(frozen=True)

class ArtifactRef:

An earlier job's artefact, passed as an input by its id.

ArtifactRef(id: str)

id: str

class AsyncArtifact(stalwart_agents.client.artifacts._Fields):

One artefact of a job, fetched through the async client that read the job.

AsyncArtifact( record: Mapping[str, Any], fetch: Callable[[str], Awaitable[stalwart_agents.Reply]])

async def bytes(self) -> bytes:

The artefact's bytes, verified against its digest.

async def text(self, encoding: str = 'utf-8') -> str:

The artefact's bytes, verified against its digest, decoded as encoding.

async def save(self, path: str | os.PathLike[str]) -> pathlib.Path:

Writes the verified bytes to @p path; on a mismatch, nothing is written.

class AsyncClient:

The Stalwart API, over HTTP, with plans and tool checks answered on your own host.

Args: api_key: else QLS_API_KEY; else ConfigurationError naming the variable. base_url: else QLS_BASE_URL; else https://api.stalwart.it: the solvers, the catalogues, the licence and its bundles. timeout: per HTTP request, not per solve. max_retries: for the calls that are safe to repeat (see _core.retry_delay). user_agent_suffix: appended to the user agent, so a customer's tooling is identifiable in the service's logs. runtime: who answers plans and tool checks. auto, the licensed local runtime where it is installed and licensed, and the router otherwise; local, the runtime alone; router, the router alone. router_url: else QLS_ROUTER_URL: the enterprise qls-router you run in your own environment, which answers plans and tool checks over HTTP. With neither a usable runtime nor a router, a plan or a check raises SelectorUnavailableError. own_identifiers: identifiers of your own that must never reach Stalwart's servers, each a string (matched ignoring case) or a compiled pattern. transport: an httpx transport, such as stalwart_agents.client.testing.replay(path).

AsyncClient( *, api_key: str | None = None, base_url: str | None = None, timeout: float = 30.0, max_retries: int = 3, user_agent_suffix: str | None = None, runtime: str = 'auto', router_url: str | None = None, own_identifiers: Sequence[Union[str, re.Pattern[str]]] = (), transport: httpx.AsyncBaseTransport | None = None)

router_url: str | None

The enterprise qls-router this client asks plans and tool checks of, where it names one.

async def aclose(self) -> None:

Close the client's connections. async with AsyncClient() as client: does it for you.

base_url: str

The API this client calls: base_url=, QLS_BASE_URL, or https://api.stalwart.it.

async def licensed(self) -> Mapping[str, Any]:

Opens this machine's lease in the runtime, as Client.licensed does.

Raises: LicenceRequired: no licence is activated here, the service refused it, or it has lapsed.

async def activate_runtime(self) -> Mapping[str, Any]:

Activates this machine's local runtime, once: its installation, the licence, the first lease.

async def release_runtime(self) -> Mapping[str, Any]:

Frees this machine's seat on the licence, so another installation can take it.

def runtime_request(self) -> Mapping[str, Any]:

An air-gapped machine's activation request: its installation id and public key, for a file.

Carry the file to a machine that reaches the service and pass it to offline_lease there.

async def offline_lease(self, request: Mapping[str, Any]) -> Mapping[str, Any]:

For an air-gapped machine's request file: its licence and a lease for the licence's whole term.

def install_offline(self, answer: Mapping[str, Any]) -> Mapping[str, Any]:

On the air-gapped machine: the licence and lease offline_lease answered, kept and opened.

async def runtime_status(self) -> Mapping[str, Any]:

Which side answers plans and checks -- local, router or none -- and the licence's state.

async def plan( self, turn: str, *, agent: str, catalogue: Any, context: Mapping[str, Any] | None = None, unread: str = 'refuse', traceparent: str | None = None) -> TurnPlan:

One turn read into a plan, statelessly. A conversation is session().

A turn that needs a calculation no tool answers is a compute step where this host can run it (compute), and is declined (compute off: ...) where it cannot.

async def compute( self, step: Any, sources: Sequence[Mapping[str, Any]], *, model: Callable[[ComputePrompt], str], runner: Callable[[str, Any, str], stalwart_agents.sandbox.launch.Outcome] | None = None) -> stalwart_agents.compute.Computation:

A calculation no tool answers, run on this host, as Client.compute runs it. The model is called, and the sandbox waited on, off the event loop.

def session( self, agent: str, *, catalogue: Any = None, on_unread: str | None = None, max_rephrase: int = 1, ask_to_rephrase: bool = True, context: Mapping[str, Any] | None = None, session_id: str | None = None) -> AsyncSession:

A conversation: the context the service returns, carried to the next turn.

session_id names the tool check's history chain; a fresh one by default. A test replaying a cassette names it, so the recorded checks are the same requests. Where on_unread is not given, the first turn takes the console's setting, or refuse.

async def check_tool_call( self, tool: str, arguments: Mapping[str, Any], *, agent: str, context: Mapping[str, Any] | None = None, origins: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, turns: Sequence[str] | None = None, history: History | None = None, session_id: str = '', confirmation: Any = None, compensation_of: int | None = None, plan: Mapping[str, Any] | None = None, traceparent: str | None = None) -> ToolCheck:

One tool call checked, statelessly: history is the History the caller carries.

async def register_catalogue( self, tools: Any, *, agent: str | None = None, agents: Sequence[Mapping[str, Any]] | None = None, source: str = 'auto', rulebook_ir: Mapping[str, Any] | None = None, stores: Mapping[str, Any] | None = None, budget: Mapping[str, Any] | None = None, modules: Mapping[str, Any] | None = None, code_check: Mapping[str, Any] | None = None, rulebook: str | None = None) -> stalwart_agents.client._models.CatalogueVersion:

Registers the tools -- an MCP tools/list, LangChain tool schemas or an OpenAPI document -- as a new version.

A tool's declarations the bug hunter reads (argument_kinds, confidential, retrieves, persists, source, server, ...) ride in its own metadata; stores, budget and modules are the catalogue's own. The lint report says where a declaration is missing or names a store that is not declared. A retry of the same registration is one version.

rulebook is an agent's Colang rulebook (1.0 or 2.x) as text: the service compiles it to the record the tool check evaluates, so the rules the agent is proved against are the rules it runs under. rulebook_ir is that record compiled already, which a tenant whose own build emits IR sends instead.

async def compile_rule_sheet(self, sheet: str) -> dict[str, Any]:

A rule sheet compiled to rulebook IR, with its catalogue overlay and its lints against the newest catalogue. Nothing is kept.

async def save_rule_sheet(self, sheet: str) -> dict[str, Any]:

Compiles the sheet, saves it, and registers it on the newest catalogue as a new version. A retry of the same save is one version.

async def rule_sheets(self) -> dict[str, Any]:

The sheet in force, with its text, and the versions saved before it.

async def catalogues(self) -> list[stalwart_agents.client._models.CatalogueVersion]:

Every registered version, newest first.

async def catalogue( self, version: int | None = None) -> stalwart_agents.client._models.CatalogueVersion | None:

One version in full, or the newest; None where none is registered.

async def restore_catalogue(self, version: int) -> stalwart_agents.client._models.CatalogueVersion:

Registers an older version again, as the newest.

async def feedback( self, record: Mapping[str, Any] | Sequence[Mapping[str, Any]]) -> Mapping[str, Any]:

A user's thumb, or several, in the toolkit's record format. A retry is one record.

Screened before it is sent, as Client.feedback is: a record holding personal data raises PersonalDataError, and nothing is sent.

async def hunt_attempt( self, fingerprint: str | None = None, *, outcome: str, kind: str = 'replay', rounds: int = 0, planted_digest: str | None = None, trace_digest: str | None = None, graph_digest: str | None = None, mode: str = '', owasp: Sequence[str] = (), project_id: str | None = None) -> Mapping[str, Any]:

What an attempt on a finding did: digests and the outcome, never a text. A retry is one attempt.

async def vocabulary(self) -> stalwart_agents.client._models.HostedVocabulary:

The tenant's live hosted vocabulary.

async def mine( self, model: Mapping[str, Any], **options: Any) -> stalwart_agents.client._models.MineRun:

Queues the hosted miner over the tenant's feedback; read the run with ledger(run).

async def ledger(self, run_id: str) -> stalwart_agents.client._models.MineRun:

A learning run: its state, what it applied and its ledger.

async def revert_vocabulary(self) -> stalwart_agents.client._models.VocabularyVersion:

Makes the version before the live one live again.

async def account(self) -> stalwart_agents.client._models.Account:

This key's tier, entitlements, limits and this month's use of each meter.

async def version(self) -> stalwart_agents.client._models.Version:

The commit, each image's version and digest, the solver's binary_version and the host this deployment runs.

async def verify_guardrail( self, policy: 'FileLike | None', action: str, context: str, *, rulebook_ir: 'FileLike | None' = None, pins: Any = None, enumerate_contexts: bool = False, whole_policy: bool = False, prohibitions: bool = False, session_steps: int | None = None, frat_proof: bool = False, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

Can action fire without context? An UNSAT verdict says no, with a certificate; SAT says how, with a repair.

A SAT verdict is the answer to the question and raises nothing. enumerate_contexts is the wire's enumerate (a builtin in Python).

async def generate_and_verify( self, prompt: str, action: str, context: str, *, provider: str, model: str, api_key: str | None = None, api_key_env: str | None = None, rounds: int = 1, base_policy: 'FileLike | None' = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Asks your own model for the policy, then verifies it: the draft, its verdict, a repair, the shipping candidate.

The model's key comes from api_key or the environment variable api_key_env names. It travels with this one request, and the SDK neither stores nor logs it.

async def verify_batch( self, properties: 'FileLike', policy: 'FileLike | None' = None, *, rulebook_ir: 'FileLike | None' = None, rails_config: 'FileLike | None' = None, concepts: 'FileLike | None' = None, expectations: 'FileLike | None' = None, declared: 'FileLike | None' = None, whole_policy: bool = False, prohibitions: bool = False, intent_axis: bool = False, exactness: bool = False, sequence: bool = False, session_steps: int | None = None, transfer: bool = False, allow_oracle_refutations: bool = False, frat_proof: bool = False, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

A whole property table: one record per row, with counters derived from the records.

if result: is refused -- read .all_passed or .hard_failures.

async def derive_pins( self, properties: 'FileLike', policy: 'FileLike | None' = None, *, rulebook_ir: 'FileLike | None' = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

The smallest pin sets under which each obligation holds; incomparable sets are all reported.

async def enumerate_properties( self, policy: 'FileLike | None' = None, *, rulebook_ir: 'FileLike | None' = None, prohibitions: bool = False, intent_axis: bool = False, exactness: bool = False, sequence: bool = False, session_steps: int | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Candidate obligations read off the policy: rows with no id, expect or gate, for review.

The expectation table a sweep's rows derive, against a rulebook.

async def replay_certificate( self, certificate: 'FileLike | None' = None, *, artifact_id: str | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Asks the service to replay a certificate. To check one here, offline, use check_certificate().

replay is replayed, failed or out_of_scope; out of scope is a sound refutation of a kind replay does not check, not an error.

async def verify_graph( self, graph: 'FileLike', *, catalogue: 'FileLike | None' = None, rulebook: 'FileLike | None' = None, session_steps: int | None = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

An exported agent graph, verified before it is deployed: findings in the hunter's form.

The guardrail bug hunter: every way through to a forbidden action, each a finding with its fix.

dependencies is the file that pins what you install -- requirements.txt, pyproject.toml, uv.lock or poetry.lock -- so a library rail a rails config names is checked against an advisory's range at the version you run. inputs carries any other input role the hunt takes (rails_config, intents, vocabulary, probes, corpus, ...), by its role name.

async def verify_forest( self, model: 'FileLike', properties: 'FileLike', *, anchors: 'FileLike | None' = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

A stall on this side concludes nothing: it is not evidence that a property holds.

Checks a trained forest (an ONNX-ML file or a Treelite JSON dump, sent as it is) against robustness, monotonicity, fairness and reachability properties.

async def enumerate_counterexamples( self, model: 'FileLike', search: 'FileLike', *, anchors: 'FileLike', profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Counterexamples searched around real records: proposals with no expect or gate, verdicts only.

async def submit( self, request: Mapping[str, Any]) -> AsyncJob:

Any solve request, as the contract states it; inputs already described.

async def job(self, job_id: str) -> Any:

A job's result, as it stands.

async def jobs( self, *, state: str | None = None, mode: str | None = None, limit: int | None = None, cursor: str | None = None) -> Mapping[str, Any]:

This key's jobs, newest first, filtered by state and mode; cursor reads the next page.

async def cancel(self, job_id: str) -> Mapping[str, Any]:

Ask the service to cancel job job_id. A job already finished is left as it is.

def stream( self, job_id: str, after: int | None = None) -> AsyncEventStream:

Job job_id's events, from after after (an event id) or from the start.

async def capabilities(self) -> Mapping[str, Any]:

The surface as this key sees it; read once per client.

async def profiles(self) -> Any:

The solver's profiles and what each one produces, as the service lists them.

class AsyncEventStream:

The async iterator await client.stream(job_id) returns.

AsyncEventStream( client: AsyncClient, job_id: str, after: int | None = None)

job_id

last_seq

async def aclose(self) -> None:

Stop reading the stream and close its connection.

def result(self) -> Any:

The job's result, once the stream has reached its terminal event.

class AsyncJob:

A submitted job, for wait=False: read it, wait for it, stream it, or cancel it.

AsyncJob( client: AsyncClient, envelope: Mapping[str, Any])

envelope

async def refresh(self) -> Mapping[str, Any]:

Fetch the job again: its envelope, also kept on envelope.

async def result( self) -> Result[AsyncArtifact]:

Waits until the job ends, and returns its result.

def stream( self, after: int | None = None) -> AsyncEventStream:

The job's events, from after after (an event id) or from the start.

async def cancel(self) -> Mapping[str, Any]:

Ask the service to cancel the job. A job already finished is left as it is.

class AsyncSession(stalwart_agents.client.session._State):

A conversation over an AsyncClient. Made by client.session(...).

AsyncSession( client: AsyncClient, agent: str, **options: Any)

async def read( self, turn: str, *, traceparent: str | None = None) -> TurnPlan:

Read one turn of the conversation into a plan, carrying the session's context.

async def answered( self, plan: TurnPlan, results: Sequence[Mapping[str, Any]], *, traceparent: str | None = None) -> TurnPlan | None:

Report the calls a plan's chain asked for, and read the turn again with their results.

None when the results do not answer the chain.

async def confirm( self, plan: TurnPlan, *, traceparent: str | None = None) -> TurnPlan:

The person said yes to the plan's question: read the turn on from there.

async def thumbs( self, plan: TurnPlan, *, up: bool, corrected: str | None = None, step: int = 0) -> Mapping[str, Any]:

Post the person's thumb on a plan's step step as one feedback record; corrected is the tool they meant.

Check one tool call against the catalogue, the rulebook and the session's history: allow, confirm or refuse.

class AuthenticationError(stalwart_agents.client.ApiError):

401: no key, or a key that does not verify.

class BatchResult(stalwart_agents.client.results._Records[~A]):

verify_batch: one record per row, and counters derived from the records.

passed: int

How many rows met their expectation.

failed: int

How many rows did not.

void: int

How many rows reached no verdict: a stall, a refusal or an error.

hard_failures: int

How many failed rows have a hard gate: the ones that fail a release.

all_passed: bool

Whether the table has rows and every one of them passed.

class CassetteMiss(stalwart_agents.client.QlsError):

A request the cassette does not hold.

class CassetteStale(stalwart_agents.client.QlsError):

A cassette recorded under a catalogue or vocabulary digest that has since moved.

@dataclass(frozen=True)

class CertificateCheck:

What the checker established: verified, not_self_contained or failed, and why.

Returned for every outcome, a failed one included: an unverifiable certificate is an answer, not an exception. scope is the sentence that travels with it -- a check establishes a derivation or a refutation, never that a policy is safe.

CertificateCheck( outcome: str, reason: str, details: Mapping[str, Any] = <factory>, scope: str = 'This establishes that a derivation or a refutation is valid. It does not establish that the policy is safe, nor that the equations are the policy.')

outcome: str

reason: str

details: Mapping[str, Any]

scope: str = 'This establishes that a derivation or a refutation is valid. It does not establish that the policy is safe, nor that the equations are the policy.'

exit_code: int

The exit code stalwart check gives this outcome: 0 for verified.

class Client:

The Stalwart API, over HTTP, with plans and tool checks answered on your own host.

Args: api_key: else QLS_API_KEY; else ConfigurationError naming the variable. base_url: else QLS_BASE_URL; else https://api.stalwart.it: the solvers, the catalogues, the licence and its bundles. timeout: per HTTP request, not per solve. max_retries: for the calls that are safe to repeat (see _core.retry_delay). user_agent_suffix: appended to the user agent, so a customer's tooling is identifiable in the service's logs. runtime: who answers plans and tool checks. auto, the licensed local runtime where it is installed and licensed, and the router otherwise; local, the runtime alone; router, the router alone. router_url: else QLS_ROUTER_URL: the enterprise qls-router you run in your own environment, which answers plans and tool checks over HTTP. With neither a usable runtime nor a router, a plan or a check raises SelectorUnavailableError. own_identifiers: identifiers of your own that must never reach Stalwart's servers -- a customer-number format, a name -- each a string (matched ignoring case) or a compiled pattern. They are scanned for with the personal data. transport: an httpx transport, such as stalwart_agents.client.testing.replay(path).

Client( *, api_key: str | None = None, base_url: str | None = None, timeout: float = 30.0, max_retries: int = 3, user_agent_suffix: str | None = None, runtime: str = 'auto', router_url: str | None = None, own_identifiers: Sequence[Union[str, re.Pattern[str]]] = (), transport: httpx.BaseTransport | None = None)

router_url: str | None

The enterprise qls-router this client asks plans and tool checks of, where it names one.

def close(self) -> None:

Close the client's connections. with Client() as client: does it for you.

base_url: str

The API this client calls: base_url=, QLS_BASE_URL, or https://api.stalwart.it.

def licensed(self) -> Mapping[str, Any]:

Opens this machine's lease in the runtime, refreshing it with the service where one is due, and returns where it stands. Everything the runtime computes needs it: the code checks, the sources' index, the reply checks and answers, a catalogue's lints.

Raises: LicenceRequired: no licence is activated here, the service refused it, or it has lapsed.

def activate_runtime(self) -> Mapping[str, Any]:

Activates this machine's local runtime, once: its installation, the licence, the first lease.

def release_runtime(self) -> Mapping[str, Any]:

Frees this machine's seat on the licence, so another installation can take it.

def runtime_request(self) -> Mapping[str, Any]:

An air-gapped machine's activation request: its installation id and public key, for a file.

Carry the file to a machine that reaches the service and pass it to offline_lease there.

def offline_lease(self, request: Mapping[str, Any]) -> Mapping[str, Any]:

For an air-gapped machine's request file: its licence and a lease for the licence's whole term.

def install_offline(self, answer: Mapping[str, Any]) -> Mapping[str, Any]:

On the air-gapped machine: the licence and lease offline_lease answered, kept and opened.

def runtime_status(self) -> Mapping[str, Any]:

Which side answers plans and checks -- local, router or none -- and the licence's state.

def plan( self, turn: str, *, agent: str, catalogue: Any, context: Mapping[str, Any] | None = None, unread: str = 'refuse', traceparent: str | None = None) -> TurnPlan:

One turn read into a plan, statelessly. A conversation is session().

A turn that needs a calculation no tool answers is a compute step where this host can run it (compute), and is declined (compute off: ...) where it cannot.

def compute( self, step: Any, sources: Sequence[Mapping[str, Any]], *, model: Callable[[ComputePrompt], str], runner: Callable[[str, Any, str], stalwart_agents.sandbox.launch.Outcome] | None = None) -> stalwart_agents.compute.Computation:

A calculation no tool answers, run on this host: written by the caller's own model, admitted by the runtime and run in the sandbox image. See stalwart_agents.compute.

step is a plan's compute step (or its instruction); sources the results of the read steps it draws on, each {"tool", "result"}. model is any callable taking the prompt (a str) and returning the function's source; the SDK holds no credential and calls no provider. Nothing of the step is sent anywhere: the run is counted, and the count goes in the usage report.

Raises: LicenceRequired: no licence is activated here, the service refused it, or it has lapsed. QuotaExceededError: this month's compute_runs or compute_seconds are spent.

A conversation: the context the service returns, carried to the next turn.

def check_tool_call( self, tool: str, arguments: Mapping[str, Any], *, agent: str, context: Mapping[str, Any] | None = None, origins: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, turns: Sequence[str] | None = None, history: History | None = None, session_id: str = '', confirmation: Any = None, compensation_of: int | None = None, plan: Mapping[str, Any] | None = None, traceparent: str | None = None) -> ToolCheck:

One tool call checked, statelessly: history is the History the caller carries.

def register_catalogue( self, tools: Any, *, agent: str | None = None, agents: Sequence[Mapping[str, Any]] | None = None, source: str = 'auto', rulebook_ir: Mapping[str, Any] | None = None, stores: Mapping[str, Any] | None = None, budget: Mapping[str, Any] | None = None, modules: Mapping[str, Any] | None = None, code_check: Mapping[str, Any] | None = None, rulebook: str | None = None) -> stalwart_agents.client._models.CatalogueVersion:

Registers the tools -- an MCP tools/list, LangChain tool schemas or an OpenAPI document -- as a new version.

def compile_rule_sheet(self, sheet: str) -> dict[str, Any]:

A rule sheet compiled to rulebook IR, with its catalogue overlay and its lints against the newest catalogue. Nothing is kept.

def save_rule_sheet(self, sheet: str) -> dict[str, Any]:

Compiles the sheet, saves it, and registers it on the newest catalogue as a new version. A retry of the same save is one version.

def rule_sheets(self) -> dict[str, Any]:

The sheet in force, with its text, and the versions saved before it.

def catalogues(self) -> list[stalwart_agents.client._models.CatalogueVersion]:

Every registered version, newest first.

def catalogue( self, version: int | None = None) -> stalwart_agents.client._models.CatalogueVersion | None:

One version in full, or the newest; None where none is registered.

def restore_catalogue(self, version: int) -> stalwart_agents.client._models.CatalogueVersion:

Registers an older version again, as the newest.

def feedback( self, record: Mapping[str, Any] | Sequence[Mapping[str, Any]]) -> Mapping[str, Any]:

A user's thumb, or several, in the toolkit's record format. A retry is one record.

A record carries the words of the turn it rates, which the miner learns from, so it is screened before it is sent: one holding personal data raises PersonalDataError, and nothing is sent.

def hunt_attempt( self, fingerprint: str | None = None, *, outcome: str, kind: str = 'replay', rounds: int = 0, planted_digest: str | None = None, trace_digest: str | None = None, graph_digest: str | None = None, mode: str = '', owasp: Sequence[str] = (), project_id: str | None = None) -> Mapping[str, Any]:

What an attempt on a finding did: digests and the outcome, never a text. A retry is one attempt.

def vocabulary(self) -> stalwart_agents.client._models.HostedVocabulary:

The tenant's live hosted vocabulary.

def mine( self, model: Mapping[str, Any], **options: Any) -> stalwart_agents.client._models.MineRun:

Queues the hosted miner over the tenant's feedback; read the run with ledger(run).

def ledger(self, run_id: str) -> stalwart_agents.client._models.MineRun:

A learning run: its state, what it applied and its ledger.

def revert_vocabulary(self) -> stalwart_agents.client._models.VocabularyVersion:

Makes the version before the live one live again.

def account(self) -> stalwart_agents.client._models.Account:

This key's tier, entitlements, limits and this month's use of each meter.

def version(self) -> stalwart_agents.client._models.Version:

The commit, each image's version and digest, the solver's binary_version and the host this deployment runs.

def verify_guardrail( self, policy: 'FileLike | None', action: str, context: str, *, rulebook_ir: 'FileLike | None' = None, pins: Any = None, enumerate_contexts: bool = False, whole_policy: bool = False, prohibitions: bool = False, session_steps: int | None = None, frat_proof: bool = False, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

Can action fire without context? An UNSAT verdict says no, with a certificate; SAT says how, with a repair.

A SAT verdict is the answer to the question and raises nothing. enumerate_contexts is the wire's enumerate (a builtin in Python).

def generate_and_verify( self, prompt: str, action: str, context: str, *, provider: str, model: str, api_key: str | None = None, api_key_env: str | None = None, rounds: int = 1, base_policy: 'FileLike | None' = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Asks your own model for the policy, then verifies it: the draft, its verdict, a repair, the shipping candidate.

The model's key comes from api_key or the environment variable api_key_env names. It travels with this one request, and the SDK neither stores nor logs it.

def verify_batch( self, properties: 'FileLike', policy: 'FileLike | None' = None, *, rulebook_ir: 'FileLike | None' = None, rails_config: 'FileLike | None' = None, concepts: 'FileLike | None' = None, expectations: 'FileLike | None' = None, declared: 'FileLike | None' = None, whole_policy: bool = False, prohibitions: bool = False, intent_axis: bool = False, exactness: bool = False, sequence: bool = False, session_steps: int | None = None, transfer: bool = False, allow_oracle_refutations: bool = False, frat_proof: bool = False, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

A whole property table: one record per row, with counters derived from the records.

if result: is refused -- read .all_passed or .hard_failures.

The smallest pin sets under which each obligation holds; incomparable sets are all reported.

def enumerate_properties( self, policy: 'FileLike | None' = None, *, rulebook_ir: 'FileLike | None' = None, prohibitions: bool = False, intent_axis: bool = False, exactness: bool = False, sequence: bool = False, session_steps: int | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Candidate obligations read off the policy: rows with no id, expect or gate, for review.

The expectation table a sweep's rows derive, against a rulebook.

def replay_certificate( self, certificate: 'FileLike | None' = None, *, artifact_id: str | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Asks the service to replay a certificate. To check one here, offline, use check_certificate().

replay is replayed, failed or out_of_scope; out of scope is a sound refutation of a kind replay does not check, not an error.

def verify_graph( self, graph: 'FileLike', *, catalogue: 'FileLike | None' = None, rulebook: 'FileLike | None' = None, session_steps: int | None = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

An exported agent graph, verified before it is deployed: findings in the hunter's form.

The guardrail bug hunter: every way through to a forbidden action, each a finding with its fix.

def verify_forest( self, model: 'FileLike', properties: 'FileLike', *, anchors: 'FileLike | None' = None, profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, stream: bool = False, wait: bool = True) -> Any:

A stall on this side concludes nothing: it is not evidence that a property holds.

Checks a trained forest (an ONNX-ML file or a Treelite JSON dump, sent as it is) against robustness, monotonicity, fairness and reachability properties.

def enumerate_counterexamples( self, model: 'FileLike', search: 'FileLike', *, anchors: 'FileLike', profile: str | None = None, config: Mapping[str, Any] | None = None, labels: Mapping[str, Any] | None = None, wait: bool = True) -> Any:

Counterexamples searched around real records: proposals with no expect or gate, verdicts only.

def submit(self, request: Mapping[str, Any]) -> Job:

Any solve request, as the contract states it; inputs already described.

def job(self, job_id: str) -> Any:

A job's result, as it stands.

def jobs( self, *, state: str | None = None, mode: str | None = None, limit: int | None = None, cursor: str | None = None) -> Mapping[str, Any]:

This key's jobs, newest first, filtered by state and mode; cursor reads the next page.

def cancel(self, job_id: str) -> Mapping[str, Any]:

Ask the service to cancel job job_id. A job already finished is left as it is.

def stream( self, job_id: str, after: int | None = None) -> EventStream:

Job job_id's events, from after after (an event id) or from the start.

def capabilities(self) -> Mapping[str, Any]:

The surface as this key sees it; read once per client.

def profiles(self) -> Any:

The solver's profiles and what each one produces, as the service lists them.

class ComputePrompt(builtins.str):

The prompt the runtime built for a calculation: the text a model is given, and its parts.

A str, so a model callable that takes one string works as it is: it is the system part and the user part, in that order. system, user and schema are there for a callable that sends roles. The schema is the sources' shape -- field names and JSON types -- and never a value of theirs.

system: str

user: str

schema: Any

class ConcurrencyLimitError(stalwart_agents.client.ApiError):

429: every concurrent slot is taken. Retried, honouring Retry-After, before it is raised.

current: int | None

How many solves were running, or None.

limit: int | None

How many may run at once, or None.

class ConfigurationError(stalwart_agents.client.QlsError):

No key, a bad base URL, or a runtime asked for that is not there.

@dataclass(frozen=True)

class ConfirmationToken:

What a confirm answer hands back, for the call once a person approved it. Opaque.

ConfirmationToken(token: str, expires: int | None, wording: Any = None)

token: str

expires: int | None

wording: Any = None

@classmethod

def from_dict( cls, data: Mapping[str, Any]) -> ConfirmationToken:

A token as a check's answer carries it.

class CounterexampleResult(stalwart_agents.client.Result[~A]):

enumerate_counterexamples: proposals with no expect or gate, verdicts only.

findings: list[Mapping[str, Any]]

The counterexamples found around the anchors, each a proposal with its verdict.

class DeriveExpectationsResult(stalwart_agents.client.results._Records[~A]):

derive_expectations: the derived table.

class DerivePinsResult(stalwart_agents.client.results._Records[~A]):

derive_pins: the obligations and their pin sets; the caveat is in caveats.

class DigestMismatch(stalwart_agents.client.QlsError):

A download whose bytes do not hash to the digest the service named. No file is left behind.

class EnumeratePropertiesResult(stalwart_agents.client.results._Records[~A]):

enumerate_properties: candidate rows with no id, expect or gate; the caveat is in caveats.

@dataclass(frozen=True)

class Event:

One frame: its type, its place in the job's sequence, when, and its body.

Event( type: str, seq: int, job_id: str, at: str = '', data: Mapping[str, Any] = <factory>)

type: str

seq: int

job_id: str

at: str = ''

data: Mapping[str, Any]

terminal: bool

Whether this event ends the job's stream: succeeded, failed, cancelled, expired or refused.

class EventStream:

The sync iterator client.stream(job_id) returns.

EventStream( client: Client, job_id: str, after: int | None = None)

job_id

last_seq

def close(self) -> None:

Stops reading. The job carries on, and client.job(id) still answers.

def result(self) -> Any:

The job's result, once the iterator reached its terminal event.

@dataclass(frozen=True)

class Finding:

One hunt or graph finding: its rule, severity, location, witness, fix, replay state and OWASP entries.

Finding( rule: str, severity: str, raw: Mapping[str, Any], owasp: Sequence[OwaspEntry] = (), owasp_root_cause: str | None = None, advisory: Mapping[str, Any] | None = None)

rule: str

severity: str

raw: Mapping[str, Any]

owasp: Sequence[OwaspEntry] = ()

owasp_root_cause: str | None = None

advisory: Mapping[str, Any] | None = None

@classmethod

def from_dict(cls, data: Mapping[str, Any]) -> Finding:

A finding as the service sends it, its OWASP entries read from the finding or its detail.

location: Any

Where in the input the finding is, as the service reports it.

witness: Any

The concrete case that triggers the finding, where there is one.

fix: Any

The corrective guard, checked by scanning again, where there is one.

replay: Any

What became of the witness replayed on the runtime, where it was.

reproduction: str | None

reproduced, not_reproduced, not_yet_replayed, or not_applicable for a finding with no action to replay to, such as a rail matched against an advisory.

class ForestResult(stalwart_agents.client.results._Records[~A]):

verify_forest: one record per property. A stall on this side concludes nothing.

class GuardrailResult(stalwart_agents.client.Result[~A]):

verify_guardrail and generate_and_verify.

@dataclass

class History:

The checks a session has made: entries of names and digests, and the token covering them.

Opaque: the SDK carries it and records the caller's report of what ran. An entry edited by hand fails the next check with history_invalid.

History(entries: list[dict[str, Any]] = <factory>, token: str = '')

entries: list[dict[str, Any]]

token: str = ''

def to_dict(self) -> dict[str, Any]:

The history as the wire carries it, to store and restore with from_dict.

@classmethod

def from_dict(cls, data: Mapping[str, Any]) -> History:

A history as to_dict wrote it.

def after( self, entry: Mapping[str, Any], token: str) -> History:

The history once @p entry is added: at the step the service numbered it.

class HuntResult(stalwart_agents.client.Result[~A]):

hunt and verify_graph: findings, the coverage statement and the assumptions the scan used.

findings: list[Finding]

Every finding the hunt reported, each with its rule and severity.

summary: Mapping[str, Any]

The hunt's summary: its counts, as the service reports them.

scan: Mapping[str, Any]

What the scan read, as the service reports it.

coverage: Any

The OWASP coverage statement, at the policy level: the entries the hunter has rules for, those with no finding in this scan, and the one it has no rule for, with the reason.

nothing_to_read: list[Mapping[str, Any]]

The rules that raised nothing because nothing they read was declared, each with why.

assumptions: list[str]

Every claim the scan used as stated, such as a tool the catalogue declares sandboxed.

advisories_version: str | None

The advisory registry's version the scan matched rails against.

advisories_digest: str | None

The digest of that registry.

class IdempotencyKeyReusedError(stalwart_agents.client.ApiError):

409: an idempotency key spent on a different request.

class InputTooLargeError(stalwart_agents.client.ApiError):

413: an input past its cap. limit_bytes is the cap.

limit_bytes: int | None

The cap the input passed, in bytes, or None.

class InvalidRequestError(stalwart_agents.client.ApiError):

400: a request of the wrong shape. field names it where the service did.

field: str | None

The field the service named, or None.

class Job:

A submitted job, for wait=False: read it, wait for it, stream it, or cancel it.

Job( client: Client, envelope: Mapping[str, Any])

envelope

def refresh(self) -> Mapping[str, Any]:

Fetch the job again: its envelope, also kept on envelope.

def result( self) -> Result[Artifact]:

Blocks until the job ends, and returns its result.

def stream( self, after: int | None = None) -> EventStream:

The job's events, from after after (an event id) or from the start.

def cancel(self) -> Mapping[str, Any]:

Ask the service to cancel the job. A job already finished is left as it is.

class JobFailedError(stalwart_agents.client.QlsError):

A job that ended failed: no payload. cause is the classified reason.

JobFailedError(job_id: str, cause: str, stderr_tail: str = '')

job_id

cause

stderr_tail

class NotEntitledError(stalwart_agents.client.ApiError):

403: the key lacks a grant. entitlement names it.

entitlement: str | None

The grant the key lacks, or None.

class NotFoundError(stalwart_agents.client.ApiError):

404: no such object for this tenant.

@dataclass(frozen=True)

class OwaspEntry:

One OWASP entry a finding realises: its list, its edition-qualified id, title and link.

OwaspEntry(list: str, id: str, title: str, url: str)

list: str

id: str

title: str

url: str

@classmethod

def from_dict( cls, data: Mapping[str, Any]) -> OwaspEntry:

An entry as the service sends it.

class PersonalDataError(stalwart_agents.client.QlsError):

A request bound for Stalwart's servers holds personal data, so nothing was sent.

findings lists each match by its kind and its field path, never its value; the message says the same in words.

PersonalDataError(findings: Any)

findings

@dataclass(frozen=True)

class PersonalDataFinding:

One match: its kind (a key of KINDS) and where it is. Never the value.

PersonalDataFinding(kind: str, path: str)

kind: str

path: str

def words(self) -> str:

The finding as the error states it: "an e-mail address at tools[0].description".

@dataclass(frozen=True)

class PlanItem:

One step of a plan: a tool, ask, compute, decline or unread step, and the rule that made it.

A compute step carries its calculation in compute: the instruction, the result type and the indices of the steps whose results it reads (sources).

PlanItem( index: int, kind: str, text: str, decided_by: str, depends_on: Sequence[int], selection: Selection, compute: Mapping[str, Any] | None, guidance: stalwart_agents.client._models.Guidance | None, raw: Mapping[str, Any])

index: int

kind: str

text: str

decided_by: str

depends_on: Sequence[int]

selection: Selection

compute: Mapping[str, Any] | None

guidance: stalwart_agents.client._models.Guidance | None

raw: Mapping[str, Any]

@classmethod

def from_dict(cls, data: Mapping[str, Any]) -> PlanItem:

A step as a plan carries it.

class QlsError(builtins.Exception):

Every error the SDK raises.

class QuotaExceededError(stalwart_agents.client.ApiError):

429 on a monthly meter: waiting will not help until it resets. Never retried.

meter: str

The monthly meter that ran out.

used: float | None

How much of the meter was used this month, or None.

limit: float | None

The meter's monthly cap, or None.

resets_at: str | None

When the meter resets, as an ISO 8601 time, or None.

class RateLimitError(stalwart_agents.client.ApiError):

429: a per-minute or per-stream limit. Retried, honouring Retry-After, before it is raised.

class RefusedError(stalwart_agents.client.ApiError):

409 refused: a combination of choices that cannot answer the question asked.

slug is the registry's (REFUSALS), so code can branch on it without matching English; explanation is the binary's own sentence.

slug: str

The refusal's registry slug, to branch on.

explanation: str

The binary's own sentence about why.

stage: str | None

Where the refusal was decided, or None.

class ReplayResult(stalwart_agents.client.Result[~A]):

replay_certificate, run by the service: replay is replayed, failed or out_of_scope.

replay: str | None

replayed, failed or out_of_scope.

stands_as_evidence: bool | None

Whether the certificate stands as evidence of its verdict.

class Result(typing.Generic[~A]):

Every job's answer: the verdict, its caveats, its artefacts and the binary's own object.

Result( envelope: Mapping[str, Any], artifact: Callable[[Mapping[str, Any]], ~A])

envelope

job_id: str

mode: str

state: str

verdict_void: bool

status: str | None

decided_by: str | None

verdict: Mapping[str, Any]

diagnostics: list[Mapping[str, Any]]

caveats: list[Mapping[str, Any]]

experimental: bool

labels: Mapping[str, Any]

result: Any

payload: Any

payload_text: str | None

solve_seconds: float | None

artifacts: list[~A]

generation: Mapping[str, Any] | None

outcome: Mapping[str, Any] | None

def artifact(self, role: str) -> Optional[~A]:

The first artefact of @p role, or None.

certificate: Optional[~A]

The certificate artefact, where the job produced one.

patched_policy: Optional[~A]

The repaired policy artefact, where the job produced one.

corrective_action: Mapping[str, Any] | None

The guard the solver proposes, its SDS and its over-blocking outcome, where it proposed one.

shipping_candidate: Mapping[str, Any] | None

For a generation request: the draft to ship, where one verified.

@dataclass(frozen=True)

class Selection:

The reader's answer to one instruction: the call, or the question to ask instead.

Selection( tool: str | None, candidates: Sequence[str], missing: Sequence[str], arguments: Mapping[str, str], sources: Mapping[str, str], rewrites: Sequence[str], question: str, confirm: str | None, chain: Sequence[Mapping[str, Any]], verdict: str)

tool: str | None

candidates: Sequence[str]

missing: Sequence[str]

arguments: Mapping[str, str]

sources: Mapping[str, str]

rewrites: Sequence[str]

question: str

confirm: str | None

chain: Sequence[Mapping[str, Any]]

verdict: str

class SelectorUnavailableError(stalwart_agents.client.ConfigurationError):

Nothing here can answer a plan or a tool check: no licensed local runtime, and no router_url.

Plans and tool checks read a user's turn and the calls an agent makes, so they are answered on the customer's own host -- by the licensed runtime the stalwart-agents wheel carries, or by an enterprise qls-router the customer runs -- and never by Stalwart's hosted API. The message says which of the two to set up.

class ServiceError(stalwart_agents.client.ApiError):

5xx, after the retries. request_id is the service's, where it sent one.

ServiceError(status: int, body: Mapping[str, Any], request_id: str | None = None)

request_id

class Session(stalwart_agents.client.session._State):

A conversation over a Client. Made by client.session(...).

Session( client: Client, agent: str, **options: Any)

def read( self, turn: str, *, traceparent: str | None = None) -> TurnPlan:

One turn, read with the session's context; the next context is kept.

The first turn takes the console's settings where the host gave none (see the module).

def answered( self, plan: TurnPlan, results: Sequence[Mapping[str, Any]], *, traceparent: str | None = None) -> TurnPlan | None:

What the host's calls returned, each {"tool", "arguments", "result"}.

Where they are the chain the asking step named, they are sent now and the plan they make is returned; otherwise they travel with the next read.

def confirm( self, plan: TurnPlan, *, traceparent: str | None = None) -> TurnPlan:

The user's yes to the write the plan put to them: the plan that calls it.

def thumbs( self, plan: TurnPlan, *, up: bool, corrected: str | None = None, step: int = 0) -> Mapping[str, Any]:

The user's thumb on a plan, posted as a feedback record; a retry is one record.

A tool call checked with the session's history, turns and agent; the history moves on.

class StreamError(stalwart_agents.client.QlsError):

A stream that could not be resumed, by streaming or by polling.

class ToolCheck:

One tool call's answer: allow, confirm or refuse. No truth value: confirm is neither.

ToolCheck( document: Mapping[str, Any], history: History)

raw

decision: str

reason: str

reasons: list[str]

decided_by: str

detail: Mapping[str, Any]

pointer: str | None

rule: str | None

schema_partial: bool

confirmation

entry: Mapping[str, Any]

catalogue_version: int | None

history: History

class TurnPlan:

A turn read into steps, with the session's next context.

outcome is run, ask, declined or unread. A plan has no truth value: "did it work" means four different things here. compute is what the host asked for a calculation, run or off: a compute step in a plan asked off runs nowhere.

TurnPlan(document: Mapping[str, Any])

raw

outcome: str

declined_because: str

steps: list[PlanItem]

guidance: stalwart_agents.client._models.Guidance | None

confirm: str | None

saga: list[Mapping[str, Any]]

to_model: list[int]

context: Mapping[str, Any]

catalogue_digest: str | None

catalogue_version: int | None

vocabulary_digest: str | None

vocabulary_version: int | None

prompt_bytes: int | None

compute: str

for_model: list[PlanItem]

The unread steps the host's setting hands to its model; none on a declined plan.

chain: list[Mapping[str, Any]]

The calls the step that asks wants made to produce what it is missing.

__version__ = '0.0.0.dev0'

The package's version: 1.0. in a release, 0.0.0.dev0 in a checkout.

def async_session_from_json( client: AsyncClient, text: str) -> AsyncSession:

A session to_json wrote, restored over an async client.

def check_certificate( certificate: str | os.PathLike[str] | bytes, *, cnf: str | os.PathLike[str] | None = None) -> CertificateCheck:

Runs here, offline: checks a certificate on this machine with the vendored reference checker.

@p certificate is a path or the document's bytes. For an external_proof certificate the CNF its proof_log names is read from cnf, or from beside a certificate given by path.

def session_from_json( client: Client, text: str) -> Session:

A session to_json wrote, restored over @p client.