Guide: Stop Your AI Support Bot from Issuing a Bad Refund

A customer writes "My order arrived late, can I have $40 back?". Your bot reads the ticket, looks up the order, issues the refund, emails the customer and closes the ticket. That bot is an agent: a loop that asks a model what to do next, runs the tool the model picks, and repeats.

Without protection, the model decides alone. By the end of this guide:

• Every tool call is checked by AgentTrust before it runs.
• A blocked call never runs. The bot gets a plain BLOCKED result and hands the ticket to a person.
• If AgentTrust cannot be reached, the bot refuses to act.
• Rules live in the Console, so you change limits without redeploying.

Prerequisites. Python 3.12, uv, git, and an account at console.agent-trust.tech. You do not need a model API key: the bot uses a scripted model, so every run is identical.

Already have the template cloned? The @harness quickstart is the short version of this guide: same result, only the changes.

The bot has seven tools, and each one is wrapped with the same @harness checkpoint. The checkpoint is the only path from the model's choice to a tool's code, so no tool runs without being asked about first.

1. The model picks a tool. For a late order it picks issue_refund with an amount. The function has not run.
2. The checkpoint asks AgentTrust. It sends the tool name and its arguments, with the agent id. This is the pre-check.
3. AgentTrust compares the call with the manifest. The manifest, CERTIFY.YAML, lists the tools the bot may use and their limits. issue_refund is declared with a max of 1000.
4. The decision is recorded. An allowed call leaves an event, and so does a refused one.
5. The call runs or is refused. A $40 refund is under the limit, so the function runs. A $5,000 refund is over 1000, so it never runs. The bot gets a plain BLOCKED result and hands the ticket to a person.

Why the limit lives in the manifest

The model decides the amount, and a model can be wrong or be talked into a number. The manifest is a limit the model does not control. Because the rules live in the Console, you change a limit without redeploying the bot.

If AgentTrust cannot be reached

The bot refuses to act. A checkpoint that let calls through when it could not ask would fail open, which is the case you are trying to prevent. Refusing keeps money from moving without a decision.

What the check does not decide

The checkpoint decides whether a tool call may run. It does not decide whether the customer's story is true. A $900 refund on a $120 order is under the cap of 1000 and would be allowed, so set the cap to the largest amount you are willing to refund without a person.

Your agent exists in the Console, but the bot does not ask it anything yet. In this section you add the checkpoint: a wrapper that makes every tool call stop and ask AgentTrust before it runs. You make seven small edits across ten files. Each edit gives you the file to open, the marker to find, the code to paste and a check.

Before you start

• Your terminal is in python/. Every command and path in this section assumes it.
• uv run refund-bot agenttrust check prints four ticks. If it does not, fix the Console setup first: nothing in the code can make up for it.
• git status is clean, so git diff shows only your edits. Optional: run git switch -c add-checkpoint and commit after each edit, so you can undo any one of them.

What you will change

Edit	File	What it adds
TODO 1 Read the settings	app/core/config.py	The gateway URL, key, agent name and failure mode, read from the environment
TODO 2 Add the SDK	pyproject.toml	agentrust-py, pinned to one version
TODO 3 Write the checkpoint	app/tools/governance.py	governed_harness, the wrapper that asks AgentTrust before a tool runs
TODO 4 Wrap every tool	app/tools/__init__.py	The wrapper around each tool, outside the logging
TODO 5 Pass the mode through	app/agents/support_agent.py	The governed setting, handed down to the tools
TODO 6 Choose protection per request	app/services/tickets.py, app/api/routes/tickets.py, app/cli.py	--governed harness on the command line and ?governed=harness in the URL
TODO 7 Keep personal data out	app/tools/customers.py, app/llm/rules.py	No SSN or card number reaches the model, and a ticket is closed even when its first read is refused

Optional: look at the tools you are protecting

Nothing to add here. Every tool gets the same first argument, a small context holding the database and the current ticket.

python/app/tools/context.py

"""What every tool gets besides the model's arguments: the database, and the ticket being worked on."""
from __future__ import annotations

from dataclasses import dataclass

from sqlalchemy.orm import Session

@dataclass
class ToolContext:
    db: Session
    ticket_id: str | None    # the ticket the agent is resolving; refunds, emails and actions are linked to it

