AgentScore Commerce (Python)

pip install agentscore-commerce[fastapi]   # or [flask], [django], [aiohttp], [sanic], [starlette], [stripe]

pip install 'agentscore-commerce[fastapi,x402,coinbase]'
# Set CDP_API_KEY_ID and CDP_API_KEY_SECRET in the environment.

from agentscore_commerce.payment import build_x402_accepts_for_402, create_x402_server

x402_server = await create_x402_server(facilitator="coinbase", rails=["x402-base-mainnet"])

x402_accepts = build_x402_accepts_for_402(
    x402_server,
    network="eip155:8453",
    price=f"${total_usd}",
    pay_to=os.environ["TREASURY_BASE_RECIPIENT"],
    max_timeout_seconds=300,
)

from agentscore_commerce import (
    A2AAgentSkill,
    AgentScoreGatePolicy,
    UCPPaymentHandlerBinding,
    UCPServiceBinding,
    UCPSigningKey,
    build_a2a_agent_card,
    build_ucp_profile,
    ucp_a2a_extension,
)

# 1. Google A2A v1.0 Signed Agent Card, published at /.well-known/agent-card.json.
#    Per UCP §A2A binding (https://ucp.dev/specification/checkout-a2a/), agents that
#    serve UCP via the A2A transport MUST declare the canonical UCP extension URI in
#    capabilities.extensions[] so platforms arriving via A2A detect UCP support
#    without re-fetching the profile. Pass an empty capabilities map when you serve
#    UCP at the discovery layer but have no formal capability bindings yet.
#    Skills declare what the agent can do; they are AgentSkill objects (top-level on
#    the card). Identity claims live in a separate AgentCardSignature (RFC 7515 JWS)
#    wrapping the serialized card, NOT in the card body itself.
card = build_a2a_agent_card(
    name="My Agent Service",
    description="Buy products via agent payments.",
    url="https://agents.example.com",
    version="1.0.0",
    skills=[
        A2AAgentSkill(
            id="purchase",
            name="Purchase",
            description="Buy products via agent payments.",
            tags=["commerce", "payment"],
        ),
    ],
    extensions=[ucp_a2a_extension()],
)

# 2. Google Universal Commerce Protocol profile, published at /.well-known/ucp
#    Output shape: {"ucp": {"version", "services", "capabilities",
#    "payment_handlers", "name?", "supported_versions?"}, "signing_keys": [...]}
#    — services / capabilities / payment_handlers are MAPS keyed by reverse-DNS
#    service / capability / handler name (UCP spec §3 + §6).
profile = build_ucp_profile(
    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=[{"network": "tempo-mainnet", "chain_id": 4217, "recipient": TEMPO_ADDR}]),
        **x402_payment_handler(networks=[{"network": "base-8453", "recipient": BASE_ADDR}]),
        **stripe_spt_payment_handler(profile_id="profile_5xKvNqM9BaH"),
    },
    signing_keys=[UCPSigningKey(kid="me", kty="EC", alg="ES256", crv="P-256", extras={"x": "...", "y": "..."})],
    # 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.
    agentscore_gate=AgentScoreGatePolicy(
        require_kyc=True, min_age=21, allowed_jurisdictions=["US"],
    ),
)

from agentscore_commerce import (
    UCPSigningKey,
    UCPVerificationError,
    build_jwks_response,
    generate_ucp_signing_key,
    sign_ucp_profile,
    verify_ucp_profile,
)

# One-time per merchant: generate or load a key. Persist private_key in your secrets manager.
key = generate_ucp_signing_key(kid="merchant-2026-05")

# On every /.well-known/ucp request:
profile = build_ucp_profile(
    name="My Agent Service",
    services={...},  # keyed by service name, see above
    payment_handlers={...},  # keyed by handler reverse-DNS name, see above
    signing_keys=[UCPSigningKey.from_jwk(key.public_jwk)],
)
signed = sign_ucp_profile(
    profile.to_dict(),
    signing_key=key.private_key,
    kid=key.public_jwk["kid"],
    alg="EdDSA",
)
return signed

