examples: align with node — full identity-only + real SDK helpers in

multi-rail/stripe-multichain/variable-cost

- identity_only.py: add create_session_on_missing + capture_wallet + public
  route to match node's teaching depth.
- multi_rail_merchant.py: replace _create_multichain_payment_intent stub with
  the real create_multichain_payment_intent helper + pi_cache writes.
- stripe_multichain_merchant.py: rename POST /buy -> POST /checkout, add the
  3-network instructions block (matches node prose).
- variable_cost_merchant.py: swap raw 402 dicts for build_402_body +
  build_accepted_methods + build_agent_instructions + build_how_to_pay +
  build_pricing_block; wire create_x402_server/create_mppx_server hooks.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: vvillait88

examples/identity_only.py

@@ -1,42 +1,88 @@
-"""Example: compliance gate without payment
+"""Example: identity gate without payment
 
-Scenario: you sell something where the gating is the whole product — your service handles
-its own billing (Stripe, invoice, prepaid credit, etc.) but you need to verify the agent
-operator is KYC'd, age-verified, sanctions-clear, and in an allowed jurisdiction before
-delivering.
+Scenario: you have an existing checkout / payment flow you don't want to change,
+but you want to verify the agent is KYC'd before letting them transact. Use the
+commerce/identity middleware as a thin wrapper over your existing endpoints.
 
-This is the smallest possible commerce integration. Mount the gate, write your route,
-done. No 402 logic, no payment plumbing — just identity gating.
+Common cases:
+    * Compliance-required content (age-gated, sanctioned-restricted)
+    * High-value transactions where you want extra identity assurance
+    * Adding agent KYC to an existing human-only Stripe checkout
+
+This is the smallest possible commerce integration. Mount the gate, write your
+route, done. No 402 logic, no payment plumbing; just identity gating.
 
 Peer deps:
     pip install agentscore-commerce[fastapi]
 
 Env vars:
-    AGENTSCORE_API_KEY — get one at agentscore.sh/dashboard
+    AGENTSCORE_API_KEY — your AgentScore API key
 
 Run: uvicorn examples.identity_only:app --port 3000
 """
 
-from fastapi import Depends, FastAPI
+from __future__ import annotations
+
+import os
+from typing import Any
 
-from agentscore_commerce.identity.fastapi import AgentScoreGate, get_agentscore_data
+from fastapi import Depends, FastAPI, Request
+
+from agentscore_commerce.identity.fastapi import (
+    AgentScoreGate,
+    capture_wallet,
+    get_agentscore_data,
+)
+from agentscore_commerce.identity.sessions import CreateSessionOnMissing
 
 app = FastAPI()
 
+API_KEY = os.environ.get("AGENTSCORE_API_KEY", "ask_test_dummy")
+
+# ── Apply identity gate to specific routes ──────────────────────────────────
 gate = AgentScoreGate(
-    api_key="ask_...",  # use os.environ["AGENTSCORE_API_KEY"] in prod
+    api_key=API_KEY,
     require_kyc=True,
     require_sanctions_clear=True,
     min_age=21,
     allowed_jurisdictions=["US"],
+    # When the agent has no identity header, auto-create a verification session
+    # so the 403 body carries verify_url + poll_secret + agent_instructions.
+    create_session_on_missing=CreateSessionOnMissing(
+        api_key=API_KEY,
+        context="restricted-access",
+    ),
 )
 
 
-@app.post("/deliver", dependencies=[Depends(gate)])
-async def deliver(assess: dict = Depends(get_agentscore_data)):
+@app.post("/restricted", dependencies=[Depends(gate)])
+async def restricted(assess: dict[str, Any] = Depends(get_agentscore_data)) -> dict[str, Any]:
     """Gated route — only reached when the agent passes the compliance policy.
 
