Agentic AI Architecture For Approvals You Can Prove: Pre-Tool Gates, Signed Sign-Off And Audit Logs In Python

"""Pre-tool-use gate with bound approvals and a hash-chained audit log.

Python 3.11+, pydantic 2.14.0, cryptography 50.0.2.
The agent side only holds the approval service's PUBLIC key, so it can check an
approval but can never create one. The private key stays with the approval service.
"""
from __future__ import annotations

import hashlib
import json
from datetime import datetime, timedelta, timezone
from enum import Enum
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
from pydantic import BaseModel, ConfigDict


class ToolCall(BaseModel):
    model_config = ConfigDict(frozen=True)
    tool: str
    args: dict[str, Any]
    agent_id: str          # which agent is asking
    on_behalf_of: str      # the person whose task this is
    session_id: str

    def digest(self) -> str:
        """SHA-256 over a canonical encoding, so an approval covers these exact arguments."""
        body = json.dumps(
            {"tool": self.tool, "args": self.args, "on_behalf_of": self.on_behalf_of},
            sort_keys=True, separators=(",", ":"), ensure_ascii=False,
        )
        return hashlib.sha256(body.encode()).hexdigest()


class Verdict(str, Enum):
    ALLOW = "allow"
    DENY = "deny"
    NEEDS_APPROVAL = "needs_approval"


class Approval(BaseModel):
    """Issued by the approval service after a named person reviews the exact call."""
    model_config = ConfigDict(frozen=True)
    call_digest: str
    approver_id: str
    expires_at: datetime
    signature: str

    @staticmethod
    def payload(call_digest: str, approver_id: str, expires_at: datetime) -> bytes:
        return f"{call_digest}|{approver_id}|{expires_at.isoformat()}".encode()


def sign_approval(key: Ed25519PrivateKey, call: ToolCall, approver_id: str, ttl_minutes: int = 15) -> Approval:
    """Runs inside the approval service, after a named person has reviewed the exact call."""
    expires = datetime.now(timezone.utc) + timedelta(minutes=ttl_minutes)
    sig = key.sign(Approval.payload(call.digest(), approver_id, expires)).hex()
    return Approval(call_digest=call.digest(), approver_id=approver_id, expires_at=expires, signature=sig)


class GateResult(BaseModel):
    verdict: Verdict
    reason: str
    call_digest: str


# Deterministic policy. Unknown tools are denied, not allowed.
READ_ONLY = {"get_trade", "search_procedures"}
NEEDS_SIGN_OFF = {"amend_trade_booking", "send_client_email"}
EGRESS_ALLOWLIST = {"api.internal.example.co.uk"}
MAX_AMEND_NOTIONAL_GBP = 5_000_000


def evaluate(call: ToolCall) -> tuple[Verdict, str]:
    if call.tool in READ_ONLY:
        return Verdict.ALLOW, "read-only tool"
    if call.tool == "http_get":
        host = urlsplit(str(call.args.get("url", ""))).hostname or ""
        if host in EGRESS_ALLOWLIST:
            return Verdict.ALLOW, f"egress to {host} is allow-listed"
        return Verdict.DENY, f"egress to {host or 'unknown host'} is not allow-listed"
    if call.tool == "amend_trade_booking":
        if float(call.args.get("notional_gbp", 0)) > MAX_AMEND_NOTIONAL_GBP:
            return Verdict.DENY, "above the amendment limit for this agent, route to a desk supervisor"
        return Verdict.NEEDS_APPROVAL, "changes a booked trade"
    if call.tool in NEEDS_SIGN_OFF:
        return Verdict.NEEDS_APPROVAL, "external or record-changing action"
    return Verdict.DENY, f"tool '{call.tool}' is not on the allow-list"


