stalwart_agents.langchain

stalwart_agents.langchain: Stalwart's routing, tool check and sagas inside LangChain and LangGraph agents.

stalwart_agents.langchain.middleware: StalwartRouterMiddleware and StalwartToolGuardMiddleware for create_agent, and register_catalogue, the tool importer.
stalwart_agents.langchain.callbacks: StalwartFeedbackCallback, a person's thumb as feedback.
stalwart_agents.langchain.graph: the graph parts -- StalwartState, stalwart_messages, the router node, the guarded tool node, declared-transition edges, the rollback and rewind, create_stalwart_agent -- and export_graph and replay_witness.
stalwart_agents.langchain.guard: Guard, the same decisions for plain Python, with no graph and no LangChain import.
stalwart_agents.langchain.server: confirmations resumed on any worker, streamed, and approved out of band.
stalwart_agents.langchain.trace and stalwart_agents.langchain.otel: a run's spans, rendered locally.
stalwart_agents.langchain.redteam: the attacker campaigns.
stalwart_agents.langchain.nemo: the NeMo Guardrails actions and rails (the [nemo] extra).

Everything here runs in the team's environment. No selector, vocabulary or check is in this package: every decision it acts on is made by the runtime on the same machine, or by the router the customer runs where the runtime is absent.

Names are loaded on first use, so from stalwart_agents.langchain.guard import Guard loads neither LangChain nor LangGraph.

@dataclass(frozen=True)

class Call:

One call a stub recorded: the tool, its arguments, its place in the run, and whether a rollback made it as an undo.

Example:

Call("refund_card", {"order_id": "7"}, 0) Call(tool='refund_card', arguments={'order_id': '7'}, index=0, undo=False)

Call( tool: str, arguments: Mapping[str, Any], index: int, undo: bool = False)

tool: str

arguments: Mapping[str, Any]

index: int

undo: bool = False

class Guard:

Routing, the tool check and sagas for plain Python.

Args: client: a stalwart_agents.client.Client; it answers through the licensed runtime, or through the enterprise qls-router it names. agent: the agent the calls are made for, as the catalogue names it. catalogue: the extracted-model catalogue; read for each tool's undo, retry safety and retryable errors. The check itself reads the tenant's own. tools: the tools a compensation may call, by name; a tool passed to run.call is added on first use. on_confirm: answers a confirmation with approve, skip or cancel_all; where not given, run.call raises ConfirmationRequired and the run pauses. on_unknown: answers an unknown outcome with happened or did_not_happen; where not given, OutcomeUnknown is raised and nothing is rolled back. on_stuck: answers a stuck rollback with retry, done_by_hand or abandon; where not given, SagaStuck is raised with the report. atomic: roll back on a terminal failure; False stops and keeps the stack. tracer: where spans go; none are kept by default. sleep: waits between an undo's tries; a test passes one that does not wait.

Example:

guard = Guard(None, agent="desk", catalogue={"tools": []}) guard.atomic True

Guard( client: Any, *, agent: str, catalogue: Mapping[str, Any] | stalwart_agents.langchain.core.catalogue.Catalogue | None = None, tools: Sequence[Any] = (), on_confirm: Callable[[Mapping[str, Any]], str] | None = None, on_unknown: Callable[[Mapping[str, Any]], str] | None = None, on_stuck: Callable[[stalwart_agents.langchain.core.saga.RollbackReport], str] | None = None, atomic: bool = True, tracer: stalwart_agents.langchain.core.tracing.Tracer | None = None, compensation_retries: int = 3, sleep: Callable[[float], None] | None = None)

client

agent

catalogue

tools: dict[str, Any]

on_confirm

on_unknown

on_stuck

atomic

tracer

compensation_retries

sleep

def session(self, **options: Any) -> Any:

A new conversation over the client, for this agent and catalogue: the SDK's Session.

Args: options: passed on to the client's session; catalogue defaults to this guard's.

Returns: The SDK's Session.

def route(self, turn: str, session: Any) -> Any:

@p turn read into a plan (run, ask, declined or unread) with the session's context.

Args: turn: the user's words. session: the conversation, from session.

Returns: The SDK's TurnPlan; the session keeps the next context.

def check( self, tool: Any, arguments: Mapping[str, Any], *, session: Any, origins: Mapping[str, str] | None = None, labels: Mapping[str, Any] | None = None, context: Mapping[str, Any] | None = None, confirmation: Any = None) -> Any:

One call checked with the session's history; nothing runs. The SDK's ToolCheck.

Args: tool: the tool, or its name. arguments: the call's arguments. session: the conversation, from session. origins: where each argument came from, by name; the model's where not given. labels: the labels the call carries. context: the host's context values for the gate. confirmation: a person's approval of exactly these arguments, where there was one.