-    `assess` is the raw `/v1/assess` response. Use it for downstream business logic that
-    depends on the verified identity (audit trail, per-operator pricing, etc.).
+    `assess` is the raw `/v1/assess` response: ``{ decision, operator,
+    kyc_verified, age_bracket, jurisdiction, ... }``. Run your own business
+    logic here; buy something via your existing Stripe flow, grant access to
+    gated content, write to your DB, whatever. AgentScore's job ends at "this
+    agent is verified, here's their operator id."
     """
-    return {"status": "delivered", "operator": assess.get("resolved_operator")}
+    return {"ok": True, "operator": assess.get("resolved_operator")}
+
+
+# ── Optional: capture an agent's wallet after payment lands ────────────────
+# (only relevant if your downstream payment flow exposes the signer wallet)
+@app.post("/restricted/capture-wallet-example", dependencies=[Depends(gate)])
+async def capture_wallet_example(request: Request) -> dict[str, Any]:
+    body = await request.json()
+    await capture_wallet(
+        request,
+        wallet_address=body["signer_address"],
+        network="evm",
+        idempotency_key=body.get("payment_intent_id"),
+    )
+    return {"ok": True}
+
+
+# ── Public routes (no gate) ────────────────────────────────────────────────
+@app.get("/public-info")
+async def public_info() -> dict[str, str]:
+    return {"message": "open access — no identity required"}

examples/multi_rail_merchant.py

@@ -65,6 +65,7 @@
     validate_x402_network_config,
 )
 from agentscore_commerce.stripe_multichain import (
+    create_multichain_payment_intent,
     create_pi_cache,
     simulate_deposit_if_test_mode,
 )
@@ -75,21 +76,17 @@
 SOLANA_NETWORK_CAIP2 = os.environ.get("SOLANA_NETWORK_CAIP2", networks.solana.mainnet.caip2)
 validate_x402_network_config(base_network=X402_BASE_NETWORK)
 
+# Singleton Stripe client + PI / deposit-address cache. Redis-backed when
+# REDIS_URL is set (multi-task deployments need this so a deposit lands on
+# whichever task settles it).
+import stripe  # noqa: E402  optional peer dep installed by the example user
+
+stripe_client = stripe.StripeClient(STRIPE_SECRET_KEY)
 pi_cache = create_pi_cache(redis_url=os.environ.get("REDIS_URL"))
 
 app = FastAPI()
 
 
-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": "..."}
-
-
 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 {}
@@ -109,12 +106,20 @@ async def _compute_pricing(ctx: Any) -> PricingResult:
 
 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)
+    total_cents = round(ctx.pricing.amount_usd * 100)
+    result = create_multichain_payment_intent(
+        stripe=stripe_client,
+        amount=total_cents,
+        networks=["tempo", "base", "solana"],
+    )
+    for addr in result.deposit_addresses.values():
+        await pi_cache.cache_address(addr)
+        pi_cache.cache_payment_intent(addr, result.payment_intent_id)
+    pi_cache.cache_network_addresses(result.payment_intent_id, result.deposit_addresses)
     return {
-        "tempo": addresses["tempo"],
-        "x402_base": addresses["base"],
-        "solana_mpp": addresses["solana"],
+        "tempo": result.deposit_addresses["tempo"],
+        "x402_base": result.deposit_addresses["base"],
+        "solana_mpp": result.deposit_addresses["solana"],
     }
 
 

examples/stripe_multichain_merchant.py

@@ -1,18 +1,19 @@
 """Example: Stripe-anchored multichain merchant
 
-Scenario: you want to accept agent payments but settle through Stripe so all your existing
-billing/refund/dashboard infrastructure keeps working. Stripe issues a single PaymentIntent
-with deposit_options for tempo/base/solana — the agent picks any chain to send USDC, and
-Stripe auto-captures the PI when the deposit lands.
+Scenario: you want crypto payments but you're already a Stripe merchant. Use Stripe's
+``deposit_options`` to issue per-PI deposit addresses on multiple chains (Tempo, Base,
+Solana). Agent picks a chain and sends USDC to the matching address; Stripe auto-captures
+when funds land. Net: one Stripe PI per purchase, multi-chain optionality, settlement
+tracked in Stripe.
 