Look at issue_refund and notice what is missing: no maximum amount, no check against the order total. A refund function should not decide how much is too much. That belongs to rules you can change without a release, and you just wrote them in the Console.

python/app/tools/refunds.py (green = lines you add, and AgentTrust lines)

"""Refund tool: send money back to a customer.

There is deliberately no upper limit and no check against the order total, so the tutorial can show a
+ $5,000 refund on a $120 order. Limits are AgentTrust's job.
"""
from __future__ import annotations

from typing import Any

from ..repositories import orders, records
from .context import ToolContext

def issue_refund(ctx: ToolContext, order_id: str, amount: float, currency: str = "USD",
                 reason: str = "") -> dict[str, Any]:
    """Refund money to the customer of an order."""
    if orders.get_order(ctx.db, order_id) is None:
        return {"ok": False, "error": "no such order"}
    existing = records.find_refund(ctx.db, ctx.ticket_id, order_id, amount)
    if existing:
        return {"ok": True, "status": "duplicate", "refund_id": existing}
    return {"ok": True, "refund": records.add_refund(ctx.db, ctx.ticket_id, order_id, amount, currency, reason)}

Before any tool runs, a line goes to the action log. It proves what really ran, whatever the model says. A blocked refund will not be in it.

python/app/tools/action_log.py (green = lines you add, and AgentTrust lines)

"""The action log: one row per tool call, written before the tool runs.

It is the proof of what really happened, whatever the agent or the model says. Tests and the tutorial
+ read it to check what ran, and later, what AgentTrust stopped.
"""
from __future__ import annotations

from typing import Any

from ..repositories import records
from .context import ToolContext

def log_action(ctx: ToolContext, tool: str, args: dict[str, Any]) -> None:
    records.add_action(ctx.db, ctx.ticket_id, tool, args)

tool_spec describes each tool to the model. It skips *keyword-only* parameters, which is how every tool carries two hidden values, user and input. The model cannot see or fill them in, but the checkpoint reads them.

python/app/llm/base.py

"""The model port. The agent loop talks to a ``Model`` and never imports a provider SDK."""
from __future__ import annotations

import inspect
from dataclasses import dataclass
from typing import Any, Callable, Protocol

@dataclass
class Message:
    role: str                      # "user" | "assistant" | "tool"
    content: Any
    tool_call_id: str | None = None

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: dict[str, Any]     # a JSON Schema object

@dataclass
class ToolCall:
    id: str
    name: str
    args: dict[str, Any]

@dataclass
class FinalAnswer:
    text: str

class Model(Protocol):
    name: str

    def next(self, messages: list[Message], tools: list[ToolSpec]) -> ToolCall | FinalAnswer: ...

_JSON_TYPES = {str: "string", float: "number", int: "integer", bool: "boolean"}

def tool_spec(fn: Callable[..., Any]) -> ToolSpec:
    """Build a spec from a function's signature and docstring.

    Keyword-only parameters are skipped: they carry hidden context (``user``, ``input``) that
    the model must never see or fill in.
    """
    properties: dict[str, Any] = {}
    required: list[str] = []
    for name, param in inspect.signature(fn, eval_str=True).parameters.items():
        if param.kind is inspect.Parameter.KEYWORD_ONLY:
            continue
        json_type = _JSON_TYPES.get(param.annotation)
        if json_type is None:
            raise TypeError(f"{fn.__name__}({name}): unsupported type hint {param.annotation!r}")
        prop: dict[str, Any] = {"type": json_type}
        if param.default is inspect.Parameter.empty:
            required.append(name)
        else:
            prop["default"] = param.default
        properties[name] = prop
    doc = inspect.getdoc(fn) or ""
    return ToolSpec(fn.__name__, doc.splitlines()[0] if doc else fn.__name__,
                    {"type": "object", "properties": properties, "required": required})

How to make each edit

Every step below has the same shape, so you always know where to go:

1. Open the file named in the step. Paths are from the repo root, and you are working in the python/ folder.
2. Find the marker with your editor's search: TODO(AgentTrust n/11). Red lines show exactly what to remove.
3. Replace those lines with the code block. Copy it with the button in the corner, and match the indentation of the line you replaced.
4. Check it works before moving on.

