AI Agent Code In Python: Approvals Bound To The Call, After This Month's Agent Framework Bugs

"""Server-side approvals for AI agent tool calls.

Each approval is bound to one agent, one tool and the exact arguments (a SHA-256 of canonical JSON),
is single-use and expires. Decisions are recorded on the server from an authenticated reviewer, never
relayed by the client, and nothing is ever approved by default.

Standard library only (sqlite3). Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import hashlib
import json
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Literal

JSON = dict[str, object]
Decision = Literal["approve", "reject", "modify"]


def args_hash(args: JSON) -> str:
    canonical = json.dumps(args, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(canonical.encode()).hexdigest()


@dataclass(frozen=True)
class Pending:
    request_id: str
    agent_id: str
    tool: str
    args: JSON


class ApprovalStore:
    def __init__(self, path: str = ":memory:", ttl: timedelta = timedelta(minutes=30)) -> None:
        self.db = sqlite3.connect(path, isolation_level=None)
        self.ttl = ttl
        self.db.execute(
            """CREATE TABLE IF NOT EXISTS approvals (
                 request_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, tool TEXT NOT NULL,
                 args_json TEXT NOT NULL, args_hash TEXT NOT NULL,
                 status TEXT NOT NULL CHECK (status IN ('pending','approved','rejected','used')),
                 reviewer TEXT, decided_at TEXT, expires_at TEXT NOT NULL)"""
        )

    def request(self, agent_id: str, tool: str, args: JSON) -> str:
        request_id = str(uuid.uuid4())
        expires = (datetime.now(timezone.utc) + self.ttl).isoformat()
        self.db.execute(
            "INSERT INTO approvals VALUES (?,?,?,?,?, 'pending', NULL, NULL, ?)",
            (request_id, agent_id, tool, json.dumps(args, sort_keys=True), args_hash(args), expires),
        )
        return request_id

    def pending(self) -> list[Pending]:
        rows = self.db.execute(
            "SELECT request_id, agent_id, tool, args_json FROM approvals WHERE status = 'pending'"
        ).fetchall()
        return [Pending(r[0], r[1], r[2], json.loads(r[3])) for r in rows]

    def decide(self, request_id: str, reviewer: str, decision: Decision,
               modified_args: JSON | None = None) -> None:
        """Call this from your authenticated review UI. `reviewer` comes from your SSO session."""
        if decision == "modify" and modified_args is None:
            decision = "reject"  # a 'modify' with nothing modified must not run the original call
        status = "rejected" if decision == "reject" else "approved"
        now = datetime.now(timezone.utc).isoformat()
        if decision == "modify" and modified_args is not None:
            self.db.execute(
                "UPDATE approvals SET args_json = ?, args_hash = ? WHERE request_id = ? AND status = 'pending'",
                (json.dumps(modified_args, sort_keys=True), args_hash(modified_args), request_id),
            )
        cur = self.db.execute(
            "UPDATE approvals SET status = ?, reviewer = ?, decided_at = ? "
            "WHERE request_id = ? AND status = 'pending'",
            (status, reviewer, now, request_id),
        )
        if cur.rowcount != 1:
            raise LookupError(f"no pending approval {request_id}")

    def consume(self, request_id: str, agent_id: str, tool: str, args: JSON) -> bool:
        """True exactly once, for the same agent, tool and arguments, before expiry."""
        cur = self.db.execute(
            "UPDATE approvals SET status = 'used' WHERE request_id = ? AND status = 'approved' "
            "AND agent_id = ? AND tool = ? AND args_hash = ? AND expires_at > ?",
            (request_id, agent_id, tool, args_hash(args), datetime.now(timezone.utc).isoformat()),
        )
        return cur.rowcount == 1

    def approved_args(self, request_id: str) -> JSON | None:
        row = self.db.execute(
            "SELECT args_json FROM approvals WHERE request_id = ? AND status = 'approved'", (request_id,)
        ).fetchone()
        return None if row is None else dict(json.loads(row[0]))


if __name__ == "__main__":
    store = ApprovalStore()
    payment_file = {"file": "payments_2026-10-15.xml", "rows": 42, "total_gbp": "18250.00"}
    rid = store.request("agent:recs-01", "submit_payment_file", payment_file)
    print("pending:", [p.tool for p in store.pending()])
    print("consume before decision:", store.consume(rid, "agent:recs-01", "submit_payment_file", payment_file))
    store.decide(rid, reviewer="j.smith@example.co.uk", decision="approve")
    tampered = {**payment_file, "total_gbp": "182500.00"}
    print("consume with changed arguments:", store.consume(rid, "agent:recs-01", "submit_payment_file", tampered))
    print("consume by another agent:", store.consume(rid, "agent:other", "submit_payment_file", payment_file))
    print("consume exact call:", store.consume(rid, "agent:recs-01", "submit_payment_file", payment_file))
    print("consume again:", store.consume(rid, "agent:recs-01", "submit_payment_file", payment_file))
    rid2 = store.request("agent:recs-01", "submit_payment_file", payment_file)
    store.decide(rid2, reviewer="j.smith@example.co.uk", decision="modify")  # modify without args
    print("modify without arguments ->", store.approved_args(rid2), "(rejected)")

Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Standard library only.

"""Call-time scope enforcement and approval-gated tool dispatch for a Claude tool-use loop.

