stalwart_agents

The words a tool may carry, as @tool(...) and agent.yaml both spell them.

An agent: its tools, its rules, its model, and the agents it hands off to.

Args: name: the agent's name, as the catalogue and the trace call it. tools: plain functions (marked with @tool for their words), or tools read from the file. rulebook: the rules the agent's writes are checked against: a Colang file (1.0 or 2.x). The graph is proved against it, and at start-up it is registered with the catalogue, compiled by the service into the record the tool check enforces, wherever the newest version does not hold it already. model: "provider:model", read with the team's own key from the environment, or a function taking the conversation and returning the reply. hands_off_to: the agents this one may pass the conversation to. privilege: low, standard or high: what a hand-off from it is trusted with. on_unread: what to do with a turn the router cannot read, once the person was asked to say it again: refuse; model to hand it to the model; or ask to ask the person first, and hand it over on their yes (confirm(user, "approve")). Where not given, each conversation takes the console's setting at its first turn, as the catalogue bundle carries it, or refuse. system: what the model is told before the conversation. client: the connection to Stalwart; one from QLS_API_KEY where not given. storage: where runs are kept: sqlite:///runs.db or postgres://...; in memory where not given. atomic: whether a failed run is undone. max_steps: the most steps one run takes. verify: whether the graph is proved before it runs -- once per graph, so a rebuild of the same graph is free; a way through its checks raises NotVerified and nothing runs. False runs it unproved. verifier: what proves it; the hosted graph mode, through @p client, where not given. replies: the reply check's settings: rules (the reply rules, or their file), phrasings (the outcome phrasings, or their file), composer (model or template) and language. Each turn is screened for personal data and each reply held to what was done (stalwart_agents.replies); no check where not given. answers: grounded answers for a question no tool answers, with on_unread="answer": index (the source index, or its file), k, floor and composer (extract, or compose with this agent's model); those not given are the console's, read as on_unread is. See stalwart_agents.answers.

Example:

from stalwart_agents import tool @tool(writes=True, undo="refund_card") ... def charge_card(order_id: str, amount: float) -> str: ... return "charged" desk = Agent("desk", tools=[charge_card]) desk Agent('desk', tools=['charge_card']) desk.catalogue()["tools"][0]["compensation"] {'tool': 'refund_card'}

What agent.yaml declares: the agents, their tools, the stores and the rulebook.

Example:

declared = AgentFile(path=Path("agent.yaml"), agents={"desk": Agent("desk", ["get_order"], [], {}, 2, 3)}, ... tools={"get_order": Tool("get_order")}, stores={}, budget={}, modules={}, rulebook=None) declared.file 'agent.yaml' [node["id"] for node in declared.graph()["nodes"]] ['__start__', 'desk_model', 'desk_tools', '__end__']

agent.yaml says something this release cannot read, at a line of the file.

Example:

failure = AgentFileError("agent 'desk' lists 'refund', which no tools source declares", file="agent.yaml", line=4) print(failure) agent.yaml:4: agent 'desk' lists 'refund', which no tools source declares failure.line 4

The checker could not be run, so nothing was checked -- never a clean result.

Example:

print(HunterUnavailable("the hunter is not at ./qls")) the hunter is not at ./qls

Nothing in this package answers without an activated licence: the message says how to get one.

The runtime holds the licence's lease in its own memory and refuses every product operation without it -- plans, tool checks, reply checks, answers, the code checks and the rest. Only activating a licence, stalwart doctor and checking a certificate work with no key.

Example:

print(LicenceRequired()) # doctest: +ELLIPSIS this needs an activated Stalwart licence: sign up at https://stalwart.it, set QLS_API_KEY ... issubclass(LicenceRequired, StalwartError) True

The model's provider needs a key the environment does not hold: variable names it.

Nothing was sent anywhere: the key is checked before any request.

Example:

print(ModelKeyMissing("anthropic", "ANTHROPIC_API_KEY")) the model's provider 'anthropic' needs a key: set ANTHROPIC_API_KEY in the environment

The model named cannot be used here: the message says what to install or change.

Example:

print(ModelUnavailable("'acme:large' is not a provider this release knows")) 'acme:large' is not a provider this release knows

The agent has a way through its checks: findings names each, with its rule and tool.

Example:

NotVerified("the agent has a way through its checks", [{"rule": "QG018", "tool": "charge_card"}]).findings [{'rule': 'QG018', 'tool': 'charge_card'}]

What the agent said, and anything that waits for the person.

text is the reply; waiting the calls that need a decision before the run goes on. rated is what a thumb on it records (Agent.thumbs): the user's words and the router's reading of them -- each step's tool, verdict and deciding rule, the agent's visible tools -- and the tool the model called where the router handed it the turn. It never holds the reply's text, an argument value or a tool's result; it is None where no router read the turn.

Example:

reply = Reply("Order 4411 is refunded.") print(reply) Order 4411 is refunded. reply.waiting []

The agent declares a rulebook the tool check does not hold, and it could not be registered.

Nothing runs: a graph proved against rules the checks do not enforce would be trusted for something it does not do. The message says why, and what to register instead.

Example:

print(RulebookNotHeld("policy.co", "the service could not be reached")) the rulebook policy.co is not the one the tool check holds, and it could not be registered: the service could not be reached

A run took more steps than its limit: limit and steps say how many.

Example:

print(RunTooLong(60, 61)) the run took 61 steps and its limit is 60: raise max_steps, or give the agent fewer tools to choose between

Something stopped a command; the message says what to change.

Every error of this package is one, so one except StalwartError catches them all.

Example:

issubclass(AgentFileError, StalwartError), issubclass(RunTooLong, StalwartError) (True, True)

Two steps of a run wrote to the same part of its state at once: steps names them.