Markers 8 to 11 belong to the direct API guide: leave them alone. ./scripts/todos.sh lists the TODOs still open. The details sections show each whole finished file, with the lines you add in green.

Shortcut: apply all seven TODOs automatically

If you only want the finished result, this script replaces every marker for you. Run it from the python/ folder, then read each step to see what changed. git diff shows it too.

python3 ../docs/guides/apply_snippets.py --only 1-7 --root ..
./scripts/todos.sh   # 7 of 11 done

What to expect as you go

• Progress. ./scripts/todos.sh goes from 0 of 11 done to 7 of 11 done, one more after each edit. TODOs 8 to 11 belong to the direct API guide and stay open.
• Checks. Most edits end with a one-line Check it. Run it before you move on.
• Tests. The six tests in tests/test_harness_governed.py are skipped until TODO 3. After that, four of them fail after TODOs 3, 4 and 5, and that is expected: they exercise the whole path, which is complete only after TODO 6. After TODO 6 they pass except one that waits for TODO 7, and after TODO 7 all six pass. Eleven other tests are skipped because they need a Postgres database. That is normal.

Read the AgentTrust settings TODO 1

Settings come from environment variables, so the same code runs on your laptop and on a server. You add four values: the gateway URL, your key, the agent name, and what to do when the gateway is unreachable. The last one defaults to closed (refuse to act), the opposite of the SDK default, so forgetting to set it never turns protection off.

Open python/app/core/config.py

Find TODO(AgentTrust 1/11). It appears twice: once inside class Settings, once inside get_settings().

First marker, below smtp_from: str. Replace the comment line with the four fields:

python/app/core/config.py

agentrust_gateway_url: str
agentrust_key: str
agentrust_agent_id: str
agentrust_failure_mode: str

Second marker, below smtp_from=os.environ.get(...). Replace the comment line with:

python/app/core/config.py

agentrust_gateway_url=os.environ.get("AGENTRUST_GATEWAY_URL", "").rstrip("/"),
agentrust_key=os.environ.get("AGENTRUST_KEY", ""),
agentrust_agent_id=os.environ.get("AGENTRUST_AGENT_ID", "refund-bot"),
agentrust_failure_mode=os.environ.get("AGENTRUST_FAILURE_MODE", "closed"),

See the whole finished file

python/app/core/config.py (green = lines you add, and AgentTrust lines)

"""Settings, read from environment variables, with defaults that work on a laptop with nothing installed.

| Variable       | Default                    | Used for                                  |
|----------------|----------------------------|-------------------------------------------|
| DATABASE_URL   | sqlite:///refund_bot.db    | where tickets, orders and refunds live    |
| SMTP_HOST      | (empty: emails stay local) | the mail server, Mailpit in Docker        |
| SMTP_PORT      | 1025                       |                                           |
| SMTP_FROM      | [email protected]        | the sender of every email                 |
"""
from __future__ import annotations

import os
from dataclasses import dataclass

@dataclass(frozen=True)
class Settings:
    database_url: str
    smtp_host: str
    smtp_port: int
    smtp_from: str
+     agentrust_gateway_url: str
+     agentrust_key: str
+     agentrust_agent_id: str
+     agentrust_failure_mode: str

def get_settings() -> Settings:
    """Read the settings now. Called when needed, so tests can change the environment."""
    return Settings(
        database_url=normalize_database_url(os.environ.get("DATABASE_URL", "sqlite:///refund_bot.db")),
        smtp_host=os.environ.get("SMTP_HOST", ""),
        smtp_port=int(os.environ.get("SMTP_PORT", "1025")),
        smtp_from=os.environ.get("SMTP_FROM", "[email protected]"),
+         agentrust_gateway_url=os.environ.get("AGENTRUST_GATEWAY_URL", "").rstrip("/"),
+         agentrust_key=os.environ.get("AGENTRUST_KEY", ""),
+         agentrust_agent_id=os.environ.get("AGENTRUST_AGENT_ID", "refund-bot"),
+         agentrust_failure_mode=os.environ.get("AGENTRUST_FAILURE_MODE", "closed"),
    )

def normalize_database_url(url: str) -> str:
    """SQLAlchemy reads ``postgresql://`` as the old psycopg2 driver; we use psycopg 3."""
    if url.startswith("postgresql://"):
        return "postgresql+psycopg://" + url[len("postgresql://"):]
    return url