Returns: The SDK's ToolCheck: allow, confirm or refuse, with the reasons.

@contextmanager

def run( self, session: Any, *, stack: Sequence[Mapping[str, Any]] = ()) -> Iterator[stalwart_agents.langchain.guard.Run]:

A saga over @p session: calls made through the Run it yields are undone together on a terminal failure.

Args: session: the conversation, from session. stack: the compensation stack of a run paused at a confirmation, to carry on with.

Returns: A context manager that yields the Run the block makes its calls through.

Raises: Whatever ended the run, after an atomic run's rollback: CallRefused, ToolFailed, RunCancelled, or the block's own exception. A rollback that got stuck raises SagaStuck instead, from the original failure.

class RespondingChatModel(langchain_core.load.serializable.Serializable, langchain_core.runnables.base.Runnable[-Input, +Output]):

Answers every call with what its function returns for the conversation so far.

The offline stand-in for a team's model in a campaign: a function of what the model was shown, so a suite can say exactly how the model treats planted text.

Example:

model = RespondingChatModel(respond=lambda messages: AIMessage(content=f"{len(messages)} message(s)")) model.invoke("hello").content '1 message(s)'

respond: Callable[[list[langchain_core.messages.base.BaseMessage]], langchain_core.messages.ai.AIMessage] = PydanticUndefined

def bind_tools( self, tools: Sequence[Any], **kwargs: Any) -> RespondingChatModel:

The function already names its tool calls; binding changes nothing.

Args: tools: the tools; not read. kwargs: binding options; not read.

Returns: This model.

class ScriptedChatModel(langchain_core.load.serializable.Serializable, langchain_core.runnables.base.Runnable[-Input, +Output]):

Returns its script's messages in order, whatever it is asked.

Each entry is a message, or a function of the conversation so far that returns one, for a model whose answer depends on what a tool returned.

Example:

model = ScriptedChatModel(script=[tool_call("get_order", {"order_id": "7"}, "call-1")]) model.invoke("where is order 7?").tool_calls[0]["name"] 'get_order' model.invoke("and now?").content 'done' model.turns 2

script: list[Any] = PydanticUndefined

def bind_tools( self, tools: Sequence[Any], **kwargs: Any) -> ScriptedChatModel:

The script already names its tool calls; binding changes nothing.

Args: tools: the tools; not read. kwargs: binding options; not read.

Returns: This model.

turns: int

How many times the model has been called.

Returns: The count of calls.

seen: list[list[langchain_core.messages.base.BaseMessage]]

The conversation the model was shown on each call, in order.

Returns: One list of messages per call.

class StalwartFeedbackCallback(langchain_core.callbacks.base.BaseCallbackHandler):

Remembers each run's turn and tool, and posts a thumb on it.

Args: client: a stalwart_agents.client.Client. agent: the agent the runs are for.

Example:

callback = StalwartFeedbackCallback(None, agent="desk") callback.runs {}

StalwartFeedbackCallback(client: Any, *, agent: str)

client

agent

runs: dict[str, dict[str, Any]]

def on_chain_start( self, serialized: Any, inputs: Any, *, run_id: uuid.UUID, parent_run_id: uuid.UUID | None = None, **kwargs: Any) -> None:

Remembers the run's turn: the last user message of a root run's input.

Args: serialized: the chain's serialised form; not read. inputs: the chain's input. run_id: the run's id. parent_run_id: the run it is part of; None for a root run. kwargs: what LangChain passes besides; not read.

def on_chain_end( self, outputs: Any, *, run_id: uuid.UUID, parent_run_id: uuid.UUID | None = None, **kwargs: Any) -> None:

Keeps the router's reading of a root run's turn, from the state the run ended with.

Nothing here raises: a state it cannot read leaves the run's record as the turn and tool alone.

Args: outputs: the chain's output: a graph's state for a root run. run_id: the run's id. parent_run_id: the run it is part of; None for a root run. kwargs: what LangChain passes besides; not read.

def on_chat_model_start( self, serialized: Any, messages: list[list[langchain_core.messages.base.BaseMessage]], *, run_id: uuid.UUID, parent_run_id: uuid.UUID | None = None, **kwargs: Any) -> None:

Remembers the turn from the model's messages, where the run's input did not carry it.

Args: serialized: the model's serialised form; not read. messages: the messages the model is shown. run_id: the run's id. parent_run_id: the run it is part of. kwargs: what LangChain passes besides; not read.

def on_tool_start( self, serialized: Any, input_str: str, *, run_id: uuid.UUID, parent_run_id: uuid.UUID | None = None, **kwargs: Any) -> None:

Remembers the tool the run called.

