Build a governed agent in 10 minutes¶
By the end you will have an agent that reads a quote and places paper trades, where every call is gated by a signed grant and a policy, large orders wait for a human, retries cannot double-fire, and everything is in a tamper-evident log. No API key and no network: the "model" is a script.
All the code on this page is one file, examples/tutorial_governed_agent.py, which the test
suite runs. If a snippet here does not work, that is a bug in the docs, so please report it.
1. Install¶
Python 3.11 or 3.12. The core needs nothing else; framework adapters are extras
(keelgate[langgraph], keelgate[mcp], ...).
2. Declare what the agent may do¶
from keelgate.approvals import ApprovalQueue, ApprovalTier, Approver
from keelgate.audit import AuditLog, verify_chain
from keelgate.capabilities import GrantSigner, GrantVerifier, issue_grant
from keelgate.loop import InMemoryCheckpointStore, LLMPlanner, Loop, StopConditions
from keelgate.policy import PolicyContext, RegoEngine
from keelgate.testing import FakeLLM, Reply
from keelgate.tools import CallContext, SideEffect, ToolGateway, ToolRegistry, tool
A tool is a typed function with a capability and a side effect. Two things matter here: the
WRITE supplies an idempotency key, so retrying the same order can never place it twice, and
resource states exactly which facts the policy may look at, derived from validated arguments
(never from model prose).
class QuoteIn(BaseModel):
symbol: str = Field(pattern=r"^[A-Z]{1,5}$")
class QuoteOut(BaseModel):
symbol: str
price: float
class TradeIn(BaseModel):
symbol: str = Field(pattern=r"^[A-Z]{1,5}$")
notional: float = Field(gt=0, allow_inf_nan=False)
client_order_id: str = Field(min_length=1, max_length=64)
class TradeOut(BaseModel):
order_id: str
@tool(capability="market_data:read", side_effect=SideEffect.READ)
def quote(args: QuoteIn) -> QuoteOut:
"""Latest price for a symbol."""
return QuoteOut(symbol=args.symbol, price=187.25)
@tool(
capability="trade:paper_execute",
side_effect=SideEffect.WRITE,
# The same client_order_id can never place the order twice, even if the agent retries.
idempotency_key=lambda a: a.client_order_id,
# What the policy is allowed to know about this action, derived from *validated* input.
resource=lambda a: {"symbol": a.symbol, "notional": a.notional},
)
def paper_trade(args: TradeIn) -> TradeOut:
"""Place a PAPER trade. Nothing real is ever executed."""
BLOTTER.append(args.model_dump())
return TradeOut(order_id=f"paper-{len(BLOTTER)}")
3. Give it authority, and put the gate in front¶
signer = GrantSigner.generate() # holds the private key; in production, load it from a vault
verifier = GrantVerifier({signer.key_id: signer.public_key_pem()}, clock=lambda: AS_OF)
audit = AuditLog(clock=lambda: AS_OF)
approvals = ApprovalQueue(audit=audit, clock=lambda: AS_OF)
registry = ToolRegistry()
registry.register(quote)
registry.register(paper_trade)
gateway = ToolGateway(
registry=registry,
verifier=verifier,
engine=RegoEngine(), # the finance_basic policy pack, evaluated in-process
audit=audit,
approvals=approvals,
)
# A signed, expiring grant: this agent, this tenant, exactly these capabilities, a budget.
grant = issue_grant(
signer,
agent_id="research-agent",
tenant_id="acme",
capabilities=["market_data:read", "trade:paper_execute"],
max_cost=50,
ttl=timedelta(hours=1),
clock=lambda: AS_OF,
)
The grant is the agent's entire authority: which capabilities, which tenant, how much budget,
until when. Anything not named is refused. The gateway is the only way to run a tool, and
RegoEngine is the finance_basic policy pack: deny by default, a restricted-symbol list, a
per-order cap, daily exposure, trading hours, and approval tiers for large orders.
4. See the gate decide¶
async def propose(tool_name: str, arguments: dict[str, Any], **extra: Any) -> Any:
return await gateway.call(
tool_name=tool_name, arguments=arguments, grant_token=grant.token, context=context(**extra)
)
async def gate_demo() -> dict[str, Any]:
results: dict[str, Any] = {}
small = {"symbol": "AAPL", "notional": 5_000, "client_order_id": "t-1"}
results["allowed"] = await propose("paper_trade", small)
results["retry"] = await propose("paper_trade", small) # an exact retry replays, never re-runs
restricted = {"symbol": "TSLA", "notional": 1_000, "client_order_id": "t-2"}
results["denied"] = await propose("paper_trade", restricted)
big = {"symbol": "AAPL", "notional": 30_000, "client_order_id": "t-3"}
results["parked"] = await propose("paper_trade", big) # needs a human
request = approvals.get("acme", results["parked"].approval_id)
alice = Approver(approver_id="alice", tenant_id="acme", max_tier=ApprovalTier.EXPLICIT_SIGNOFF)
approvals.approve("acme", request.request_id, alice, signoff_code=request.signoff_code)
results["approved"] = await propose("paper_trade", big, approval_id=request.request_id)
return results
Running it prints the following (the parenthesised notes are ours, not program output):
allowed -> OK
retry -> OK (replayed: the order ran once)
denied -> DENIED policy_denied (TSLA is on the restricted list)
parked -> APPROVAL_REQUIRED EXPLICIT_SIGNOFF (30,000 is above the sign-off threshold)
approved -> OK (a human approved exactly these arguments)
Notice what did not happen. The denied order never reached your function. The parked order waited for a person, and the approval was good for those exact arguments, once. The model was never asked to be careful; none of this depends on it.
5. Let a model drive, durably¶
_checkpoints = InMemoryCheckpointStore() # use SqliteCheckpointStore(path) to survive restarts
def build_loop(llm: FakeLLM) -> Loop:
return Loop(
gateway=gateway,
registry=registry,
planner=LLMPlanner(llm, "any-model"),
checkpoints=_checkpoints,
grant_token=grant.token,
policy_context=lambda as_of: PolicyContext(
as_of=as_of, execution_mode="paper", limits=LIMITS, exposure={"daily_notional": 0}
),
stop=StopConditions(max_iterations=6), # also: tokens, dollars, a timeout, a goal predicate
audit=audit,
approvals=approvals,
clock=lambda: AS_OF,
)
The loop plans, acts, observes and verifies, checkpointing every step. Here a scripted model quotes MSFT, places a paper order and reports:
print("2. The loop (a scripted model: quote, then trade, then answer)")
script = [
Reply.call("quote", symbol="MSFT"),
Reply.call("paper_trade", symbol="MSFT", notional=2_000, client_order_id="loop-1"),
Reply.say("Bought 2,000 of MSFT at the quoted price of 187.25."),
]
result = asyncio.run(
build_loop(FakeLLM(script, indexed=True)).run(
goal="Check MSFT and buy a small position.",
tenant_id="acme",
agent_id="research-agent",
as_of=AS_OF,
run_id="tutorial-1",
)
)
print(f" goal reached: {result.ok}; trace id: {result.state.trace_id}")
Swap FakeLLM for a real client (AnthropicClient, OpenAIClient, GoogleClient, OllamaClient
or VLLMClient) and nothing else changes. The loop stops on explicit limits (StopConditions:
iterations, tokens, dollars, a timeout, a goal), and if the process dies it resumes from the last
checkpoint without repeating an order that already ran.
6. Prove what happened¶
chain = verify_chain(audit.records("acme"), tenant_id="acme")
print(f"3. Audit: {chain.records_checked} records, chain intact: {chain.ok}")
Every grant check, policy decision, approval and tool result is in a hash chain per tenant, written
before the side effect. Edit or delete a record and verify_chain fails. Run it from somewhere
that does not trust the writer for an independent check.
Where next¶
| You want to | Read |
|---|---|
| Understand why it is built this way | Concepts |
| Write your own rules | Custom policy pack |
| Use LangGraph | Governed LangGraph agent |
| Give Claude Desktop or Cursor governed tools | MCP server for Claude Desktop and Cursor |
| Trace, replay and test a run | Telemetry, replay and evals |
| Depend on Keelgate from your own library | Integration contract |
| Know what can and cannot go wrong | Security model |