Example:

print(StateConflict("both wrote the messages", ["desk_tools", "billing_tools"])) two steps wrote the run's state at once (desk_tools and billing_tools): both wrote the messages

The store the runs are kept in could not be read or written: store names it.

Example:

print(StorageError("sqlite:///runs.db", "the database is locked")) the run store sqlite:///runs.db failed: the database is locked

A thumb that would record a false case, refused: the message says what to thumb instead.

Nothing is posted. A whole turn of several instructions is thumbed one step at a time; an answer the conversation decided -- a reply to a question, a confirmation -- is thumbed on the turn that named the tool.

Example:

print(ThumbRefused("a turn of 2 instructions is not one case: thumb the step the user meant")) a turn of 2 instructions is not one case: thumb the step the user meant

A tool as the file declares it: its words, its schema, the line that declared it, and what runs it.

call is the function a run calls -- the Python function, or the request to the MCP server or the HTTP operation -- and None for a tool declared by its words alone.

Example:

entry = Tool("charge_card", {"writes": True, "undo": "refund_card"}).catalogue_entry() entry["read_only"], entry["compensation"], entry["action"], entry["object"] (False, {'tool': 'refund_card'}, 'charge', 'card')

An agent lists a tool with nothing to run it: declare its source (python, mcp, openapi).

Example:

print(ToolUnavailable("'refund_card' has nothing to run it")) 'refund_card' has nothing to run it

What verify found: the lints, the findings, the scan's summary and whether the gate held.

Example:

result = VerifyResult(file="agent.yaml", lints=[], findings=[], scan={}, gate_failed=False) result.exit_code 0

A call that waits for a person: what it is, why, and the decisions it takes.

Example:

waiting = Waiting.of({"id": "i-1", "tool": "charge_card", "wording": "Charge 5.00 to the card?", ... "decisions": ["approve", "skip", "cancel_all"]}) waiting.tool, waiting.irreversible ('charge_card', False)

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

Writes the graph agent.yaml builds, as the verifier reads it, to @p out: stalwart verify --export-only.

Args: path: the agent.yaml. out: the file the graph is written to.

Returns: The file written.

Raises: AgentFileError: the file says something this release cannot read.

Example:

import tempfile with tempfile.TemporaryDirectory() as folder: ... path = Path(folder, "agent.yaml") ... _ = path.write_text("{agents: {desk: {tools: [get_order], hands_off_to: [billing]}, " ... "billing: {tools: [get_order]}}, tools: [{name: get_order}]}", encoding="utf-8") ... written = export(path, Path(folder, "agent.graph.json")) ... graph = json.loads(written.read_text(encoding="utf-8")) written.name, graph["format"] ('agent.graph.json', 'qls-graph/1')

Read @p path, an agent.yaml.

Args: fetch: read each MCP server's tools/list over the network; without it a server's tools are known by the names the file lists (a saved listing file is always read). transport: an HTTP transport for the MCP and OpenAPI calls, in a test.

Returns: What the file declares.

Raises: AgentFileError: at the line of anything this release cannot read -- an unknown word, an agent naming a tool no source declares, a hand-off to an agent the file does not have.

Example:

import tempfile with tempfile.TemporaryDirectory() as folder: ... path = Path(folder, "agent.yaml") ... _ = path.write_text("{agents: {desk: {tools: [get_order]}}, tools: [{name: get_order}]}", encoding="utf-8") ... declared = load(path) declared.agents["desk"].tools, declared.runs (['get_order'], 'atomic')

The report stalwart verify prints.

Args: result: what verify found.

Returns: The report: the lints, each finding in a block, the rules with nothing to read, and the gate.

Example:

print(render(VerifyResult(file="agent.yaml", lints=[], findings=[], scan={}, gate_failed=False)), end="") stalwart verify agent.yaml Lints at registration: none No findings. Gate: passed

Declares a plain function a tool, with the words of agent.yaml.

@tool(writes=True, undo="refund_card")
def charge_card(order_id: str, amount: Decimal) -> Charge: ...

Args: words: the tool's words, as TOOL_WORDS spells them.

Returns: What marks a function with @p words and returns it unchanged.

Raises: TypeError: a word is not one of TOOL_WORDS.

Example:

@tool(writes=True, undo="refund_card") ... def charge_card(order_id: str) -> str: ... return order_id getattr(charge_card, MARK) {'writes': True, 'undo': 'refund_card'} tool(write=True) Traceback (most recent call last): ... TypeError: @tool does not take write; it takes writes, undo, ...

Check the agent @p path declares: the code checks here, then the bug hunter on the service.

Args: path: the agent.yaml. keep: a directory to write the export, the catalogue and the findings into. dependencies: the lock file of the packages you install (requirements.txt, pyproject.toml, uv.lock or poetry.lock), for the advisory registry. replay: replay each graph finding's witness on the agents, every tool a stub. client: the connection the catalogue is registered, the hunt run and the replays posted through; one from QLS_API_KEY where not given. Where no lease is open in this process yet, it opens one, before anything runs.

Returns: What the check found, and whether the gate held.

Raises: AgentFileError: the file says something this release cannot read. LicenceRequired: no licence is activated here, the service refused it, or it has lapsed. HunterUnavailable: the hunt could not be run, so nothing was checked.

Example: No licence is an error, never a clean result, and a machine without one is told to activate one:

>>> import tempfile
>>> with tempfile.TemporaryDirectory() as folder:
...     path = Path(folder, "agent.yaml")
...     _ = path.write_text("{agents: {desk: {tools: [get_order]}}, tools: [{name: get_order}]}", encoding="utf-8")
...     try:
...         verify(path)
...     except (LicenceRequired, HunterUnavailable):
...         print("nothing was checked")
nothing was checked