-Distinct from the Stripe SPT (Shared Payment Token) flow — this is the "agent sends crypto,
-Stripe handles settlement on your behalf" path.
+Distinct from Stripe SPT (Shared Payment Token), which is for user-approved cards via
+the ``link-cli`` flow. This example is the "merchant funds via crypto rails" path.
 
 Peer deps:
-    pip install agentscore-commerce[fastapi,stripe]
+    pip install 'agentscore-commerce[fastapi,stripe]'
 
 Env vars:
-    STRIPE_SECRET_KEY — your sk_... secret key (sk_test_ for testnet)
+    STRIPE_SECRET_KEY — sk_live_... or sk_test_...
 
 Run: uvicorn examples.stripe_multichain_merchant:app --port 3000
 """
@@ -33,37 +34,51 @@
 app = FastAPI()
 
 
-@app.post("/buy")
-async def buy(body: dict):
-    # Create a multichain PaymentIntent — Stripe issues deposit addresses for each requested chain.
+@app.post("/checkout")
+async def checkout(body: dict) -> dict:
+    amount_cents = round(float(body["amount_usd"]) * 100)
+
+    # 1. Create a Stripe PI with deposit addresses on tempo + base + solana.
     result = create_multichain_payment_intent(
         stripe=stripe_client,
-        amount=body.get("amount_cents", 25000),
+        amount=amount_cents,
         networks=["tempo", "base", "solana"],
-        metadata={"order_id": body.get("order_id", "ord_demo"), "merchant": "example"},
-        idempotency_key=body.get("order_id"),
+        metadata={"order_id": body.get("order_id"), "merchant": "example-store"},
+        idempotency_key=f"pi-{body['order_id']}-{amount_cents}" if body.get("order_id") else None,
     )
 
+    # 2. Return per-network deposit addresses to the agent (or 402 with
+    # addresses embedded — see multi_rail_merchant.py for the full 402-builder
+    # pattern).
+    amount_usd = body["amount_usd"]
+    tempo = result.deposit_addresses.get("tempo")
+    base = result.deposit_addresses.get("base")
+    solana = result.deposit_addresses.get("solana")
     return {
         "payment_intent_id": result.payment_intent_id,
         "deposit_addresses": result.deposit_addresses,
-        "pay_to": {
-            "base": result.deposit_addresses.get("base"),
-            "tempo": result.deposit_addresses.get("tempo"),
+        "instructions": {
+            "tempo": (f"Send {amount_usd} USDC on Tempo to {tempo}" if tempo else "Tempo not available for this PI"),
+            "base": (f"Send {amount_usd} USDC on Base to {base}" if base else "Base not available for this PI"),
+            "solana": (
+                f"Send {amount_usd} USDC on Solana to {solana}" if solana else "Solana not available for this PI"
+            ),
         },
     }
 
 
-# Testnet helper: simulate a deposit landing on a PI. Useful for end-to-end testing without
-# real on-chain transfers. For the typical "fire after PI mint if sk_test_" pattern, prefer
-# `simulate_deposit_if_test_mode` which gates internally — see multi_rail_merchant.py.
+# ── Testnet helper: simulate a deposit landing on a PI ──────────────────────
+# Useful for end-to-end testing without real on-chain transfers. For the
+# typical "fire after PI mint if sk_test_" pattern, prefer
+# `simulate_deposit_if_test_mode` which gates internally — see
+# multi_rail_merchant.py.
 @app.post("/testnet/simulate-deposit")
-async def simulate_deposit(body: dict):
+async def simulate_deposit(body: dict) -> dict:
     await simulate_crypto_deposit(
         payment_intent_id=body["payment_intent_id"],
         network=body["network"],
         stripe_secret_key=os.environ["STRIPE_SECRET_KEY"],
-        stripe_version="2026-03-04.preview",
+        stripe_version="2026-03-04.preview",  # if you're on a preview API
         token_currency="usdc",
         transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS,
     )