diff --git a/examples/api_provider.py b/examples/api_provider.py index d9a3862..61c6430 100644 --- a/examples/api_provider.py +++ b/examples/api_provider.py @@ -11,12 +11,13 @@ The 402 lists all rails neutrally; the agent picks based on what their wallet supports. -Python merchants on Solana implement MPP `solana/charge` server-side themselves; -there is no `@solana/mpp` Python equivalent today. This example only advertises the -Solana rail in the 402 directives; settle the credential via your facilitator API. +`Checkout(...)` collapses the ~150 lines of hand-rolled 402 envelope + header +parsing + rail dispatch in pre-2.0 examples to a single `compute_pricing` + +`on_settled` configuration. Discovery probes are still handled inline because +they advertise SAMPLE rails for crawlers (not the merchant's real rails). Peer deps: - pip install agentscore-commerce[fastapi] + pip install 'agentscore-commerce[fastapi,x402,mppx]' Env vars: TEMPO_RECIPIENT your Tempo wallet for receiving USDC.e @@ -26,17 +27,22 @@ override to eip155:84532 for Sepolia testnet) SOLANA_NETWORK_CAIP2 CAIP-2 (default solana mainnet; override to solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1 for devnet) + MPP_SECRET_KEY secret_key for create_mppx_server (auto-wired) + CDP_API_KEY_ID Coinbase CDP key id; when set, the x402 facilitator + auto-promotes from public x402.org to Coinbase + CDP_API_KEY_SECRET Coinbase CDP key secret Run: uvicorn examples.api_provider:app --port 3000 """ import json import os -from base64 import b64encode +from typing import Any from fastapi import FastAPI, Request from fastapi.responses import JSONResponse +from agentscore_commerce import Checkout, PricingResult, SettleOutcome from agentscore_commerce.discovery import ( NoindexNonDiscoveryMiddleware, X402SampleProbe, @@ -44,113 +50,87 @@ is_discovery_probe_request, ) from agentscore_commerce.payment import ( - USDC, + SolanaMppRailSpec, + TempoRailSpec, + X402BaseRailSpec, networks, - payment_directive, - www_authenticate_header, ) PRICE_USDC = 0.01 # per-call price in USD REALM = "api.example.com" -# Read network selection from env so the same example serves mainnet + testnet. X402_BASE_NETWORK = os.environ.get("X402_BASE_NETWORK", networks.base.mainnet.caip2) SOLANA_NETWORK_CAIP2 = os.environ.get("SOLANA_NETWORK_CAIP2", networks.solana.mainnet.caip2) -_BASE_USDC = ( - USDC.base.sepolia.address if networks.base.sepolia.caip2 == X402_BASE_NETWORK else USDC.base.mainnet.address -) -_TEMPO_RAIL = "tempo-testnet" if networks.base.sepolia.caip2 == X402_BASE_NETWORK else "tempo-mainnet" +_TEMPO_RAIL_NAME = "tempo-testnet" if networks.base.sepolia.caip2 == X402_BASE_NETWORK else "tempo-mainnet" app = FastAPI() # noindex non-discovery paths so /search doesn't end up in human-shaped SERPs. -# Defaults cover /openapi.json, /llms.txt, /.well-known/{mpp.json,agent-card.json,ucp}, -# /favicon.{png,ico} — pass `custom_paths={"/sitemap.xml"}` to extend or -# `replace_paths=True` to swap the set entirely. app.add_middleware(NoindexNonDiscoveryMiddleware) -@app.post("/search") -async def search(request: Request): - body = await request.body() - body_text = body.decode() if body else "" +async def _run_your_search(_query: str) -> list[Any]: + """Vendor's actual search implementation.""" + return [] + + +async def _compute_pricing(_ctx: Any) -> PricingResult: + return PricingResult(amount_usd=PRICE_USDC) + + +async def _on_settled(ctx: Any, _outcome: SettleOutcome) -> dict[str, Any]: + body = ctx.request.body if isinstance(ctx.request.body, dict) else {} + results = await _run_your_search(body.get("query", "")) + return {"results": results} + + +checkout = Checkout( + rails={ + # Static treasury recipients; fail-fast at import time on missing env so + # misconfigured deploys never reach the 402 emit path with empty rails. + "tempo": TempoRailSpec( + recipient=os.environ["TEMPO_RECIPIENT"], + testnet=networks.base.sepolia.caip2 == X402_BASE_NETWORK, + ), + "x402_base": X402BaseRailSpec( + recipient=os.environ["X402_BASE_RECIPIENT"], + network=X402_BASE_NETWORK, + ), + "solana": SolanaMppRailSpec( + recipient=os.environ["SOLANA_RECIPIENT"], + network=SOLANA_NETWORK_CAIP2, + ), + }, + url=f"https://{REALM}/search", + compute_pricing=_compute_pricing, + on_settled=_on_settled, + cdp_api_key_id=os.environ.get("CDP_API_KEY_ID"), + cdp_api_key_secret=os.environ.get("CDP_API_KEY_SECRET"), + mppx_secret_key=os.environ.get("MPP_SECRET_KEY"), +) + +@app.post("/search") +async def search(request: Request) -> JSONResponse: + body_bytes = await request.body() + body_text = body_bytes.decode() if body_bytes else "" auth = request.headers.get("authorization") - x402_header = request.headers.get("payment-signature") or request.headers.get("x-payment") - # Discovery probe — empty-body POST without any payment header → return sample 402. + # Discovery probe: empty-body POST without any payment header. Return sample + # 402 so crawlers (`awal x402 details`, x402-proxy, ...) can find this surface + # without committing to a real charge. Handle inline because the probe + # advertises SAMPLE accepts (not the merchant's real settle rails). if await is_discovery_probe_request(request.method, auth, body_text): probe = build_discovery_probe_response( realm=REALM, - sample_rail=_TEMPO_RAIL, + sample_rail=_TEMPO_RAIL_NAME, sample_amount_usd=PRICE_USDC, sample_recipient=os.environ["TEMPO_RECIPIENT"], - # Advertise x402 support so crawlers (e.g. ``awal x402 details``) - # can find it on an empty-body POST. Commerce synthesizes USDC - # sample accepts from the registry per CAIP-2 network passed. x402_sample=X402SampleProbe( networks=[X402_BASE_NETWORK, SOLANA_NETWORK_CAIP2], - resource_url=f"{REALM}/search", + resource_url=f"https://{REALM}/search", ), ) return JSONResponse(json.loads(probe.body), status_code=probe.status, headers=probe.headers) - # No payment? Return a 402 with directives for all accepted rails. - if not (auth and auth.startswith("Payment ")) and not x402_header: - challenge_id = f"chg_{os.urandom(8).hex()}" - x402_base_rail = ( - "x402-base-sepolia" if networks.base.sepolia.caip2 == X402_BASE_NETWORK else "x402-base-mainnet" - ) - solana_mpp_rail = ( - "mpp-solana-devnet" if networks.solana.devnet.caip2 == SOLANA_NETWORK_CAIP2 else "mpp-solana-mainnet" - ) - directives = [ - payment_directive(rail=_TEMPO_RAIL, id=f"{challenge_id}_tempo", realm=REALM, request=""), - payment_directive(rail=x402_base_rail, id=f"{challenge_id}_base", realm=REALM, request=""), - payment_directive(rail=solana_mpp_rail, id=f"{challenge_id}_solana", realm=REALM, request=""), - ] - accepts = [ - { - "scheme": "exact", - "network": X402_BASE_NETWORK, - "amount": str(int(PRICE_USDC * 1_000_000)), - "asset": _BASE_USDC, - "payTo": os.environ["X402_BASE_RECIPIENT"], - "maxTimeoutSeconds": 300, - # EIP-712 domain required by every x402 EVM client to sign - # EIP-3009 TransferWithAuthorization. ``name`` MUST match the - # on-chain USDC contract's ``name()`` — base mainnet returns - # "USD Coin", base sepolia returns "USDC". Wrong value silently - # breaks signature verify at the facilitator. Production code - # should use ``build_x402_accepts_for_402(server, ...)`` which - # derives ``extra`` from the registered scheme metadata. - "extra": { - "name": "USD Coin" if X402_BASE_NETWORK.split(":")[-1] == "8453" else "USDC", - "version": "2", - }, - }, - ] - return JSONResponse( - {"payment_required": True, "x402Version": 2, "accepts": accepts}, - status_code=402, - headers={ - "www-authenticate": www_authenticate_header(directives), - "PAYMENT-REQUIRED": b64encode( - json.dumps({"x402Version": 2, "accepts": accepts, "resource": {"url": str(request.url)}}).encode() - ).decode(), - }, - ) - - # Payment present; branch on which header arrived: - # Authorization: Payment ... → MPP (tempo or solana); validate via your facilitator's MPP API - # payment-signature / x-payment → x402 base; validate via x402 facilitator - # Both shapes settle through the configured facilitator HTTP API, then run your operation. - - body_json = json.loads(body_text) - results = await run_your_search(body_json.get("query", "")) - return {"results": results} - - -async def run_your_search(_query: str) -> list: - # Vendor's actual search implementation - return [] + return await checkout.handle_fastapi(request) diff --git a/examples/compliance_merchant.py b/examples/compliance_merchant.py index 49d6eb2..0c6badd 100644 --- a/examples/compliance_merchant.py +++ b/examples/compliance_merchant.py @@ -1,27 +1,31 @@ -"""Example: regulated-goods merchant showcasing the gate + denial helpers +"""Example: regulated-goods merchant showcasing the gate + denial helpers. -Scenario: you sell something that needs identity gating — wine (age 21+, US-only), cannabis -(age 21+, state allowlist), high-value items (KYC + sanctions). The agent needs to know how -to recover from each kind of denial. +Scenario: you sell something that needs identity gating; wine (age 21+, US-only), +cannabis (age 21+, state allowlist), high-value items (KYC + sanctions). The +agent needs to know how to recover from each kind of denial. What this example demonstrates: - - AgentScoreGate with full compliance policy (KYC + sanctions + age + jurisdiction) - - Custom on_denied composing commerce helpers: - * verification_agent_instructions for the canonical poll-and-retry instructions - * is_fixable_denial defensive fallback for fixable (KYC re-do) vs unfixable - (sanctions / age / jurisdiction_restricted) compliance fails. Gate normally - re-routes fixable reasons to identity_verification_required upstream — this - branch only fires if the /v1/sessions mint blipped. - * build_contact_support_next_steps for the unfixable branch - * denial_reason_to_body + denial_reason_status for the standard fall-through - (token_expired, invalid_credential, api_error get the right status + body for free) - - get_signer_verdict (cached signer_match read) + build_signer_mismatch_body for wallet-auth verification - -The pattern: vendors only write the BUSINESS-SPECIFIC denial branches. Everything else is a -one-line helper call. + +* `Checkout(gate=CheckoutGateConfig(...))` runs the SDK gate on the settle leg. +* Custom `on_denied` callback composes the canonical denial helpers: + - `verification_agent_instructions` for the canonical poll-and-retry block + - `is_fixable_denial` for fixable (KYC re-do) vs unfixable + (sanctions / age / jurisdiction_restricted) compliance fails. Gate normally + re-routes fixable reasons to identity_verification_required upstream; + the fixable branch is a defensive fallback if /v1/sessions mint blipped. + - `build_contact_support_next_steps` for the unfixable branch + - `denial_reason_to_body` + `denial_reason_status` for the standard + fall-through (token_expired, invalid_credential, api_error get the + right status + body for free). +* Signer-match enforcement (wallet_signer_mismatch / wallet_auth_requires_wallet_signing) + is now automatic inside the gate; consumers don't call + `build_signer_mismatch_body` from inside the handler anymore. + +Pattern: vendors only write the BUSINESS-SPECIFIC denial branches. Everything +else is a one-line helper call. Peer deps: - pip install agentscore-commerce[fastapi] + pip install 'agentscore-commerce[fastapi]' Env vars: AGENTSCORE_API_KEY — your AgentScore API key @@ -32,31 +36,34 @@ import os from typing import Any -from fastapi import Depends, FastAPI, Request +from fastapi import FastAPI, Request from fastapi.responses import JSONResponse -from agentscore_commerce.identity import ( +from agentscore_commerce import ( + Checkout, + CheckoutGateConfig, DenialReason, + PricingResult, + SettleOutcome, build_contact_support_next_steps, - build_signer_mismatch_body, denial_reason_status, denial_reason_to_body, is_fixable_denial, verification_agent_instructions, ) -from agentscore_commerce.identity.fastapi import AgentScoreGate, get_agentscore_data, get_signer_verdict +from agentscore_commerce.payment import TempoRailSpec SUPPORT_EMAIL = "support@example.com" -# Vendor-specific extension of the canonical agent_instructions block. The commerce default -# covers steps 1-4 (present verify_url, poll, user verifies, extract token) plus a generic -# "retry the original merchant request" at step 5. ``retry_step`` REPLACES that generic step 5 -# with our merchant-specific retry (include order_id to resume the pending order). ``extra_steps`` -# adds the genuinely-additional 402-payment step that comes AFTER retry. +# Vendor-specific extension of the canonical agent_instructions block. The commerce +# default covers steps 1-4 (present verify_url, poll, user verifies, extract token) +# plus a generic "retry the original merchant request" at step 5. `retry_step` +# REPLACES that generic step 5; `extra_steps` adds the 402-payment step that +# comes AFTER retry. VERIFICATION_INSTRUCTIONS = verification_agent_instructions( retry_step=( - "Retry the request with header X-Operator-Token set to the operator_token value AND include the " - "order_id from this 403 in the body to resume the pending order." + "Retry the request with header X-Operator-Token set to the operator_token value AND " + "include the order_id from this 403 in the body to resume the pending order." ), extra_steps=[ "The retry returns 402 Payment Required with a payment challenge. Pay via tempo request or agentscore-pay pay.", @@ -65,84 +72,91 @@ ) -def _on_denied(_request: Request, reason: DenialReason) -> tuple[dict[str, Any], int]: - # missing_identity → bare 403 (no auto-session created — agent must bootstrap). +async def _on_denied(_ctx: Any, reason: DenialReason) -> dict[str, Any] | None: + """Reshape the canonical denial body for vendor-specific copy. + + Return `{"status": , "body": , "headers": ?}` to override + the gate's default envelope, or `None` to keep the canonical body. + """ + # missing_identity → bare 403; agent must bootstrap. if reason.code == "missing_identity": body = denial_reason_to_body(reason) body["error"] = {"code": "identity_required", "message": "Identity verification is required for this purchase."} - return body, 403 + return {"status": 403, "body": body} - # identity_verification_required → gate auto-minted a session. Overlay vendor-specific - # agent_instructions on top of the commerce body. + # identity_verification_required → gate auto-minted a session. Overlay + # vendor-specific agent_instructions on top of the commerce body. if reason.code == "identity_verification_required": body = denial_reason_to_body(reason) body["agent_instructions"] = VERIFICATION_INSTRUCTIONS - return body, 403 + return {"status": 403, "body": body} - # wallet_not_trusted = UNFIXABLE compliance fail (sanctions / age / jurisdiction_restricted). - # The gate auto-routes fixable reasons (kyc_required / kyc_pending / kyc_failed) to - # identity_verification_required upstream — by the time on_denied sees wallet_not_trusted, - # the reasons should be unfixable. The is_fixable_denial branch below is a defensive - # fallback in case the gate's /v1/sessions mint blipped and fell back to bare denial. + # wallet_not_trusted = UNFIXABLE compliance fail (sanctions / age / + # jurisdiction_restricted). The gate auto-routes fixable reasons upstream; + # the is_fixable_denial branch here is a defensive fallback. if reason.code == "wallet_not_trusted": reasons = reason.reasons or [] if is_fixable_denial(reasons): - # Defensive: gate normally bootstraps these into identity_verification_required. - # If we hit this branch, the gate's /v1/sessions mint failed — surface verify_url - # so the agent can recover via the manual session flow. return { - "error": {"code": "compliance_recoverable", "message": "Re-verify identity and retry."}, - "reasons": reasons, - "verify_url": reason.verify_url, - }, 403 + "status": 403, + "body": { + "error": {"code": "compliance_recoverable", "message": "Re-verify identity and retry."}, + "reasons": reasons, + "verify_url": reason.verify_url, + }, + } return { - "error": { - "code": "compliance_denied", - "message": "Purchase denied by compliance policy. Not resolvable through re-verification.", + "status": 403, + "body": { + "error": { + "code": "compliance_denied", + "message": "Purchase denied by compliance policy. Not resolvable through re-verification.", + }, + "reasons": reasons, + "next_steps": build_contact_support_next_steps(SUPPORT_EMAIL), }, - "reasons": reasons, - "next_steps": build_contact_support_next_steps(SUPPORT_EMAIL), - }, 403 - - # token_expired (401), invalid_credential (401), api_error (503) → standard body+status from commerce. - return denial_reason_to_body(reason), denial_reason_status(reason) + } + + # token_expired (401), invalid_credential (401), api_error (503) → + # standard body+status from commerce. + return {"status": denial_reason_status(reason), "body": denial_reason_to_body(reason)} + + +async def _compute_pricing(_ctx: Any) -> PricingResult: + return PricingResult(amount_usd=250.0) # vendor pricing logic goes here. + + +async def _on_settled(ctx: Any, outcome: SettleOutcome) -> dict[str, Any]: + return { + "ok": True, + "reference_id": ctx.reference_id, + "tx_hash": outcome.tx_hash, + "identity_status": ctx.identity_status, + } + + +checkout = Checkout( + # Minimal rails so the 402 emit path has something to advertise; vendor + # swaps in their real rails (multi-rail, Stripe-anchored, etc.). + rails={"tempo": TempoRailSpec(recipient=os.environ.get("TEMPO_RECIPIENT", "0xfeedface"))}, + url="https://api.example.com/buy", + compute_pricing=_compute_pricing, + on_settled=_on_settled, + gate=CheckoutGateConfig( + api_key=os.environ["AGENTSCORE_API_KEY"], + merchant_name="Compliance Demo", + require_kyc=True, + require_sanctions_clear=True, + min_age=21, + allowed_jurisdictions=["US"], + on_denied=_on_denied, + ), +) app = FastAPI() -_gate = AgentScoreGate( - api_key=os.environ["AGENTSCORE_API_KEY"], - require_kyc=True, - require_sanctions_clear=True, - min_age=21, - allowed_jurisdictions=["US"], - on_denied=_on_denied, -) -# Conditional gate. Fires only when a payment credential is already attached so -# anonymous discovery returns a 402 challenge (not a 403 missing_identity). Compliance -# gating + signer-match still run on the retry leg when X-Payment / Authorization: -# Payment arrives — the full denial branching above triggers there. -async def gate_on_settle(request: Request) -> None: - has_payment_header = bool( - request.headers.get("payment-signature") - or request.headers.get("x-payment") - or (request.headers.get("authorization") or "").startswith("Payment ") - ) - if not has_payment_header: - return None - return await _gate(request) - - -@app.post("/buy", dependencies=[Depends(gate_on_settle)]) -async def buy(request: Request, assess: dict = Depends(get_agentscore_data)): - # Wallet-auth: read the cached signer_match verdict the gate composed on its - # primary /v1/assess call (single round trip). Returns None on operator_token paths. - verdict = get_signer_verdict(request) - if verdict is not None and verdict.signer_match is not None: - mismatch_body = build_signer_mismatch_body(verdict.signer_match) - if mismatch_body: - return JSONResponse(mismatch_body, status_code=403) - - # Compliance + signer-match passed. Run the actual purchase. - return {"ok": True, "identity_method": assess.get("identity_method")} +@app.post("/buy") +async def buy(request: Request) -> JSONResponse: + return await checkout.handle_fastapi(request) diff --git a/examples/multi_rail_merchant.py b/examples/multi_rail_merchant.py index 58b2892..cc35cb4 100644 --- a/examples/multi_rail_merchant.py +++ b/examples/multi_rail_merchant.py @@ -1,65 +1,64 @@ """Example: full regulated-commerce merchant. -Scenario: you sell a regulated good. Identity gate (KYC + age + jurisdiction + sanctions), -plus 402 payment challenge advertising multiple rails so agents can pay with whatever they -have: Tempo USDC (MPP `tempo/charge`), x402 USDC on Base, Solana USDC (MPP `solana/charge`), -Stripe SPT. - -The flow on each /purchase POST: - 1. Identity gate (AgentScoreGate): KYC + age + jurisdiction + sanctions - 2. If ``X-Payment`` header present (x402 client paying base) → ``verify_x402_request`` → - ``process_x402_settle`` → return 200 with ``payment-response`` header - 3. Else mint a Stripe multichain PI (deposit addresses for tempo/base/solana) - and run pympp's compose() to validate any ``Authorization: Payment`` header - (covers tempo/charge AND solana/charge directives) - 4. If pympp returns 402 → ``respond_402`` (preserves pympp's WWW-Auth + adds x402's - PAYMENT-REQUIRED) with the rich body - 5. If pympp returns 200 → also fire ``simulate_deposit_if_test_mode`` for testnet +Scenario: you sell a regulated good. Identity gate (KYC + age + jurisdiction + +sanctions), plus a 402 payment challenge advertising multiple rails so agents +pay with whatever they have: Tempo USDC (MPP `tempo/charge`), x402 USDC on +Base, Solana USDC (MPP `solana/charge`), Stripe SPT. + +`Checkout(...)` orchestrates the flow: + +1. Identity gate runs only on the settle leg (a payment header is attached); + the discovery leg flows through anonymously and gets a 402 with all rails. +2. `mint_recipients` hook calls into Stripe to mint per-PI deposit addresses + for tempo/base/solana before the 402 emits, so the body advertises the + right addresses. +3. `compute_pricing` returns the subtotal + tax block for the current cart. +4. x402-base header → Checkout dispatches to `process_x402_settle` internally. +5. `Authorization: Payment` header → Checkout dispatches to the auto-derived + `compose_mppx` hook (built from `mppx_secret_key`). +6. `on_settled` persists the order + fires `simulate_deposit_if_test_mode` + for Stripe testnet round-trip on base settles. Peer deps:: - pip install agentscore-commerce[fastapi,x402,pympp] + pip install 'agentscore-commerce[fastapi,x402,mppx,coinbase,stripe]' Env vars: - AGENTSCORE_API_KEY — your AgentScore API key - APP_URL — public URL of your service - STRIPE_SECRET_KEY — sk_test_... or sk_live_... - STRIPE_PROFILE_ID — your Stripe Connect profile id (for SPT) - TEMPO_USDC_ADDRESS — USDC token address on Tempo (mainnet or testnet) - X402_BASE_NETWORK — CAIP-2 - SOLANA_NETWORK_CAIP2 — CAIP-2 - REDIS_URL — optional; in-memory PI cache otherwise + AGENTSCORE_API_KEY your AgentScore API key + APP_URL public URL of your service + STRIPE_SECRET_KEY sk_test_... or sk_live_... + STRIPE_PROFILE_ID your Stripe Connect profile id (for SPT) + X402_BASE_NETWORK CAIP-2 (default eip155:8453) + SOLANA_NETWORK_CAIP2 CAIP-2 (default solana mainnet) + MPP_SECRET_KEY secret_key for the auto-derived mppx server + CDP_API_KEY_ID Coinbase CDP key id (auto-promotes x402 facilitator) + CDP_API_KEY_SECRET Coinbase CDP key secret + REDIS_URL optional; in-memory PI cache otherwise Run: uvicorn examples.multi_rail_merchant:app --port 3000 """ import os +from typing import Any -from fastapi import Depends, FastAPI, Request +from fastapi import FastAPI, Request from fastapi.responses import JSONResponse -from agentscore_commerce.challenge import ( - build_402_body, - build_accepted_methods, - build_agent_instructions, - build_how_to_pay, - build_pricing_block, - build_validation_error, - first_encounter_agent_memory, - respond_402, +from agentscore_commerce import ( + Checkout, + CheckoutGateConfig, + CheckoutValidationError, + PricingResult, + SettleOutcome, ) -from agentscore_commerce.identity.fastapi import AgentScoreGate, get_agentscore_data +from agentscore_commerce.challenge import build_pricing_block from agentscore_commerce.payment import ( - USDC, SolanaMppRailSpec, StripeRailSpec, TempoRailSpec, X402BaseRailSpec, - build_x402_accepts_for_402, networks, - process_x402_settle, validate_x402_network_config, - verify_x402_request, ) from agentscore_commerce.stripe_multichain import ( create_pi_cache, @@ -67,193 +66,107 @@ ) APP_URL = os.environ["APP_URL"] +STRIPE_SECRET_KEY = os.environ["STRIPE_SECRET_KEY"] X402_BASE_NETWORK = os.environ.get("X402_BASE_NETWORK", networks.base.mainnet.caip2) SOLANA_NETWORK_CAIP2 = os.environ.get("SOLANA_NETWORK_CAIP2", networks.solana.mainnet.caip2) - -# Boot-time guard: validate the configured x402 networks are in the supported set. -# Raises on misconfigured deploys before the first request. validate_x402_network_config(base_network=X402_BASE_NETWORK) -# Singleton Stripe PI / deposit-address cache. Backed by Redis when REDIS_URL is set -# (multi-instance deployments need this so a deposit lands on whichever instance -# settles it); falls back to in-process dict for single-instance dev. pi_cache = create_pi_cache(redis_url=os.environ.get("REDIS_URL")) app = FastAPI() -_gate = AgentScoreGate( - api_key=os.environ["AGENTSCORE_API_KEY"], - require_kyc=True, - require_sanctions_clear=True, - min_age=21, - allowed_jurisdictions=["US"], -) -# Conditional gate: fires only when a payment credential is already attached. Anonymous -# requests (no payment header) fall through to the handler unauthenticated and receive -# a clean 402 with all rails advertised — so any spec-compliant x402 wallet (Coinbase -# awal, Phantom, Solflare, etc.) can discover prices before AgentScore identity exists. -# Identity is verified at settle time (when X-Payment / Authorization: Payment arrives), -# and `create_session_on_missing` then auto-mints a verification session. -async def gate_on_settle(request: Request) -> None: - has_payment_header = bool( - request.headers.get("payment-signature") - or request.headers.get("x-payment") - or (request.headers.get("authorization") or "").startswith("Payment ") - ) - if not has_payment_header: - return None - return await _gate(request) +async def _create_multichain_payment_intent(_total_usd: str) -> dict[str, str]: + """Vendor's actual Stripe multichain PI mint call. + Returns deposit addresses for {tempo, base, solana}. In production this + calls `stripe.PaymentIntent.create(...)` with `payment_method_types` set + + reads back the per-network deposit addresses Stripe minted. + """ + return {"tempo": "0x...", "base": "0x...", "solana": "..."} -# Vendor-instantiated x402 server + pympp server are stubs in this example — -# replace with your `create_x402_server(...)` + `create_mppx_server(...)` setup. -x402_server: object = ... # type: ignore[assignment] +async def _validate_purchase(ctx: Any) -> dict[str, Any]: + """preValidate hook: shape-check the request body before pricing/gate runs.""" + body = ctx.request.body if isinstance(ctx.request.body, dict) else {} + if "shipping" not in body: + raise CheckoutValidationError(code="missing_shipping", message="`shipping` is required.") + return {"shipping_state": body["shipping"].get("state", "CA")} -@app.post("/purchase", dependencies=[Depends(gate_on_settle)]) -async def purchase(request: Request, assess: dict = Depends(get_agentscore_data)): - body = await request.json() - # Compute pricing (vendor-specific — wine tax by state, dynamic SKU pricing, etc.) - subtotal_cents = 25000 # $250.00 +async def _compute_pricing(ctx: Any) -> PricingResult: + subtotal_cents = 25000 # $250.00; vendor pricing logic goes here. tax_cents = 2000 total_cents = subtotal_cents + tax_cents - total_usd = f"{total_cents / 100:.2f}" pricing = build_pricing_block( subtotal_cents=subtotal_cents, tax_cents=tax_cents, tax_rate=0.08, - tax_state=body.get("shipping", {}).get("state", "CA"), + tax_state=ctx.state.get("shipping_state", "CA"), currency="USD", ) + return PricingResult(amount_usd=total_cents / 100, body_extras={"pricing": pricing}) - # ────────────────────────────────────────────────────────────────────────── - # Path A: x402 X-Payment header present → verify + settle on chain - # ────────────────────────────────────────────────────────────────────────── - if request.headers.get("payment-signature") or request.headers.get("x-payment"): - verified = await verify_x402_request( - headers=dict(request.headers), - is_cached_address=pi_cache.has_address, - accepted_network=X402_BASE_NETWORK, - ) - if not verified.ok: - return JSONResponse(verified.body, status_code=verified.status) - settle = await process_x402_settle( - x402_server=x402_server, - payload=verified.payload, - resource_config={ - "scheme": "exact", - "network": verified.signed_network, - "price": f"${total_usd}", - "payTo": verified.signed_pay_to, - "maxTimeoutSeconds": 300, - }, - resource_meta={ - "url": str(request.url), - "description": "Agent purchase via x402", - "mimeType": "application/json", - }, - ) - if not settle.success: - return JSONResponse( - build_validation_error( - code="payment_proof_invalid", - message=f"Payment failed during settlement (phase: {settle.phase or 'unknown'}).", - next_steps={"action": "regenerate_payment_credential"}, - extra={"phase": settle.phase}, - ), - status_code=400, - ) +async def _mint_recipients(ctx: Any) -> dict[str, str]: + """Per-order recipient mint: Stripe multichain PI → per-network deposit addresses.""" + total_usd = f"{ctx.pricing.amount_usd:.2f}" + addresses = await _create_multichain_payment_intent(total_usd) + return { + "tempo": addresses["tempo"], + "x402_base": addresses["base"], + "solana_mpp": addresses["solana"], + } + - # Fire Stripe testnet sim; no-ops on live keys. x402 settle only ever - # lands on base in 1.4+ (Solana moved to MPP `solana/charge`). +async def _on_settled(ctx: Any, outcome: SettleOutcome) -> dict[str, Any]: + # Fire Stripe testnet deposit simulation on real on-chain base settles + # (no-op on live keys). Gate on `tx_hash` so $0 zero-settle carve-outs + # (which have signer_address but no tx_hash) don't trigger a PI sim. + if outcome.rail == "x402" and outcome.tx_hash is not None: await simulate_deposit_if_test_mode( get_payment_intent_id=pi_cache.get_payment_intent_id, - deposit_address=verified.signed_pay_to, + deposit_address=ctx.recipients.get("x402_base", ""), network="base", - stripe_secret_key=os.environ["STRIPE_SECRET_KEY"], + stripe_secret_key=STRIPE_SECRET_KEY, ) + return { + "ok": True, + "reference_id": ctx.reference_id, + "tx_hash": outcome.tx_hash, + "identity_status": ctx.identity_status, + } - headers: dict[str, str] = {} - if settle.payment_response_header: - headers["payment-response"] = settle.payment_response_header - return JSONResponse({"ok": True, "operator": assess.get("resolved_operator")}, headers=headers) - # ────────────────────────────────────────────────────────────────────────── - # Path B: cold call OR Authorization: Payment (pympp) — mint PI + compose pympp - # ────────────────────────────────────────────────────────────────────────── - # ... your createMultichainPaymentIntent + cache writeback here ... - # ... your pympp.compose() to validate Authorization: Payment header ... - # If pympp returns 402, build the rich 402 with respond_402 (preserves pympp's - # WWW-Auth + adds x402's PAYMENT-REQUIRED): - pympx_challenge_headers = {"www-authenticate": 'Payment id="..."'} # from pympp.compose - deposit_addresses = {"tempo": "0x...", "base": "0x...", "solana": "..."} # from create_multichain_payment_intent - # Declare every rail once — every helper consumes the same RailSpec instances. - rails = { - "tempo": TempoRailSpec(recipient=deposit_addresses["tempo"]), - "x402_base": X402BaseRailSpec(recipient=deposit_addresses["base"]), - "solana_mpp": SolanaMppRailSpec(recipient=deposit_addresses["solana"]), +checkout = Checkout( + rails={ + # Per-order-mint pattern: empty-string `recipient` declares the rail + # in discovery; `mint_recipients` resolves the real per-PI address. + "tempo": TempoRailSpec(recipient=""), + "x402_base": X402BaseRailSpec(recipient="", network=X402_BASE_NETWORK), + "solana_mpp": SolanaMppRailSpec(recipient="", network=SOLANA_NETWORK_CAIP2), "stripe": StripeRailSpec(profile_id=os.environ["STRIPE_PROFILE_ID"]), - } - accepted = await build_accepted_methods( - tempo=rails["tempo"], - x402_base=rails["x402_base"], - solana_mpp=rails["solana_mpp"], - stripe=rails["stripe"], - ) - how_to_pay = await build_how_to_pay( - url=APP_URL, - retry_body_json=str(body), - total_usd=total_usd, - rails=rails, - ) + }, + url=f"{APP_URL}/purchase", + pre_validate=_validate_purchase, + compute_pricing=_compute_pricing, + mint_recipients=_mint_recipients, + on_settled=_on_settled, + is_cached_address=pi_cache.has_address, + cdp_api_key_id=os.environ.get("CDP_API_KEY_ID"), + cdp_api_key_secret=os.environ.get("CDP_API_KEY_SECRET"), + mppx_secret_key=os.environ.get("MPP_SECRET_KEY"), + gate=CheckoutGateConfig( + api_key=os.environ["AGENTSCORE_API_KEY"], + merchant_name="Regulated Goods Co.", + require_kyc=True, + require_sanctions_clear=True, + min_age=21, + allowed_jurisdictions=["US"], + ), +) - result = respond_402( - mppx_challenge_headers=pympx_challenge_headers, - body=build_402_body( - accepted_methods=accepted, - agent_instructions=build_agent_instructions(how_to_pay=how_to_pay), - pricing=pricing, - amount_usd=total_usd, - retry_body=body, - # Production merchants track first-encounter state in their own DB; - # for demo purposes we always emit the cross-merchant pattern hint. - agent_memory=first_encounter_agent_memory(first_encounter=True), - ), - x402={ - "x402_version": 2, - # Base accept comes from the registered x402 scheme — `extra` (incl. the - # network-correct USDC `name`) is filled in automatically. Solana goes - # through MPP `solana/charge` not x402's exact scheme, so it stays inline. - "accepts": [ - *build_x402_accepts_for_402( - x402_server, - network=X402_BASE_NETWORK, - price=f"${total_usd}", - pay_to=deposit_addresses["base"], - max_timeout_seconds=300, - ), - { - "scheme": "exact", - "network": SOLANA_NETWORK_CAIP2, - "amount": str(round(float(total_usd) * 1_000_000)), - "asset": ( - USDC.solana.devnet.mint - if networks.solana.devnet.caip2 == SOLANA_NETWORK_CAIP2 - else USDC.solana.mainnet.mint - ), - "payTo": deposit_addresses["solana"], - "maxTimeoutSeconds": 300, - # SVM transactions require feePayer in extra. Default to - # the recipient (round-trip safe for dev). Production - # merchants typically point at the Coinbase facilitator's - # payer address. - "extra": {"feePayer": deposit_addresses["solana"]}, - }, - ], - "resource": {"url": str(request.url), "mimeType": "application/json"}, - }, - ) - return JSONResponse(result.body, status_code=result.status, headers=result.headers) + +@app.post("/purchase") +async def purchase(request: Request) -> JSONResponse: + return await checkout.handle_fastapi(request) diff --git a/examples/per_product_policy_merchant.py b/examples/per_product_policy_merchant.py index c754d07..b9c044c 100644 --- a/examples/per_product_policy_merchant.py +++ b/examples/per_product_policy_merchant.py @@ -1,24 +1,26 @@ -"""Example: multi-product merchant with per-product compliance policy + soft mode +"""Example: multi-product merchant with per-product compliance policy + soft mode. Scenario: you sell several products with different compliance needs. -- Wine: hard gate, KYC + 21+ + US-only + state allowlist (regulated alcohol) -- Tee: no gate at all — fully anonymous, ship anywhere -- Limited print: SOFT gate — request KYC for fraud signals, but don't block sale - if the buyer skips it; record identity_status="unverified" instead - -Each product carries its own policy block (in this example, a Python dict the -merchant looks up from a database row). The route uses three helpers from -``agentscore_commerce.identity.policy``: - - - build_gate_from_policy(policy, *, api_key) → AgentScoreGate | None - Returns None when the policy has no enforcement (no gate fires). - - run_gate_with_enforcement(request, gate, *, enforcement) → GateResult - Runs the gate, swallows soft denials, returns a structured result. - - shipping_country_allowed / shipping_state_allowed - Per-product shipping allowlists (NULL = ship anywhere). + +* Wine: hard gate, KYC + 21+ + US-only + state allowlist (regulated alcohol) +* Tee: no gate at all; fully anonymous, ship anywhere +* Limited print: SOFT gate; request KYC for fraud signals, but don't block the + sale if the buyer skips it; record `identity_status="unverified"` instead. + +Each product carries its own `PolicyBlock`. `Checkout(gate=CheckoutGateConfig( +per_request_policy=...))` resolves it per request: + +1. `pre_validate` looks up the product row by slug and stashes the policy block + onto `ctx.state` for downstream hooks. +2. `per_request_policy(ctx)` returns the merged policy dict (including + `enforcement: "hard"|"soft"|None`) — the SDK gate runs hard/soft based on + the field. +3. Soft denials are swallowed by the SDK and stamp + `identity_status="unverified"` onto the order; hard denials propagate the + canonical 403 envelope. Peer deps: - pip install agentscore-commerce[fastapi] + pip install 'agentscore-commerce[fastapi]' Env vars: AGENTSCORE_API_KEY — your AgentScore API key @@ -32,18 +34,23 @@ from fastapi import FastAPI, Request from fastapi.responses import JSONResponse -from agentscore_commerce.identity.policy import ( +from agentscore_commerce import ( + Checkout, + CheckoutGateConfig, + CheckoutValidationError, PolicyBlock, - build_gate_from_policy, - run_gate_with_enforcement, + PricingResult, + SettleOutcome, +) +from agentscore_commerce.identity.policy import ( shipping_country_allowed, shipping_state_allowed, ) +from agentscore_commerce.payment import TempoRailSpec API_KEY = os.environ.get("AGENTSCORE_API_KEY", "ask_test_dummy") -# A merchant would normally read these from a `products` table. Each row carries -# its own compliance config; the keys match `PolicyBlock`. +# A merchant would normally read these from a `products` table. PRODUCTS: dict[str, dict[str, Any]] = { "wine-cabernet": { "name": "Reserve Cabernet", @@ -55,7 +62,7 @@ min_age=21, allowed_jurisdictions=["US"], allowed_shipping_countries=["US"], - allowed_shipping_states=["CA", "NY", "TX", "FL", "WA"], # abridged + allowed_shipping_states=["CA", "NY", "TX", "FL", "WA"], ), }, "tee": { @@ -73,53 +80,71 @@ } -app = FastAPI() - - -@app.post("/purchase") -async def purchase(request: Request) -> JSONResponse: - body = await request.json() +async def _validate_purchase(ctx: Any) -> dict[str, Any]: + body = ctx.request.body if isinstance(ctx.request.body, dict) else {} slug = body.get("product_slug") shipping = body.get("shipping", {}) - product = PRODUCTS.get(slug) + product = PRODUCTS.get(slug or "") if product is None: - return JSONResponse({"error": {"code": "product_not_found"}}, status_code=400) + raise CheckoutValidationError(code="product_not_found", message=f"No product with slug {slug!r}.") policy = product["policy"] - - # Per-product shipping allowlists. NULL policy → ship anywhere. if not shipping_country_allowed(shipping.get("country", ""), policy): - return JSONResponse( - {"error": {"code": "unsupported_jurisdiction", "message": f"Cannot ship to {shipping.get('country')}."}}, - status_code=400, + raise CheckoutValidationError( + code="unsupported_jurisdiction", + message=f"Cannot ship to {shipping.get('country')}.", ) if not shipping_state_allowed(shipping.get("state", ""), shipping.get("country", ""), policy): - return JSONResponse( - {"error": {"code": "unsupported_jurisdiction", "message": f"Cannot ship to {shipping.get('state')}."}}, - status_code=400, + raise CheckoutValidationError( + code="unsupported_jurisdiction", + message=f"Cannot ship to {shipping.get('state')}.", ) + return {"product": product, "policy": policy} + + +async def _compute_pricing(ctx: Any) -> PricingResult: + product = ctx.state["product"] + return PricingResult(amount_usd=float(product["price_usd"])) + + +def _per_request_policy(ctx: Any) -> dict[str, Any] | None: + policy = ctx.state.get("policy") + if policy is None: + return None # Skip the gate entirely for no-policy products (anonymous). + # The SDK gate reads `enforcement` to switch hard/soft mode. `PolicyBlock` + # already carries the field; spread it through verbatim. + return dict(policy) + + +async def _on_settled(ctx: Any, outcome: SettleOutcome) -> dict[str, Any]: + product = ctx.state["product"] + return { + "order": {"product": product["name"], "total_usd": product["price_usd"]}, + "identity_status": ctx.identity_status, + "tx_hash": outcome.tx_hash, + } + + +checkout = Checkout( + # Minimal rails so the 402 emit path has something to advertise; vendor + # swaps in their real rails (multi-rail, Stripe-anchored, etc.). + rails={"tempo": TempoRailSpec(recipient=os.environ.get("TEMPO_RECIPIENT", "0xfeedface"))}, + url="https://api.example.com/purchase", + pre_validate=_validate_purchase, + compute_pricing=_compute_pricing, + on_settled=_on_settled, + gate=CheckoutGateConfig( + api_key=API_KEY, + merchant_name="Multi-Product Co.", + per_request_policy=_per_request_policy, + ), +) - # Per-product identity gate. - enforcement = policy["enforcement"] if policy and "enforcement" in policy else None - gate = build_gate_from_policy(policy, api_key=API_KEY) - gate_result = await run_gate_with_enforcement(request, gate, enforcement=enforcement) - - if gate_result.status == "denied": - # Hard mode: propagate the gate's structured 403 verbatim. - return JSONResponse(content=gate_result.denial_body, status_code=gate_result.denial_status or 403) - - # gate_result.status is one of: "verified" (gate ran + passed), - # "unverified" (soft mode swallowed a denial), "anonymous" (no gate fired). - # Persist this on the order row so ops can distinguish soft passes from hard - # passes and from no-gate-product orders. For the limited print, an - # "unverified" status is a real fraud signal worth flagging in ops. - identity_status = gate_result.status - - # ... settle payment, create order with `identity_status` column, return 200 ... - return JSONResponse( - { - "order": {"product": product["name"], "total_usd": product["price_usd"]}, - "identity_status": identity_status, - } - ) + +app = FastAPI() + + +@app.post("/purchase") +async def purchase(request: Request) -> JSONResponse: + return await checkout.handle_fastapi(request) diff --git a/examples/signed_ucp_merchant.py b/examples/signed_ucp_merchant.py index 7afcb9f..07d6a83 100644 --- a/examples/signed_ucp_merchant.py +++ b/examples/signed_ucp_merchant.py @@ -1,13 +1,15 @@ -"""Signed UCP profile example — ``/.well-known/ucp`` + ``/.well-known/jwks.json``. +"""Signed UCP profile example: ``/.well-known/ucp`` + ``/.well-known/jwks.json``. -AgentScore's ``agentscore-profile+jws`` is a vendor extension layered on top of -the UCP profile for trust-mode verifiers (regulated-commerce, AP2-aware) that -opt into auditable cryptographic provenance. UCP §6 itself does NOT mandate -profile-body signing; production UCP merchants commonly ship unsigned, and -vanilla UCP-aware agents read the canonical body and ignore the ``signature`` -field. This example wires both routes against a persistent signing key -(env-loaded for prod, ephemeral for dev) for verifiers that DO opt into the -signed envelope. +AgentScore's ``agentscore-profile+jws`` is a vendor extension on top of UCP for +trust-mode verifiers (regulated-commerce, AP2-aware) that opt into auditable +cryptographic provenance. UCP §6 itself does NOT mandate profile-body signing; +production UCP merchants commonly ship unsigned, and vanilla UCP-aware agents +read the canonical body and ignore the ``signature`` field. + +The 2.0 SDK ships `build_signed_ucp_response` + `build_signed_jwks_response` +which fold loading + signing + Cache-Control + CORS into one call. Pass a +`Checkout` instance and the helpers compose the `payment_handlers` block +from the configured rails automatically. Run:: @@ -18,116 +20,106 @@ * Set ``UCP_SIGNING_KEY_JWK_PRIVATE`` to a JSON-encoded private JWK (mint via :func:`generate_ucp_signing_key` once, persist in your secret manager). * The kid in the env JWK MUST match what verifiers will see in your published - profile — pick a stable name like ``merchant-2026-05``. -* Configure ``Cache-Control: public, max-age=300`` (or longer) on - ``/.well-known/jwks.json`` so verifiers don't hammer the endpoint. + profile; pick a stable name like ``merchant-2026-05``. * Rotate by minting a new key + new kid, publishing both in the JWKS, signing new profiles with the new key, then dropping the old JWK after your verifier cache TTL expires. + +Call `bootstrap_ucp_signing_key()` in your lifespan handler so a malformed +``UCP_SIGNING_KEY_JWK_PRIVATE`` env value fails the deploy fast instead of +surfacing on the first ``/.well-known/ucp`` hit. """ from __future__ import annotations -import json -import logging +from contextlib import asynccontextmanager +from typing import Any -from fastapi import FastAPI +from fastapi import FastAPI, Request, Response from fastapi.responses import JSONResponse -from agentscore_commerce.identity import ( - AgentScoreGatePolicy, - UCPServiceBinding, - UCPSigningKey, - UCPVerificationError, - build_jwks_response, - build_ucp_profile, - load_ucp_signing_key_from_env, - mpp_payment_handler, - sign_ucp_profile, - verify_ucp_profile, +from agentscore_commerce import AgentScoreGatePolicy, Checkout, PricingResult +from agentscore_commerce.discovery import ( + bootstrap_ucp_signing_key, + build_signed_jwks_response, + build_signed_ucp_response, + default_a2a_services, + well_known_preflight_response, ) from agentscore_commerce.payment import TempoRailSpec -logger = logging.getLogger("signed_ucp_merchant") +SIGNING_KID = "merchant-2026-05" + + +async def _compute_pricing(_ctx: Any) -> PricingResult: + return PricingResult(amount_usd=1.0) + + +checkout = Checkout( + rails={"tempo": TempoRailSpec(recipient="0xfeedface")}, + url="https://agents.example.com/purchase", + compute_pricing=_compute_pricing, +) + -# Env-loader kwargs pin the production kid + alg defaults for this example. -# ``UCP_SIGNING_KEY_JWK_PRIVATE`` (env) wins when set; ``UCP_SIGNING_KEY_KID`` -# and ``UCP_SIGNING_KEY_ALG`` override these defaults at runtime. The helper -# caches the loaded key across requests and serializes concurrent first-callers -# so two threads can never publish a JWKS that disagrees with the just-signed JWS. -_SIGNING_KEY_OPTS = {"default_kid": "merchant-2026-05"} +@asynccontextmanager +async def lifespan(_app: FastAPI): + # Eager-load the signing key so a malformed env JWK fails the deploy fast. + bootstrap_ucp_signing_key(default_kid=SIGNING_KID) + yield -app = FastAPI() +app = FastAPI(lifespan=lifespan) @app.get("/.well-known/ucp") -async def well_known_ucp() -> JSONResponse: - key = load_ucp_signing_key_from_env(**_SIGNING_KEY_OPTS) - profile = build_ucp_profile( +async def well_known_ucp(request: Request) -> Response: + resp = build_signed_ucp_response( + checkout=checkout, name="My Agent Service", - services={ - "dev.ucp.shopping": [ - UCPServiceBinding( - version="2026-04-08", - spec="https://ucp.dev/2026-04-08/specification/overview", - transport="mcp", - endpoint="https://agents.example.com/api/ucp/mcp", - schema="https://ucp.dev/services/shopping/mcp.openrpc.json", - ), - ], - }, - payment_handlers={ - **mpp_payment_handler( - networks=[TempoRailSpec(recipient="0xfeedface")], - ), - }, - signing_keys=[UCPSigningKey.from_jwk(key.public_jwk)], - # Optional: declare merchant gate policy as an `sh.agentscore.identity` capability - # binding inside the public profile. Static policy declaration only — no per-operator - # claims. Per-operator identity attestation flows through the AP2 risk-signal endpoint. + well_known_ucp_url="https://agents.example.com/.well-known/ucp", + services=default_a2a_services(agent_card_url="https://agents.example.com/.well-known/agent-card.json"), + request_headers=dict(request.headers), + signing_kid=SIGNING_KID, + # Optional: declare merchant gate policy as an `sh.agentscore.identity` + # capability binding inside the public profile. Static policy + # declaration only; per-operator identity attestation flows through the + # AP2 risk-signal endpoint. agentscore_gate=AgentScoreGatePolicy( require_kyc=True, min_age=21, allowed_jurisdictions=["US"], ), ) - signed = sign_ucp_profile( - profile.to_dict(), - signing_key=key.private_key, - kid=key.public_jwk["kid"], - alg=key.public_jwk.get("alg", "EdDSA"), - ) - return JSONResponse(signed, headers={"Cache-Control": "public, max-age=60"}) + return Response(content=resp.content, status_code=resp.status, media_type=resp.media_type, headers=resp.headers) @app.get("/.well-known/jwks.json") -async def well_known_jwks() -> JSONResponse: - key = load_ucp_signing_key_from_env(**_SIGNING_KEY_OPTS) - return JSONResponse( - build_jwks_response([key.public_jwk]), - headers={ - "Cache-Control": "public, max-age=300", - "Content-Type": "application/jwk-set+json", - }, - ) +async def well_known_jwks(request: Request) -> Response: + resp = build_signed_jwks_response(request_headers=dict(request.headers), signing_kid=SIGNING_KID) + return Response(content=resp.content, status_code=resp.status, media_type=resp.media_type, headers=resp.headers) + + +@app.options("/.well-known/ucp") +@app.options("/.well-known/jwks.json") +async def well_known_preflight(request: Request) -> Response: + preflight = well_known_preflight_response(dict(request.headers)) + return Response(status_code=preflight.status, headers=preflight.headers) @app.get("/_selftest/ucp") -async def selftest() -> JSONResponse: +async def selftest(request: Request) -> JSONResponse: """Local round-trip: sign+serve+fetch+verify, return UCPVerificationError code on failure.""" - profile_resp = await well_known_ucp() - jwks_resp = await well_known_jwks() - # FastAPI's `JSONResponse.body` is typed as `bytes | memoryview[int]`; coerce to - # plain `bytes` so `json.loads` accepts both branches without a type error. + import json + + from agentscore_commerce.identity import UCPVerificationError, verify_ucp_profile + + profile_resp = await well_known_ucp(request) + jwks_resp = await well_known_jwks(request) profile = json.loads(bytes(profile_resp.body)) jwks = json.loads(bytes(jwks_resp.body)) try: verify_ucp_profile(profile, jwks) return JSONResponse({"ok": True, "kid": profile["signing_keys"][0]["kid"]}) except UCPVerificationError as exc: - logger.exception("UCP self-test verification failed") - return JSONResponse( - {"ok": False, "code": exc.code, "error": type(exc).__name__}, - status_code=500, - ) + return JSONResponse({"ok": False, "code": exc.code}, status_code=500) diff --git a/examples/variable_cost_merchant.py b/examples/variable_cost_merchant.py index 88f4dc6..8405a3f 100644 --- a/examples/variable_cost_merchant.py +++ b/examples/variable_cost_merchant.py @@ -1,31 +1,33 @@ -"""Example: variable-cost merchant supporting BOTH x402 upto AND MPP tempo session - -Scenario: you sell something where the cost depends on output — LLM completions, -transcription, video transcode, etc. You don't know the final price until the work is done. -Two protocols solve this; both are advertised on the 402 so agents can pick whichever -they support. - - x402 upto (one-shot) - - Agent signs Permit2 authorizing up to a max amount - - Vendor does the work, knows actual cost after - - Response sets Settlement-Overrides: {"amount":""} - - Facilitator settles for actual; difference auto-refunds - - MPP tempo session (streaming) - - Agent opens a channel with on-chain deposit - - Vendor streams output as SSE - - Cumulative cost grows → vendor emits voucher requests - - Agent signs each voucher mid-stream - - Final settle on close reclaims unspent deposit - -Python rail-verification peer deps are now wrapped — see `create_x402_server` and -`create_mppx_server` in `agentscore_commerce.payment` for one-call setup. This example -keeps the response shape direct (helper-composed) so the variable-cost flow is readable; -in production wire `create_x402_server(rails=["x402-base-mainnet-upto"], facilitator="coinbase")` -and call `process_x402_settle(...)` for verification + settlement. +"""Example: variable-cost merchant supporting BOTH x402 upto AND MPP tempo session. + +Scenario: you sell something where the cost depends on output (LLM completions, +transcription, video transcode, etc.). You don't know the final price until the +work is done. Two protocols solve this; both are advertised on the 402 so +agents can pick whichever they support. + +x402 upto (one-shot) + * Agent signs Permit2 authorizing up to a max amount. + * Vendor does the work, knows actual cost after. + * Response sets ``Settlement-Overrides: {"amount":""}``. + * Facilitator settles for actual; difference auto-refunds. + +MPP tempo session (streaming) + * Agent opens a channel with on-chain deposit. + * Vendor streams output as SSE. + * Cumulative cost grows; vendor emits voucher requests. + * Agent signs each voucher mid-stream. + * Final settle on close reclaims unspent deposit. + +These flows are too custom to fit the one-shot `Checkout(...)` model: +`compute_pricing` returns a single amount, but variable-cost discovers the +amount AFTER the request runs (upto) or grows it cumulatively (session). The +example keeps the 402-emit body custom (the warnings + dynamic `max_usd` block +aren't in the canonical 402 schema) and the settle path manual; vendors +compose `create_x402_server` + Permit2 extensions or `create_mppx_server` +(TempoSessionRailSpec) at the vendor layer. Peer deps: - pip install agentscore-commerce[fastapi,x402,mppx,coinbase] + pip install 'agentscore-commerce[fastapi,x402,mppx,coinbase]' Env vars: X402_BASE_RECIPIENT — your Base wallet (USDC payouts for upto rail) @@ -35,6 +37,9 @@ Run: uvicorn examples.variable_cost_merchant:app --port 3000 """ +import json +from base64 import b64encode + from fastapi import FastAPI, Request from fastapi.responses import JSONResponse @@ -45,7 +50,7 @@ ) REALM = "llm.example.com" -MAX_USDC = 0.5 # upper bound advertised; actual bill <= this +MAX_USDC = 0.5 # upper bound advertised; actual bill <= this. app = FastAPI() @@ -57,28 +62,41 @@ def _build_402_body(url: str) -> tuple[dict, dict]: ] body = { "payment_required": True, + "x402Version": 2, "product_name": "LLM completion", "pricing": {"max_usd": MAX_USDC, "billing": "pay-per-token"}, "warnings": [ - "Cost is variable — final amount depends on output length.", + "Cost is variable; final amount depends on output length.", "For one-shot completions use x402 upto. For long streams use tempo session.", ], } - headers = {"www-authenticate": www_authenticate_header(directives)} + headers = { + "www-authenticate": www_authenticate_header(directives), + # `PAYMENT-REQUIRED` (x402 wire) is the base64-encoded body. Spec-strict + # clients (Coinbase awal, purl) parse this header first; the JSON body + # is the fallback for clients that don't. + "PAYMENT-REQUIRED": b64encode(json.dumps({"x402Version": 2, "resource": {"url": url}}).encode()).decode(), + } return body, headers +async def _run_your_llm(_prompt: str) -> tuple[str, int]: + return "completion text here", 1234 + + @app.post("/llm/complete") async def complete(request: Request): - """x402 upto path — single JSON response with Settlement-Overrides.""" - if not request.headers.get("x-payment"): + """x402 upto path: single JSON response with Settlement-Overrides.""" + # x402 carries the credential in either `x-payment` or `payment-signature` + # depending on client (purl uses payment-signature; awal uses x-payment). + if not (request.headers.get("x-payment") or request.headers.get("payment-signature")): body, headers = _build_402_body(str(request.url)) return JSONResponse(body, status_code=402, headers=headers) body = await request.json() - text, tokens_used = await run_your_llm(body.get("prompt", "")) + text, tokens_used = await _run_your_llm(body.get("prompt", "")) - # Calculate actual cost based on tokens consumed + # Calculate actual cost based on tokens consumed. actual_usd = tokens_used * 0.000_002 # $2 per 1M tokens actual_atomic = str(int(actual_usd * 1_000_000)) # USDC atomic units @@ -92,19 +110,15 @@ async def complete(request: Request): @app.post("/llm/stream") async def stream(request: Request): - """MPP tempo session path — agent opens channel, server streams SSE with mid-stream vouchers. + """MPP tempo session path: agent opens channel, server streams SSE with mid-stream vouchers. Production wiring: ``mpp = await create_mppx_server(secret_key=MPP_SECRET, rails={ - "tempo_session": TempoSessionRailSpec(recipient=TEMPO_RECIPIENT, escrow_contract=TEMPO_ESCROW, - store=YourChannelStore())})`` — parse channel state from ``Authorization: Payment``, - emit SSE chunks, request fresh voucher signatures as cumulative cost grows, close - channel on completion. + "tempo_session": TempoSessionRailSpec(recipient=TEMPO_RECIPIENT, + escrow_contract=TEMPO_ESCROW, store=YourChannelStore())})``; parse channel + state from ``Authorization: Payment``, emit SSE chunks, request fresh voucher + signatures as cumulative cost grows, close channel on completion. """ if not request.headers.get("authorization"): body, headers = _build_402_body(str(request.url)) return JSONResponse(body, status_code=402, headers=headers) return JSONResponse({"error": "stream-not-implemented"}, status_code=501) - - -async def run_your_llm(_prompt: str) -> tuple[str, int]: - return "completion text here", 1234