Args: serialized: the tool's serialised form; its name is read. input_str: the tool's input; not read. run_id: the run's id. parent_run_id: the run it is part of. kwargs: what LangChain passes besides; name is read where @p serialized has none.

def record( self, run_id: str | uuid.UUID, *, up: bool, corrected: str | None = None, step: int | None = None) -> dict[str, Any]:

The feedback record for @p run_id, as /v1/feedback takes it.

Args: run_id: the root run rated. up: a thumb up; False for a thumb down. corrected: the person's correction, where they gave one. step: the step rated, counting from 0, where the router read several instructions in the turn.

Returns: The record: the agent, the rated turn's words, the thumb and the tool; and, where a router read the turn, its verdict, deciding rule and the agent's visible tools.

Raises: KeyError: the callback saw no run with this id. ThumbRefused: a thumb the toolkit refuses, naming what to thumb instead.

def thumbs( self, run_id: str | uuid.UUID, *, up: bool = True, corrected: str | None = None, step: int | None = None) -> Mapping[str, Any]:

Posts the person's thumb on @p run_id: one record; a retry of the call is one record too.

Returns: The service's answer to the posted record.

Raises: KeyError: the callback saw no run with this id. ThumbRefused: a thumb the toolkit refuses; nothing is posted.

class StalwartRouterMiddleware(typing.Generic[~StateT, ~ContextT, ~ResponseT]):

Routes the latest user turn before the model is called. See the module's table.

Args: client: a stalwart_agents.client.Client: plans come from the licensed runtime, or the qls-router it names. agent: the agent the turns are read for. catalogue: the catalogue to plan against: an extracted model, or a registered version. on_unread: for a turn the router cannot read, once it asked the user to rephrase it: refuse; model to let the model answer it; or ask to let the model answer it once the user says yes.

