stalwart_agents.gateway

stalwart_agents.gateway: every tool call an agent makes, checked on its way to the tool.

Gateway: the core -- tools/list filtered to the agent, tools/call checked and dispatched only on allow, confirmations as elicitations, idempotency keys, the untrusted-read label, the undo stack and the audit record.
GatewayApp: the same over MCP's streamable HTTP transport.
stalwart gateway --config gateway.yaml: runs it.

The gateway is an enforcement point only if the tools are reachable through it alone: deploy it with the network policy shipped beside the container.

CONFIRMATION_UNAVAILABLE = 'confirmation_unavailable'

The answer a confirmation gets where nobody can be asked.

class Audit:

Append-only decisions, one JSON object a line; in memory, and in a file where one is named.

Args: path: the file each decision is appended to; in memory only where not given.

Example:

audit = Audit() record = audit.append(tool="charge_card", decision="block", reason="declined", arguments={"amount": 5}, ... dispatched=False, session="s-1", agent="desk") record["decision_kind"], len(audit.records) ('gating', 1)

Audit(path: str | None = None)

path

records: list[dict[str, Any]]

def append( self, *, tool: str, decision: str, reason: str, arguments: Mapping[str, Any], dispatched: bool, session: str, agent: str, policy_sha256: str = '') -> dict[str, Any]:

Records one decision, in the gating record's form.

Args: tool: the tool called. decision: allow or block. reason: why, as the check gave it. arguments: the call's arguments; only their digest is recorded. dispatched: whether the call was sent upstream. session: the session it was made in. agent: the agent that made it. policy_sha256: the policy's digest.

Returns: The record, as appended.

class Gateway:

The enforcement point. See the module's account.

Args: client: a stalwart_agents.client.Client: checks answered by the licensed runtime, or the service. catalogue: the tenant's catalogue, in the extracted-model format. upstreams: the MCP server holding each tool, by tool name; "*" for every tool not named. audit: where decisions are recorded. context: the host's context values for the gate. compensation_retries: how often a failed undo is tried again. sleep: waits between those tries.

Example:

gateway = Gateway(None, catalogue={"tools": [{"name": "get_order", "read_only": True}, ... {"name": "charge_card", "read_only": False}], ... "agents": [{"name": "desk", "allowed_tools": ["get_order"]}]}, upstreams={}) [tool["name"] for tool in gateway.tools_list("desk")] ['get_order']

Gateway( client: Any, *, catalogue: Mapping[str, Any], upstreams: Mapping[str, stalwart_agents.gateway.gateway.Upstream], audit: Audit | None = None, context: Mapping[str, Any] | None = None, compensation_retries: int = 3, sleep: Callable[[float], None] = <built-in function sleep>)

client

catalogue

upstreams

audit

context

retries

sleep

sessions: dict[str, GatewaySession]

def open( self, agent: str, *, can_elicit: bool = False, session_id: str | None = None) -> GatewaySession:

A new session for @p agent, as an MCP initialize makes one.

Args: agent: the agent the session is for. can_elicit: whether the client can be asked for a person's decision. session_id: the session's id; a new one where not given.

Returns: The session, open.

def session(self, identifier: str) -> GatewaySession:

The session @p identifier names. Raises KeyError for one that is not open.

Returns: The session.

def tools_list(self, agent: str) -> list[dict[str, Any]]:

The tools @p agent may reach, as MCP lists them, with the catalogue's annotations.

Returns: Each tool's name, description, input schema and annotations.

def call( self, session_id: str, name: str, arguments: Mapping[str, Any], *, elicit: Callable[[Mapping[str, Any]], str] | None = None) -> dict[str, Any]:

One tools/call: checked, dispatched upstream only on allow, recorded.

Args: session_id: the session the call is made in. name: the tool. arguments: the call's arguments. elicit: asks the client's person, where the session's client can be asked.

Returns: The MCP result: the upstream's own on a dispatched call; a refusal (isError) with one sentence otherwise, and nothing sent upstream.

def rollback( self, session_id: str, *, failed: str = 'the session was aborted') -> stalwart_agents.langchain.core.saga.RollbackReport:

Undoes the session's writes last-in first-out, each undo checked with compensation_of.