- Scope is checked when a tool is called, not when tools are listed. Anything not listed is out of scope.
- Read-only tools run. Mutating tools are parked as a pending approval and the model is told to stop.
- Only the orchestrator executes an approved call, using the arguments stored on the server.
- Every decision is written to a hash-chained audit log.

Requires anthropic 1.14.0. Uses tool_choice {"type": "auto"} with strict tools, because forced tool use
returns a 400 error on Claude Opus 5.5 and Sonnet 5.5. Runs here against a stub client that returns real
SDK types, so no API key is needed. Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import hashlib
import json
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime, timezone

from anthropic.types import Message, MessageParam, ToolParam, ToolResultBlockParam, ToolUseBlock

from approvals import JSON, ApprovalStore


@dataclass
class AuditLog:
    entries: list[dict[str, object]] = field(default_factory=list)

    def write(self, **event: object) -> None:
        prev = str(self.entries[-1]["entry_hash"]) if self.entries else "0" * 64
        body = {**event, "prev_hash": prev, "at": datetime.now(timezone.utc).isoformat()}
        digest = hashlib.sha256(json.dumps(body, sort_keys=True, default=str).encode()).hexdigest()
        self.entries.append({**body, "entry_hash": digest})


@dataclass(frozen=True)
class Tool:
    spec: ToolParam
    run: Callable[[JSON], str]
    mutates: bool


@dataclass
class Dispatcher:
    agent_id: str
    scope: frozenset[str]  # the only tools this agent may call
    tools: dict[str, Tool]
    approvals: ApprovalStore
    audit: AuditLog

    def handle(self, call: ToolUseBlock) -> ToolResultBlockParam:
        args: JSON = dict(call.input)
        if call.name not in self.scope or call.name not in self.tools:
            self.audit.write(agent=self.agent_id, tool=call.name, args=args, outcome="refused_out_of_scope")
            return {"type": "tool_result", "tool_use_id": call.id, "is_error": True,
                    "content": f"{call.name} is out of scope for this agent. Stop and report to a person."}
        tool = self.tools[call.name]
        if tool.mutates:
            request_id = self.approvals.request(self.agent_id, call.name, args)
            self.audit.write(agent=self.agent_id, tool=call.name, args=args, outcome="pending_approval",
                             request_id=request_id)
            return {"type": "tool_result", "tool_use_id": call.id,
                    "content": f"Queued for human approval as {request_id}. Do not retry. Continue other work."}
        result = tool.run(args)
        self.audit.write(agent=self.agent_id, tool=call.name, args=args, outcome="ran")
        return {"type": "tool_result", "tool_use_id": call.id, "content": result}

    def execute_approved(self, request_id: str, tool_name: str) -> str:
        """Called by the orchestrator after a person approves. The model never calls this."""
        args = self.approvals.approved_args(request_id)
        if args is None or not self.approvals.consume(request_id, self.agent_id, tool_name, args):
            self.audit.write(agent=self.agent_id, tool=tool_name, request_id=request_id, outcome="not_approved")
            raise PermissionError(f"{request_id} is not an unused approval for {tool_name}")
        result = self.tools[tool_name].run(args)
        self.audit.write(agent=self.agent_id, tool=tool_name, args=args, request_id=request_id,
                         outcome="ran_after_approval")
        return result


