File size: 3,347 Bytes
e411199 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """API surface for the Secure Agent Action Profile demo (PR B §18).
Clearly labelled demo: local keys, mock agents, mock provider, no real-money authority.
Exposes the machine-readable conformance report, the actor registry, and a labelled
end-to-end agent-payment demo flow that reaches money movement ONLY through the
deterministic PDP + single-use ExecutionCapability consumed by the Payment Core.
"""
from __future__ import annotations
import time
from fastapi import APIRouter, Body, HTTPException
from amanpay.agent_security.conformance import run as run_conformance
from amanpay.agent_security.orchestrator import AgentPaymentOrchestrator
from amanpay.agent_security.registry import REGISTRY
router = APIRouter(prefix="/agent-security", tags=["agent-security"])
# A single process-local demo orchestrator (labelled; not production).
_orch = AgentPaymentOrchestrator()
_LABEL = ("DEMO ONLY — local identity issuer (not production SPIFFE/SPIRE), mock agents, "
"mock provider, no real-money authority. Protocol invariants are still enforced.")
@router.get("/info")
def info() -> dict:
return {"profile": "amanpay-secure-agent-action-profile", "version": "0.1.0",
"label": _LABEL, "authoritative": ["deterministic-pdp", "payment-core"],
"interop_only": ["mcp", "a2a"]}
@router.get("/actors")
def actors() -> dict:
return {"actors": [{"spiffe_id": s.spiffe_id, "trust_boundary": s.trust_boundary,
"operations": sorted(s.operations),
"can_execute_payment": s.can_execute_payment,
"reasoning_agent": s.reasoning_agent,
"max_authority": s.max_authority}
for s in REGISTRY.values()]}
@router.get("/conformance")
def conformance() -> dict:
"""Run the conformance + attack suite and return the machine-readable report."""
return run_conformance(commit="api", generated_at=0)
@router.post("/demo/payee")
def register_payee(body: dict = Body(...)) -> dict:
try:
p = _orch.payees.register(body["iban"], country=body.get("country", "SA"))
except (KeyError, ValueError) as exc:
raise HTTPException(status_code=422, detail=str(exc))
# never returns the raw IBAN — only the opaque ref + masked display + binding hash
return {"payee_ref": p.payee_ref, "display": p.display, "binding_hash": p.binding_hash,
"label": _LABEL}
@router.post("/demo/flow")
def demo_flow(body: dict = Body(...)) -> dict:
"""Run a labelled end-to-end agent-payment flow through the profile."""
try:
r = _orch.run(
user_id=body.get("user_id", "demo"), payee_ref=body["payee_ref"],
amount_minor=int(body["amount_minor"]), country=body.get("country", "SA"),
currency=body.get("currency", "SAR"), merchant=body.get("merchant", "m1"),
reference=body.get("reference", "invoice"),
approve=bool(body.get("approve", False)), now=float(body.get("now", time.time())))
except (KeyError, ValueError) as exc:
raise HTTPException(status_code=422, detail=str(exc))
return {"decision": r.decision, "reason_codes": r.reason_codes,
"required_auth": r.required_auth, "agent_state": r.state,
"payment": r.payment, "audit": r.audit, "labels": r.labels}
|