Args: session_id: the session to roll back. failed: why, as the report gives it.

Returns: The report: what was undone, and what remains.

Raises: UndoRefused: inside an undo the check refuses; the report records it. RuntimeError: inside an undo that failed upstream; it is tried again, then recorded.

def close( self, session_id: str, *, abort: bool = False) -> stalwart_agents.langchain.core.saga.RollbackReport | None:

Ends a session; an aborted one is rolled back first.

Args: session_id: the session to end. abort: roll its writes back first.

Returns: The rollback's report where the session was aborted with writes to undo; None otherwise.

Raises: KeyError: no session @p session_id is open.

class GatewayApp:

The ASGI application in front of a Gateway.

Args: gateway: the core. agents: the agent each bearer token authenticates as. admin_token: the token the admin route takes; the route is off where not given. elicitation_timeout: how long an elicitation waits for the person, in seconds; one not answered by then is declined. The call's session waits with it.

Example:

app = GatewayApp(Gateway(None, catalogue={"tools": [], "agents": []}, upstreams={}), agents={"t-1": "desk"}) app.agents, app.admin_token ({'t-1': 'desk'}, None)

GatewayApp( gateway: Gateway, *, agents: Mapping[str, str], admin_token: str | None = None, elicitation_timeout: float = 600.0)

gateway

agents

admin_token

timeout

@dataclass

class GatewaySession:

One MCP session: the agent it is for, the check's history, the labels it carries, the undo stack.

Example:

session = GatewaySession("s-1", "desk", checks=None) session.can_elicit, session.stack (False, [])

GatewaySession( id: str, agent: str, checks: Any, can_elicit: bool = False, carried: set[str] = <factory>, stack: list[dict[str, Any]] = <factory>, lock: _thread.lock = <factory>)

id: str

agent: str

checks: Any

can_elicit: bool = False

carried: set[str]

stack: list[dict[str, Any]]

lock: _thread.lock

class HttpUpstream:

An MCP server over streamable HTTP, at @p url.

Each gateway session gets an MCP session of its own: initialize, then notifications/initialized, before its first call, and the Mcp-Session-Id the server gave it on every later one. One session's calls never ride in another's. A session the server has ended (404) is initialized again once, and the call sent again: the server did not run it. end closes a gateway session's MCP session.

Args: url: the server's endpoint. headers: headers sent with every request, such as its credentials. transport: an HTTP transport, in a test.

Example:

upstream = HttpUpstream("https://tools.internal/mcp", headers={"Authorization": "Bearer t-1"}) upstream.url, upstream.headers ('https://tools.internal/mcp', {'Authorization': 'Bearer t-1'})

HttpUpstream( url: str, *, headers: Mapping[str, str] | None = None, transport: Any = None)

url

headers

transport

def call( self, name: str, arguments: Mapping[str, Any], meta: Mapping[str, Any]) -> Mapping[str, Any]:

tools/call on the server, in the MCP session of the gateway session it is made for.

Args: name: the tool. arguments: the call's arguments. meta: sent as the call's _meta, such as its idempotency key; nothing where empty.

Returns: The MCP result, {"content": [...], "isError": bool, ...}.

def end(self, session: str) -> None:

Ends gateway session @p session's MCP session: a DELETE, as the protocol has a client end one.

A server that keeps no sessions, or one that has gone, is not an error: the session is forgotten here either way.

__version__ = '0.0.0.dev0'

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

def gating_decision_id( policy_sha256: str, tool: str, action: str, argument_digest: str, context: Mapping[str, bool]) -> str:

The gating record's identity, as data/mcp_audit.py and AuditRecord.h compute it, byte for byte.

Args: policy_sha256: the policy's digest; empty where there is none. tool: the tool called. action: the action checked: the tool's name, for a gating record. argument_digest: SHA-256 of the call's canonical arguments. context: the host's context values, by name.

Returns: The identity: SHA-256, in hex.

Example:

first = gating_decision_id("", "charge_card", "charge_card", "ab12", {"business_hours": True}) first == gating_decision_id("", "charge_card", "charge_card", "ab12", {"business_hours": True}), len(first) (True, 64)