def run_agent(create: Callable[..., Message], dispatcher: Dispatcher, task: str, model: str) -> str:
    messages: list[MessageParam] = [{"role": "user", "content": task}]
    while True:
        reply = create(model=model, max_tokens=2048, messages=messages,
                       system="You prepare operations work. A person approves anything that leaves the firm.",
                       tools=[dispatcher.tools[n].spec for n in sorted(dispatcher.scope & dispatcher.tools.keys())],
                       tool_choice={"type": "auto"})
        if reply.stop_reason != "tool_use":
            return "".join(b.text for b in reply.content if b.type == "text")
        messages.append({"role": "assistant", "content": reply.content})
        results = [dispatcher.handle(b) for b in reply.content if isinstance(b, ToolUseBlock)]
        messages.append({"role": "user", "content": results})


def strict_tool(name: str, description: str, properties: dict[str, object]) -> ToolParam:
    return {"name": name, "description": description, "strict": True,
            "input_schema": {"type": "object", "properties": properties,
                             "required": sorted(properties), "additionalProperties": False}}


class StubMessages:
    """Scripted replies built from real anthropic.types models, standing in for client.messages."""

    def __init__(self) -> None:
        usage = {"input_tokens": 900, "output_tokens": 120}
        calls = [
            {"type": "tool_use", "id": "toolu_01", "name": "read_breaks", "input": {"date": "2026-10-15"}},
            {"type": "tool_use", "id": "toolu_02", "name": "update_ssi", "input": {"account": "ACC-11893"}},
            {"type": "tool_use", "id": "toolu_03", "name": "submit_payment_file",
             "input": {"file": "payments_2026-10-15.xml"}},
        ]
        self.replies = [
            Message.model_validate({"id": "msg_01", "type": "message", "role": "assistant",
                                    "model": "claude-sonnet-5-5", "content": calls,
                                    "stop_reason": "tool_use", "stop_sequence": None, "usage": usage}),
            Message.model_validate({"id": "msg_02", "type": "message", "role": "assistant",
                                    "model": "claude-sonnet-5-5", "stop_reason": "end_turn",
                                    "stop_sequence": None, "usage": usage,
                                    "content": [{"type": "text", "text": "3 breaks found. Payment file "
                                                 "queued for approval. SSI change refused as out of scope."}]}),
        ]
        self.seen_tools: list[list[str]] = []

    def create(self, **kwargs: object) -> Message:
        tools = kwargs["tools"]
        assert isinstance(tools, list)
        self.seen_tools.append([str(t["name"]) for t in tools])
        return self.replies.pop(0)


if __name__ == "__main__":
    tools = {
        "read_breaks": Tool(strict_tool("read_breaks", "Read the day's reconciliation breaks (read-only).",
                                        {"date": {"type": "string"}}),
                            run=lambda a: json.dumps({"date": a["date"], "breaks": 3}), mutates=False),
        "submit_payment_file": Tool(strict_tool("submit_payment_file", "Submit a prepared payment file.",
                                                {"file": {"type": "string"}}),
                                    run=lambda a: f"submitted {a['file']}", mutates=True),
        "update_ssi": Tool(strict_tool("update_ssi", "Change standing settlement instructions.",
                                       {"account": {"type": "string"}}),
                           run=lambda a: "changed", mutates=True),
    }
    audit = AuditLog()
    dispatcher = Dispatcher(agent_id="agent:recs-01", scope=frozenset({"read_breaks", "submit_payment_file"}),
                            tools=tools, approvals=ApprovalStore(), audit=audit)
    stub = StubMessages()
    print("agent:", run_agent(stub.create, dispatcher, "Prepare today's breaks and payment file.",
                              model="claude-sonnet-5-5"))
    print("tools offered to the model:", stub.seen_tools[0])
    pending = dispatcher.approvals.pending()
    print("waiting for a person:", [(p.tool, p.args) for p in pending])
    dispatcher.approvals.decide(pending[0].request_id, reviewer="j.smith@example.co.uk", decision="approve")
    print("orchestrator:", dispatcher.execute_approved(pending[0].request_id, "submit_payment_file"))
    try:
        dispatcher.execute_approved(pending[0].request_id, "submit_payment_file")
    except PermissionError as exc:
        print("replay refused:", exc)
    for e in audit.entries:
        print(f"  {e['outcome']:<22} {e['tool']:<20} {str(e['entry_hash'])[:12]}")

Requires anthropic 1.14.0. Imports approvals.py. Runs against a stub that returns real anthropic.types.Message objects.