Reads: messages, stalwart.session, stalwart.history, stalwart.plan. Writes: stalwart.session, stalwart.history, stalwart.plan (the plan, where it is, each call's argument origins, and rated: what a thumb on the reply records).

Example:

router = StalwartRouterMiddleware(None, agent="desk") router.on_unread 'refuse'

StalwartRouterMiddleware( client: Any, *, agent: str, catalogue: Any = None, on_unread: str = 'refuse')

state_schema = <class 'stalwart_agents.langchain.middleware.StalwartAgentState'>

The schema for state passed to the middleware nodes.

client

agent

catalogue

on_unread

def before_model(self, state: Any, runtime: Any) -> dict[str, Any] | None:

A new user turn read into a plan, kept in stalwart.plan; nothing on any other model call.

Args: state: the agent's state. runtime: LangGraph's runtime; not read.

Returns: The state update with the plan, the session and the history; None where the last message is not a user turn.

def wrap_model_call( self, request: langchain.agents.middleware.types.ModelRequest, handler: Callable[[langchain.agents.middleware.types.ModelRequest], langchain.agents.middleware.types.ModelResponse]) -> Any:

The plan's step in place of the model call, or the model call where the plan hands the turn on.

Args: request: the model call LangChain is about to make. handler: makes the model call.

Returns: The plan's step as the model's response, with the plan's progress as a state update; or the model's own response.

class StalwartToolGuardMiddleware(typing.Generic[~StateT, ~ContextT, ~ResponseT]):

Checks every tool call before it runs, whoever chose it.

Args: client: a stalwart_agents.client.Client. agent: the agent the calls are made for. catalogue: the extracted-model catalogue, for the confirmation's irreversible warning. context: the host's context values for the gate, or a function of the state.

Reads: stalwart.session, stalwart.history, stalwart.labels, stalwart.plan (the router's origins). Writes: messages (the call's result), stalwart.session, stalwart.history, stalwart.labels.

Example:

guard = StalwartToolGuardMiddleware(None, agent="desk") guard.agent 'desk'

StalwartToolGuardMiddleware( client: Any, *, agent: str, catalogue: Mapping[str, Any] | None = None, context: Mapping[str, Any] | Callable[[Any], Mapping[str, Any]] | None = None)

state_schema = <class 'stalwart_agents.langchain.middleware.StalwartAgentState'>

The schema for state passed to the middleware nodes.

client

agent

catalogue

context

def wrap_tool_call( self, request: langgraph.prebuilt.tool_node.ToolCallRequest, handler: Callable[[langgraph.prebuilt.tool_node.ToolCallRequest], Union[langchain_core.messages.tool.ToolMessage, langgraph.types.Command[Any]]]) -> Any:

Check, then run through @p handler only on allow; a refusal runs nothing.

Args: request: the tool call LangChain is about to run. handler: runs it.

Returns: A Command with the call's result, or the refusal in one sentence, and the session's update.

@dataclass

class StubTools:

The catalogue's tools as stubs, with the calls they recorded and the text planted in them.

Example:

stubs = StubTools({"tools": [{"name": "get_order"}, {"name": "refund_card"}]}) stubs.as_list(["get_order"])[0].invoke({}) 'fixture data' stubs.called("get_order"), stubs.called("refund_card") (True, False)

StubTools( catalogue: Mapping[str, Any], calls: list[Call] = <factory>, planted: dict[str, str] = <factory>)

catalogue: Mapping[str, Any]

calls: list[Call]

planted: dict[str, str]

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

The catalogue's entry for @p name.

Returns: The tool's entry, in the extracted-model format.

def as_list( self, names: Sequence[str] | None = None) -> list[langchain_core.tools.base.BaseTool]:

The stubs, all of them or those named, in catalogue order.

Args: names: the tools to take; every one where not given.

Returns: The stubs, as LangChain tools.

def plant(self, tool: str, text: str) -> None:

Put @p text in @p tool's stubbed output from its next call on.

Raises: KeyError: @p tool is not in the catalogue.

def called(self, tool: str) -> bool:

Whether @p tool's stub ran.

Returns: True where it recorded a call.

def calls_to(self, tool: str) -> list[Call]:

Every recorded call to @p tool, in order.

Returns: The calls.

__version__ = '0.0.0.dev0'

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

def arguments_for(tool: Mapping[str, Any], *, planted: str | None = None) -> dict[str, Any]:

Arguments for a call to @p tool: each declared property, filled by its declared kind.

@p planted, where given, goes into every string argument instead, which is how a campaign makes a call carry the text it planted.

Returns: The arguments, by name.

Example:

tool = {"name": "run_code", "input_schema": {"properties": {"source": {"type": "string"}}}, ... "argument_kinds": {"source": "code"}} arguments_for(tool) {'source': "print('witness')"} arguments_for(tool, planted="qls-r1") {'source': 'qls-r1'}

def current_idempotency_key() -> str | None:

The idempotency key of the write being made, inside a tool a guarded call runs; None outside one.

A tool that calls an API honouring idempotency keys passes it on, in the header the catalogue's idempotency: {header: ...} names.

Returns: The key, or None outside a guarded call.

Example:

current_idempotency_key() is None True

def fixture_value(schema: Mapping[str, Any] | None) -> Any:

The smallest value of a JSON schema's type: what a stub returns for it.

Args: schema: the JSON schema; None or empty for a tool that declares none.

Returns: The value: the first of an enum, an object of its properties' values, an empty list, zero, False, None, or FIXTURE_LINE for a string or no schema.

Example:

fixture_value({"type": "object", "properties": {"id": {"type": "integer"}, "note": {"type": "string"}}}) {'id': 0, 'note': 'fixture data'} fixture_value(None) 'fixture data'

def register_catalogue( tools: Sequence[Any], *, agent: str | None = None, client: Any, agents: Sequence[Mapping[str, Any]] | None = None, **options: Any) -> Any:

Registers @p tools as the tenant's catalogue: a new version, linted (POST /v1/catalogues).

Args: tools: the agent's LangChain tools. agent: the agent they are for. client: a stalwart_agents.client.Client. agents: every agent and the tools each may reach, where more than one shares the catalogue. options: the catalogue's own declarations, as register_catalogue on the client takes them (stores, budget, modules, rulebook_ir).

Returns: The SDK's CatalogueVersion: its version, digest and lint report.

Example:

class Client: ... def register_catalogue(self, definitions, **options): ... return [d["name"] for d in definitions], options["source"] register_catalogue([{"name": "get_order"}], agent="desk", client=Client()) (['get_order'], 'langchain')

def tool_call( name: str, arguments: dict[str, Any], call_id: str) -> langchain_core.messages.ai.AIMessage:

A model turn that calls one tool.

Args: name: the tool's name. arguments: the call's arguments. call_id: the call's id, which the tool's result answers.

Returns: The model's message, with the one tool call.

Example:

tool_call("get_order", {"order_id": "7"}, "call-1").tool_calls [{'name': 'get_order', 'args': {'order_id': '7'}, 'id': 'call-1', 'type': 'tool_call'}]

def tool_definitions(tools: Sequence[Any]) -> list[dict[str, Any]]:

The tools as LangChain tool definitions: name, description, argument schema and metadata.

Args: tools: LangChain tools (BaseTool), or definitions already in that shape.

Returns: One definition per tool, in the order given.

Example:

from langchain_core.tools import tool @tool ... def get_order(order_id: str) -> str: ... '''Look an order up.''' ... return order_id tool_definitions([get_order])[0]["name"] 'get_order'