class AuditLog:
    """Append-only JSON Lines file. Each entry carries the hash of the one before it."""

    def __init__(self, path: Path) -> None:
        self.path = path
        self.path.touch(exist_ok=True)

    def _last_hash(self) -> str:
        lines = self.path.read_text().splitlines()
        return json.loads(lines[-1])["entry_hash"] if lines else "0" * 64

    def append(self, event: str, call: ToolCall, result: GateResult, approver_id: str | None = None) -> None:
        entry = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "event": event,
            "session_id": call.session_id,
            "agent_id": call.agent_id,
            "on_behalf_of": call.on_behalf_of,
            "tool": call.tool,
            "call_digest": result.call_digest,
            "verdict": result.verdict.value,
            "reason": result.reason,
            "approver_id": approver_id,
            "prev_hash": self._last_hash(),
        }
        entry["entry_hash"] = hashlib.sha256(json.dumps(entry, sort_keys=True).encode()).hexdigest()
        with self.path.open("a") as f:
            f.write(json.dumps(entry, sort_keys=True) + "\n")

    def verify(self) -> bool:
        prev = "0" * 64
        for line in self.path.read_text().splitlines():
            entry = json.loads(line)
            claimed = entry.pop("entry_hash")
            if entry["prev_hash"] != prev:
                return False
            if hashlib.sha256(json.dumps(entry, sort_keys=True).encode()).hexdigest() != claimed:
                return False
            prev = claimed
        return True


class ApprovalGate:
    def __init__(self, approval_service_key: Ed25519PublicKey, audit: AuditLog) -> None:
        self._key = approval_service_key
        self.audit = audit

    def check(self, call: ToolCall, approval: Approval | None = None) -> GateResult:
        verdict, reason = evaluate(call)
        result = GateResult(verdict=verdict, reason=reason, call_digest=call.digest())
        if verdict is Verdict.NEEDS_APPROVAL and approval is not None:
            result = self._verify(call, approval, result)
        self.audit.append("gate_decision", call, result, approval.approver_id if approval else None)
        return result

    def _verify(self, call: ToolCall, a: Approval, pending: GateResult) -> GateResult:
        try:
            self._key.verify(bytes.fromhex(a.signature), Approval.payload(a.call_digest, a.approver_id, a.expires_at))
        except (InvalidSignature, ValueError):
            return pending.model_copy(update={"verdict": Verdict.DENY, "reason": "approval signature invalid"})
        if a.call_digest != call.digest():
            return pending.model_copy(update={"verdict": Verdict.DENY, "reason": "approval was for different arguments"})
        if a.expires_at <= datetime.now(timezone.utc):
            return pending.model_copy(update={"verdict": Verdict.DENY, "reason": "approval expired"})
        if a.approver_id == call.on_behalf_of:
            return pending.model_copy(update={"verdict": Verdict.DENY, "reason": "requester cannot approve their own action"})
        return pending.model_copy(update={"verdict": Verdict.ALLOW, "reason": f"approved by {a.approver_id}"})


if __name__ == "__main__":
    service_key = Ed25519PrivateKey.generate()  # demo only: in production this lives in the approval service's HSM or KMS
    log = AuditLog(Path("audit.jsonl"))
    gate = ApprovalGate(service_key.public_key(), log)
    base = {"agent_id": "ops-agent-01", "on_behalf_of": "j.patel", "session_id": "s-1001"}

    read = ToolCall(tool="get_trade", args={"trade_id": "T-88213"}, **base)
    amend = ToolCall(tool="amend_trade_booking", args={"trade_id": "T-88213", "settle_date": "2026-10-12", "notional_gbp": 1_250_000}, **base)
    tampered = ToolCall(tool="amend_trade_booking", args={"trade_id": "T-88213", "settle_date": "2026-10-12", "notional_gbp": 4_900_000}, **base)
    exfil = ToolCall(tool="http_get", args={"url": "https://paste.example.net/upload"}, **base)

    print(gate.check(read).verdict.value)              # allow
    print(gate.check(amend).verdict.value)             # needs_approval: nothing runs yet
    print(gate.check(exfil).reason)                    # egress denied
    ok = sign_approval(service_key, amend, approver_id="s.okafor")
    print(gate.check(tampered, ok).reason)             # approval was for different arguments
    print(gate.check(amend, ok).reason)                # approved by s.okafor
    self_ok = sign_approval(service_key, amend, approver_id="j.patel")
    print(gate.check(amend, self_ok).reason)           # requester cannot approve their own action
    forged = ok.model_copy(update={"approver_id": "x.unknown"})
    print(gate.check(amend, forged).reason)            # approval signature invalid
    print("audit chain intact:", log.verify())

