| """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"]) |
|
|
| |
| _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)) |
| |
| 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} |
|
|