# On /.well-known/jwks.json:
return build_jwks_response([key.public_jwk])

import os
from fastapi import FastAPI, Request
from agentscore_commerce import (
    Checkout, CheckoutGateConfig, DiscoveryProbeConfig, PricingResult, pricing_result,
    SolanaMppRailSpec, StripeRailSpec, TempoRailSpec, X402BaseRailSpec,
    validate_shipping_against_policy,
)
from agentscore_commerce.discovery import default_a2a_services

app = FastAPI()


async def _pre_validate(ctx):
    body = ctx.request.body or {}
    product = await lookup_product(body.get("product_slug"))
    validate_shipping_against_policy(
        country=body.get("shipping", {}).get("country", ""),
        state=body.get("shipping", {}).get("state", ""),
        policy=product,
        product_name=product["name"],
    )
    return {"product": product}


async def _compute_pricing(ctx) -> PricingResult:
    return pricing_result(
        subtotal_cents=ctx.state["product"]["price_cents"],
        tax_cents=ctx.state["product"]["tax_cents"],
        tax_rate=ctx.state["product"]["tax_rate"],
        tax_state=ctx.state["product"]["tax_state"],
    )


async def _on_settled(ctx, outcome):
    return {"ok": True, "order_id": ctx.reference_id, "tx_hash": outcome.tx_hash}


checkout = Checkout(
    rails={
        "tempo":     TempoRailSpec(recipient=os.environ["TEMPO_RECIPIENT"]),
        "x402_base": X402BaseRailSpec(recipient=os.environ["X402_BASE_RECIPIENT"], network="eip155:8453"),
        "solana_mpp":SolanaMppRailSpec(recipient=os.environ["SOLANA_RECIPIENT"], network="solana:mainnet"),
        "stripe":    StripeRailSpec(profile_id=os.environ["STRIPE_PROFILE_ID"]),
    },
    url="https://merchant.example/purchase",
    pre_validate=_pre_validate,
    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"),
    gate=CheckoutGateConfig(
        api_key=os.environ["AGENTSCORE_API_KEY"],
        merchant_name="Merchant",
        require_kyc=True, require_sanctions_clear=True, min_age=21, allowed_jurisdictions=["US"],
    ),
    # Optional: empty-body POSTs without a payment header auto-route to a sample 402
    # so x402 crawlers (awal x402 details, x402-proxy, ...) can discover the surface.
    discovery_probe=DiscoveryProbeConfig(
        realm="merchant.example",
        sample_rail="tempo-mainnet",
        sample_amount_usd=1.0,
        sample_recipient=os.environ["TEMPO_RECIPIENT"],
    ),
)

# One-call mount: GET /.well-known/ucp + GET /.well-known/jwks.json + OPTIONS preflights.
# Flask / Django / aiohttp / Sanic each have a sibling helper
# (`mount_ucp_routes_flask` etc.) with the same kwargs.
checkout.mount_ucp_routes_fastapi(
    app,
    name="Merchant",
    well_known_ucp_url="https://merchant.example/.well-known/ucp",
    services=default_a2a_services(agent_card_url="https://merchant.example/.well-known/agent-card.json"),
    signing_kid="merchant-2026-05",
)


@app.post("/purchase")
async def purchase(request: Request):
    return await checkout.handle_fastapi(request)

from fastapi import Depends, FastAPI, Request
from agentscore_commerce.identity.fastapi import AgentScoreGate, get_gate_degraded_state

app = FastAPI()
gate = AgentScoreGate(api_key=os.environ["AGENTSCORE_API_KEY"], fail_open=True)

@app.post("/purchase", dependencies=[Depends(gate)])
async def purchase(request: Request):
    state = get_gate_degraded_state(request)
    if state["degraded"]:
        logger.warning("gate degraded: %s", state["infra_reason"])
    # ...