Check it. uv run python -c "from app.core.config import get_settings; print(get_settings().agentrust_failure_mode)" prints closed.

Add the SDK TODO 2

Pin one exact version so nothing changes under you.

Open python/pyproject.toml

Find TODO(AgentTrust 2/11), the first line inside dependencies = [.

Replace that comment line with this line, keeping the quotes and trailing comma:

python/pyproject.toml

"agentrust-py==0.0.4",

uv sync

Check it. uv run python -c "import agentrust_sdk; print('sdk ok')" prints sdk ok.

Write the checkpoint TODO 3

This is the core of the guide. governed_harness takes one tool and returns a protected version of it, using the SDK's @harness to do the asking. The code around it fixes four SDK defaults and behaviours:

• "Needs review" is a no. The SDK lets it through by default.
• A failing check stops the tool. The SDK runs the tool anyway by default.
• Refusals become results. A refusal returns BLOCKED, an unreachable gateway returns UNAVAILABLE. The bot reads them like any other result instead of crashing.
• It says who is asking. @harness reads the user only from keyword arguments. Without that line every Console event shows the user as unknown.

It also refuses to start without a gateway URL and key. Otherwise the SDK would send checks to localhost:8000, your own app, and run every tool unchecked.

Open python/app/tools/governance.py. It holds a docstring, from __future__ import annotations and two TODO comments.

Find TODO(AgentTrust 3/11), the first comment.

Replace only that line with the code below. Keep the from __future__ line above it, and leave the TODO 8 comment at the bottom: it belongs to the direct API guide.

python/app/tools/governance.py

import functools
import os
from typing import Any, Callable

from ..core.config import get_settings

Tool = Callable[..., dict[str, Any]]

class GovernanceNotConfigured(RuntimeError):
    """AgentTrust is asked for, but the gateway url or the key is missing."""

def governed_harness(tool: Tool) -> Tool:
    """Guard a tool with ``@harness``: AgentTrust is asked before it runs, and again with what it returned.

    A refused call comes back as a BLOCKED result and a gateway that cannot be reached as UNAVAILABLE, so
    the agent loop and the model see a normal tool result instead of an exception.
    """
    settings = get_settings()
    if not settings.agentrust_gateway_url or not settings.agentrust_key:
        raise GovernanceNotConfigured("set AGENTRUST_GATEWAY_URL and AGENTRUST_KEY in .env (see .env-sample)")
    os.environ.setdefault("AGENTRUST_FAILURE_MODE", settings.agentrust_failure_mode)
    from agentrust_sdk import BlockedError, GatewayUnavailableError, harness

    guarded = harness(
        agent_id=settings.agentrust_agent_id,
        action=tool.__name__,
        base_url=settings.agentrust_gateway_url,
        api_key=settings.agentrust_key,
        block_on_block=True,        # a block stops the call
        block_on_review=True,       # so does "a person must look first": the SDK default lets it through
        raise_on_error=True,        # a failing check stops the call: the SDK default runs the tool anyway
    )(tool)

    @functools.wraps(tool)
    def call(*args: Any, **kwargs: Any) -> dict[str, Any]:
        kwargs.setdefault("user", "support-bot")        # @harness reads the user only from the keyword arguments
        try:
            return guarded(*args, **kwargs)
        except BlockedError as exc:
            return {"status": "BLOCKED", "tool": tool.__name__, "reason": exc.reason, "outcome": exc.outcome}
        except GatewayUnavailableError as exc:
            return {"status": "UNAVAILABLE", "tool": tool.__name__, "reason": str(exc)}
        except Exception as exc:                        # the check itself failed: do not run the tool
            return {"status": "UNAVAILABLE", "tool": tool.__name__, "reason": f"AgentTrust check failed: {exc}"}

    return call

Check it. uv run python -c "from app.tools.governance import governed_harness; print('guard ok')" prints guard ok.

Wrap every tool TODO 4

_guard returns the wrapper make_tools puts around each tool. With no mode it returns tools unchanged. Add the case for harness.

Order matters. The checkpoint wraps the *outside*, around the logging. If AgentTrust says no, the log line is never written, so a blocked call leaves no trace in the action log.

Open python/app/tools/__init__.py, function _guard.

Find TODO(AgentTrust 4/11), between if governed is None: and the TODO 9 comment.

Replace only that line. Leave the TODO 9 comment and the final raise ValueError as they are.

python/app/tools/__init__.py

if governed == "harness":
    from .governance import governed_harness

    return governed_harness

See the whole finished file

python/app/tools/__init__.py (green = lines you add, and AgentTrust lines)

+ """The tools the agent can call. This is where AgentTrust guards each tool.

Each tool is a plain function in this folder whose first argument is a ``ToolContext`` (the database and the
ticket being worked on). ``make_tools`` turns them into the functions the model sees:

* the context is filled in, so the model only sees the tool's own arguments;
+ * two hidden keyword-only arguments, ``user`` and ``input``, are added for AgentTrust to read later;
* every call is written to the action log before the tool runs.
"""
from __future__ import annotations

import functools
import inspect
from typing import Any, Callable

from sqlalchemy.orm import Session

from .action_log import log_action
from .context import ToolContext
from .customers import lookup_customer
from .email import send_email
from .orders import lookup_order, purge_orders
from .refunds import issue_refund
from .tickets import create_ticket, read_ticket, update_ticket

ALL_TOOLS = [read_ticket, lookup_order, lookup_customer, issue_refund, send_email, create_ticket, update_ticket,
             purge_orders]

Tool = Callable[..., dict[str, Any]]

def make_tools(db: Session, ticket_id: str | None, governed: str | None = None) -> dict[str, Tool]:
+     """The tools for one ticket, by name, ready for the model. ``governed`` names how AgentTrust guards them."""
    ctx = ToolContext(db, ticket_id)
    guard = _guard(governed)
    return {tool.__name__: guard(_for_the_model(tool, ctx)) for tool in ALL_TOOLS}

def _guard(governed: str | None) -> Callable[[Tool], Tool]:
+     """The wrapper that asks AgentTrust before a tool runs. With no mode set, tools run as they are.

    It is applied outside the action log, so a blocked call never enters the log of what really ran.
    """
    if governed is None:
        return lambda tool: tool
+     if governed == "harness":
+         from .governance import governed_harness

+         return governed_harness
    raise ValueError(f"unknown governed mode {governed!r}")

def _for_the_model(tool: Tool, ctx: ToolContext) -> Tool:
    parameters = list(inspect.signature(tool, eval_str=True).parameters.values())[1:]   # drop ctx
    hidden = [inspect.Parameter("user", inspect.Parameter.KEYWORD_ONLY, default="support-bot", annotation=str),
              inspect.Parameter("input", inspect.Parameter.KEYWORD_ONLY, default="", annotation=str)]
    visible = inspect.Signature(parameters)

    @functools.wraps(tool)
    def call(*args: Any, user: str = "support-bot", input: str = "", **kwargs: Any) -> dict[str, Any]:
        bound = visible.bind(*args, **kwargs)
        bound.apply_defaults()
        log_action(ctx, tool.__name__, dict(bound.arguments))
        return tool(ctx, **bound.arguments)

    call.__signature__ = inspect.Signature(parameters + hidden)
    del call.__wrapped__        # so inspect shows the signature above, not the original with ctx
    return call

Pass the mode through the agent loop TODO 5

The loop does not know AgentTrust exists: a BLOCKED result is just another result to hand back to the model. It only needs to pass governed down. The step limit stops a confused model looping forever. It protects your program, not your data.

Open python/app/agents/support_agent.py, function run. Its signature already accepts governed.

Find TODO(AgentTrust 5/11) and the line right under it. Replace both lines:

- # TODO(AgentTrust 5/11): take a governed= argument and pass it to make_tools. ...
- tools = make_tools(db, ticket_id)

with:

python/app/agents/support_agent.py

tools = make_tools(db, ticket_id, governed)

See the whole finished file

python/app/agents/support_agent.py (green = lines you add, and AgentTrust lines)

"""The agent loop: ask the model what to do, run the tool it names, give it the result, repeat.

The loop knows nothing about refunds or tickets. It only moves messages between the model and the tools,
so the model can be swapped (scripted, rule-based, later a real LLM) and the tools can be wrapped
+ (by AgentTrust) without changing this file.
"""
from __future__ import annotations

import json
import logging
from typing import Any, Callable

from sqlalchemy.orm import Session

from ..llm.base import FinalAnswer, Message, Model, ToolCall, tool_spec
from ..repositories import records, tickets
from ..tools import make_tools
from .prompts import ticket_request
from .state import RunResult

STEP_LIMIT_ANSWER = "Stopped: step limit reached"

log = logging.getLogger("refund_bot.agent")

def call_tool(fn: Callable[..., dict[str, Any]], call: ToolCall, request: str) -> dict[str, Any]:
    """Run one tool call. The request goes in as the hidden ``input`` argument. A tool that raises
    becomes an ERROR result, so the model can recover instead of the loop crashing."""
    try:
        return fn(**call.args, input=request)
    except Exception as exc:
        return {"status": "ERROR", "tool": call.name, "reason": str(exc)}

def summary(result: dict[str, Any]) -> str:
    """One short word for a tool result, for the log: ok, duplicate, an error, or BLOCKED and why."""
    if result.get("status") in ("ERROR", "BLOCKED", "UNAVAILABLE"):
        return f"{result['status']}: {result.get('reason', '')}"
    if result.get("ok") is False:
        return f"error: {result['error']}"
    if result.get("status") == "duplicate":
        return f"duplicate of {result['refund_id']}"
    for key, label in (("refund", "refund_id"), ("ticket", "ticket_id")):
        if key in result:
            item = result[key]
            return f"ok {item[label]}" + (f" ({item['status']})" if key == "ticket" else "")
    return "ok"

def run(ticket_id: str, model: Model, db: Session, max_steps: int = 10, governed: str | None = None) -> RunResult:
    """Work on one ticket until the model gives its final answer.

    ``max_steps`` is a local safety net so a looping model cannot run forever. It is not a governance control.
+     ``governed`` says how AgentTrust guards the tools: None (not at all), "harness" or "client".
    """
+     tools = make_tools(db, ticket_id, governed)
    specs = [tool_spec(fn) for fn in tools.values()]
    before = {t["ticket_id"] for t in tickets.list_tickets(db)}
    request = ticket_request(ticket_id)
    messages = [Message("user", request)]
    answer = STEP_LIMIT_ANSWER
    log.info("%s | start | model=%s", ticket_id, getattr(model, "name", type(model).__name__))

    for _ in range(max_steps):
        step = model.next(messages, specs)
        if isinstance(step, FinalAnswer):
            answer = step.text
            messages.append(Message("assistant", answer))
            break
        messages.append(Message("assistant", {"tool": step.name, "args": step.args}, tool_call_id=step.id))
        fn = tools.get(step.name)
        if fn is None:
            result = {"status": "ERROR", "tool": step.name, "reason": f"unknown tool {step.name!r}"}
        else:
            result = call_tool(fn, step, request)
        log.info("%s | call  | %s %s -> %s", ticket_id, step.name, json.dumps(step.args), summary(result))
        log.debug("%s | result | %s", ticket_id, json.dumps(result))
        messages.append(Message("tool", result, tool_call_id=step.id))

    ticket = tickets.get_ticket(db, ticket_id)
    log.info("%s | done  | ticket %s | answer: %s", ticket_id, ticket["status"] if ticket else "missing", answer)
    return RunResult(
        ticket_id=ticket_id,
        transcript=messages,
        answer=answer,
        ticket=ticket,
        actions=records.list_actions(db, ticket_id=ticket_id),
        created_tickets=[t for t in tickets.list_tickets(db) if t["ticket_id"] not in before],
    )

Choose protection per request TODO 6

Protection is off unless you ask, so you can run the same ticket both ways. Ask with --governed harness on the command line or ?governed=harness in the URL. This TODO touches four places. Work from the inside out:

A. Service python/app/services/tickets.py, function resolve_ticket (it already has a governed parameter). Replace the TODO(AgentTrust 6/11) line and the return run(...) line under it:

- # TODO(AgentTrust 6/11): add a governed= parameter and pass it to run. ...
- return run(ticket_id, RuleModel(), db)

with:

python/app/services/tickets.py

return run(ticket_id, RuleModel(), db, governed=governed)

B. Route python/app/api/routes/tickets.py, function resolve (it already reads ?governed=). Replace the marker and the return under it:

- # TODO(AgentTrust 6/11): read ?governed=harness and pass it on. ...
- return result_out(resolve_ticket(db, ticket_id))

with:

python/app/api/routes/tickets.py

return result_out(resolve_ticket(db, ticket_id, governed))

C. CLI option python/app/cli.py, near line 74, under listing.add_argument("--status", ...). Replace the marker line TODO(AgentTrust 6/11): add --governed to "ticket" with:

python/app/cli.py

ticket.add_argument("--governed", choices=["harness", "client"], help="ask AgentTrust before each tool runs")

D. CLI call same file, near line 105, inside if args.command == "ticket":. Replace the marker TODO(AgentTrust 6/11): pass args.governed on and the print(...) line under it:

- # TODO(AgentTrust 6/11): pass args.governed on. ...
- print(show(resolve_ticket(db, created["ticket_id"])))

with:

python/app/cli.py

print(show(resolve_ticket(db, created["ticket_id"], args.governed)))

See the finished service

python/app/services/tickets.py (green = lines you add, and AgentTrust lines)

"""The ticket workflow, shared by the command line and the API.

1. ``submit_ticket``: a customer writes in. This is not an agent action, so it is not in the action log.
2. ``resolve_ticket``: the agent works on an open ticket with the rule-based mock LLM.
"""
from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.orm import Session

from ..agents.state import RunResult
from ..agents.support_agent import run
from ..llm.rules import RuleModel
from ..repositories import tickets
from ..tools.context import ToolContext
from ..tools.tickets import create_ticket

log = logging.getLogger("refund_bot.tickets")

class TicketError(Exception):
    """A request about a ticket that cannot be carried out. ``status`` is the HTTP status to answer with."""

    def __init__(self, message: str, status: int) -> None:
        super().__init__(message)
        self.message, self.status = message, status

def submit_ticket(db: Session, customer_id: str, body: str, order_id: str = "") -> dict[str, Any]:
    """Open a ticket for a customer. Raises TicketError (400) for an unknown customer or order, or an empty body."""
    result = create_ticket(ToolContext(db, None), customer_id, body, order_id)
    if not result["ok"]:
        raise TicketError(result["error"], 400)
    ticket = result["ticket"]
    log.info("%s | submitted | customer %s, order %s: %s", ticket["ticket_id"], customer_id, order_id or "-", body)
    return ticket

def resolve_ticket(db: Session, ticket_id: str, governed: str | None = None) -> RunResult:
    """Let the agent work on an open ticket. Raises TicketError: 404 if unknown, 409 if not open."""
    ticket = tickets.get_ticket(db, ticket_id)
    if ticket is None:
        raise TicketError(f"no such ticket {ticket_id!r}", 404)
    if ticket["status"] != "open":
        raise TicketError(f"ticket {ticket_id} is already {ticket['status']}", 409)
+     return run(ticket_id, RuleModel(), db, governed=governed)

See the finished route

python/app/api/routes/tickets.py (green = lines you add, and AgentTrust lines)

"""Tickets: list them, submit one, read one, and let the agent resolve one."""
from __future__ import annotations

from dataclasses import asdict
from typing import Literal

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from ...agents.state import RunResult
from ...db.session import get_db
from ...repositories import tickets
from ...schemas.tickets import RunResultOut, Ticket, TicketIn
from ...services.tickets import TicketError, resolve_ticket, submit_ticket

router = APIRouter(prefix="/tickets", tags=["tickets"])

def result_out(result: RunResult) -> RunResultOut:
    return RunResultOut(ticket_id=result.ticket_id, answer=result.answer, ticket=result.ticket,
                        actions=result.actions, created_tickets=result.created_tickets,
                        transcript=[asdict(m) for m in result.transcript])

@router.get("", response_model=list[Ticket])
def list_tickets(status: str | None = None, db: Session = Depends(get_db)):
    """All tickets, or only those with this status (open, resolved, escalated)."""
    return tickets.list_tickets(db, status)

@router.post("", response_model=Ticket, status_code=201)
def create(ticket: TicketIn, db: Session = Depends(get_db)):
    """A customer writes in. The ticket starts open."""
    return submit_ticket(db, ticket.customer_id, ticket.body, ticket.order_id)

@router.get("/{ticket_id}", response_model=Ticket)
def read(ticket_id: str, db: Session = Depends(get_db)):
    ticket = tickets.get_ticket(db, ticket_id)
    if ticket is None:
        raise TicketError(f"no such ticket {ticket_id!r}", 404)
    return ticket

@router.post("/{ticket_id}/resolve", response_model=RunResultOut)
def resolve(ticket_id: str, governed: Literal["harness", "client"] | None = None, db: Session = Depends(get_db)):
+     """The agent works on an open ticket and answers with everything it did. ``?governed=harness`` asks AgentTrust first."""
+     return result_out(resolve_ticket(db, ticket_id, governed))

Check it. ./scripts/todos.sh says 6 of 11 done. uv run refund-bot ticket --help lists --governed {harness,client}.

Keep personal data out and handle a refused read TODO 7

Two last fixes, both about what the model sees and does.

A. No SSN or card number. The customer lookup returns them. The model never needs them, and once a model has seen something no later check can take it back.

Open python/app/tools/customers.py, function lookup_customer. Replace the TODO(AgentTrust 7/11) line and the customer = ... line under it:

- # TODO(AgentTrust 7/11): do not return the SSN or the card number. ...
- customer = orders.get_customer(ctx.db, customer_id)

with:

python/app/tools/customers.py

customer = orders.get_customer(ctx.db, customer_id)
if customer:
    customer = {key: value for key, value in customer.items() if key not in ("ssn", "card_number")}

B. A refused first read. If read_ticket itself is refused, the model never sees the ticket. Make the scripted model close it by id instead of giving up silently.

Open python/app/llm/rules.py, method _decide, inside if self._ticket is None:. Replace the marker and the two lines under it:

- # TODO(AgentTrust 7/11): escalate the ticket even when read_ticket itself was blocked. ...
- if self._failed:
-     return FinalAnswer(ESCALATED_TEXT)

with:

python/app/llm/rules.py

if self._failed:            # read_ticket was refused: close the ticket from the id in the request
    if "closed" in self._done:
        return FinalAnswer(ESCALATED_TEXT)
    ticket_id = TICKET_ID.search(str(messages[0].content)).group(0)
    return "update_ticket", {"ticket_id": ticket_id, "status": "escalated", "note": "Passed to a colleague."}

./scripts/todos.sh   # 7 of 11 done (8 to 11 belong to the direct API guide)
uv run pytest -q

Check it. 7 of 11 done, and the tests pass. They run the whole bot against an offline stand-in for the gateway, so you can check the wiring without a network.

If you get stuck

You see	What it means
You cannot find a marker	Search for the exact text, for example TODO(AgentTrust 3/11). TODO 1 appears twice and TODO 6 four times. ./scripts/todos.sh lists the markers still open and the files they are in.
todos.sh still counts a TODO as open	The marker comment is still in the file. A TODO is done only when its marker line is gone, so replace the comment: do not paste above it.
IndentationError or SyntaxError after pasting	Python is strict about indentation. Paste with the copy button and match the indentation of the line you replaced.
ModuleNotFoundError: No module named 'agentrust_sdk'	You skipped uv sync after TODO 2. Run it.
Four governed tests fail	Expected until TODO 6. If they still fail after TODO 7, run uv run pytest -q tests/test_harness_governed.py and read the first error.

You are done when ./scripts/todos.sh prints 7 of 11 done, uv run pytest -q passes with all six governed tests, and uv run refund-bot ticket --help lists --governed {harness,client}. Next, run the bot with the checkpoint on.

• Your key is limited by design. It can only ask questions, never change the manifest, mode or policies. Keep it in .env and rotate it in the Console if it leaks.
• It fails closed. With AGENTRUST_FAILURE_MODE=closed an unreachable gateway refuses every tool and the ticket stays open. open runs everything unchecked during an outage.
• It cannot wait for a human. A call that needs review is refused, not held and resumed. An agent key cannot read the reviewer's answer.
• The check before does not read the customer's words. Injection is caught only by the check after, and only on a tool that returns the text.
• The check after cannot undo. It withholds a result. It does not reverse a tool that already ran.
• Two extra network calls per guarded tool. Latency was not measured here, so measure it for your plan before you guard read-only lookups in production.

The shorter Python decorator, on the support assistant, is Stop Risky AI Agent Tool Calls in Python with @harness. Framework wrappers: LangGraph and CrewAI. The full guide is in the refund bot docs.