"""A Claude tool loop where every tool call passes the gate before anything runs.

Python 3.11+, anthropic 1.13.0. Needs ANTHROPIC_API_KEY. Trade data is sample data.
"""
from __future__ import annotations

import json
from typing import Any, Callable

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

from approval_gate import Approval, ApprovalGate, ToolCall, Verdict

MODEL = "claude-sonnet-5-5"
MAX_TURNS = 8  # a hard budget: the loop stops even if the model wants to continue
# Sonnet 5.5 rejects forced tool use (tool_choice "any" or "tool") with a 400, so tools are strict and choice is auto.

TOOLS: list[ToolParam] = [
    {
        "name": "get_trade",
        "description": "Read one booked trade by its ID.",
        "strict": True,
        "input_schema": {
            "type": "object",
            "properties": {"trade_id": {"type": "string"}},
            "required": ["trade_id"],
            "additionalProperties": False,
        },
    },
    {
        "name": "amend_trade_booking",
        "description": "Propose a settlement-date change to a booked trade. A person approves it before it runs.",
        "strict": True,
        "input_schema": {
            "type": "object",
            "properties": {
                "trade_id": {"type": "string"},
                "settle_date": {"type": "string", "description": "ISO date"},
                "notional_gbp": {"type": "number"},
            },
            "required": ["trade_id", "settle_date", "notional_gbp"],
            "additionalProperties": False,
        },
    },
]


def get_trade(trade_id: str) -> dict[str, Any]:
    # Sample data standing in for a read-only replica of the booking system.
    return {"trade_id": trade_id, "isin": "GB00SAMPLE001", "settle_date": "2026-10-13", "notional_gbp": 1_250_000}


def amend_trade_booking(trade_id: str, settle_date: str, notional_gbp: float) -> dict[str, Any]:
    return {"trade_id": trade_id, "settle_date": settle_date, "status": "amended"}


EXECUTORS: dict[str, Callable[..., dict[str, Any]]] = {
    "get_trade": get_trade,
    "amend_trade_booking": amend_trade_booking,
}


def run(client: anthropic.Anthropic, gate: ApprovalGate, task: str, ctx: dict[str, str]) -> dict[str, Any]:
    messages: list[MessageParam] = [{"role": "user", "content": task}]
    pending: list[dict[str, Any]] = []
    for _ in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL, max_tokens=4096, tools=TOOLS, tool_choice={"type": "auto"}, messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            text = "".join(b.text for b in response.content if b.type == "text")
            return {"status": "prepared", "summary": text, "pending_approvals": pending}

        results: list[ToolResultBlockParam] = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            call = ToolCall(tool=block.name, args=dict(block.input), **ctx)
            decision = gate.check(call)
            if decision.verdict is Verdict.ALLOW:
                output = EXECUTORS[call.tool](**call.args)
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(output)})
            elif decision.verdict is Verdict.NEEDS_APPROVAL:
                pending.append({"call": call.model_dump(), "call_digest": decision.call_digest, "reason": decision.reason})
                results.append({
                    "type": "tool_result", "tool_use_id": block.id, "is_error": True,
                    "content": f"Not executed. Queued for a person to approve (ref {decision.call_digest[:12]}). "
                               "Do not retry or rephrase this call. Summarise what you prepared.",
                })
            else:
                results.append({"type": "tool_result", "tool_use_id": block.id, "is_error": True,
                                "content": f"Denied by policy: {decision.reason}"})
        messages.append({"role": "user", "content": results})
    return {"status": "turn_budget_exhausted", "pending_approvals": pending}


def execute_approved(gate: ApprovalGate, call: ToolCall, approval: Approval) -> dict[str, Any]:
    """Called by the approval service once a person signs. The gate checks again, then the tool runs."""
    decision = gate.check(call, approval)
    if decision.verdict is not Verdict.ALLOW:
        raise PermissionError(decision.reason)
    return EXECUTORS[call.tool](**call.args)