refactor(identity, payment): flatten LoadUCPSigningKeyOptions + delete dead CreateX402ServerOptions

Last remaining wrapper dataclass in the SDK. load_ucp_signing_key_from_env
now takes the 5 fields directly as kwargs; the env-var-name overrides remain
on the public surface so multi-env-namespace merchants still pin once via
a module-level dict + spread.

While auditing, CreateX402ServerOptions surfaced as a leftover wrapper that
no longer matched its function (create_x402_server was already kwarg-shaped);
deleted along with its barrel export.

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

Author: vvillait88

README.md

@@ -25,7 +25,7 @@ pip install 'agentscore-commerce[fastapi,x402,coinbase]'
 | Submodule | What it provides |
 |---|---|
 | `agentscore_commerce.identity.{fastapi,flask,django,aiohttp,sanic,middleware}` | Trust gate middleware: KYC, sanctions (account name + signer wallet), age, jurisdiction. `AgentScoreGate(...)` (or `agentscore_gate(app, ...)` on Flask/Sanic), `get_agentscore_data(...)`, `capture_wallet(...)`, `get_signer_verdict(...)`. The gate extracts the payment signer pre-evaluate and passes it to `/v1/assess`, so the API composes both wallet-binding (`signer_match`) and OFAC SDN wallet-address (`signer_sanctions`) verdicts on one round trip. |
-| `agentscore_commerce.identity` (package level) | Re-exports the denial helpers: `denial_reason_status`, `denial_reason_to_body`, `build_signer_mismatch_body`, `build_contact_support_next_steps`, `verification_agent_instructions`, `is_fixable_denial`, `FIXABLE_DENIAL_REASONS`. The per-framework adapter modules also expose `get_gate_quota_info(request)` for surfacing X-RateLimit info from gate state. Also re-exports the per-product policy helpers: `PolicyBlock`, `GateResult`, `EnforcementMode`, `IdentityStatus`, `build_gate_from_policy`, `run_gate_with_enforcement`, `shipping_country_allowed`, `shipping_state_allowed` (for multi-product merchants where each product carries its own compliance config: hard gate vs soft vs none, per-product shipping allowlists). Key + token helpers: `load_ucp_signing_key_from_env` + `LoadUCPSigningKeyOptions` (cached env-driven loader for the UCP signing key — reads `UCP_SIGNING_KEY_JWK_PRIVATE` JSON JWK, detects alg from shape, falls back to ephemeral when unset, sanitizes errors so key bytes never reach logs, concurrent-safe via `threading.Lock`); `hash_operator_token` (sha256 hex of plaintext `opc_...` — for merchants persisting `operator_token_id` to their own DB without ever storing the plaintext). |
+| `agentscore_commerce.identity` (package level) | Re-exports the denial helpers: `denial_reason_status`, `denial_reason_to_body`, `build_signer_mismatch_body`, `build_contact_support_next_steps`, `verification_agent_instructions`, `is_fixable_denial`, `FIXABLE_DENIAL_REASONS`. The per-framework adapter modules also expose `get_gate_quota_info(request)` for surfacing X-RateLimit info from gate state. Also re-exports the per-product policy helpers: `PolicyBlock`, `GateResult`, `EnforcementMode`, `IdentityStatus`, `build_gate_from_policy`, `run_gate_with_enforcement`, `shipping_country_allowed`, `shipping_state_allowed` (for multi-product merchants where each product carries its own compliance config: hard gate vs soft vs none, per-product shipping allowlists). Key + token helpers: `load_ucp_signing_key_from_env` (cached env-driven loader for the UCP signing key — reads `UCP_SIGNING_KEY_JWK_PRIVATE` JSON JWK, detects alg from shape, falls back to ephemeral when unset, sanitizes errors so key bytes never reach logs, concurrent-safe via `threading.Lock`; env-var names and `default_kid` / `default_alg` are overridable as kwargs); `hash_operator_token` (sha256 hex of plaintext `opc_...` — for merchants persisting `operator_token_id` to their own DB without ever storing the plaintext). |
 | `agentscore_commerce.payment` | `networks`, `USDC`, `rails` registries; `payment_directive`, `build_payment_directive`, `www_authenticate_header`, `payment_required_header`, `alias_amount_fields` (v1↔v2 amount field shim that emits both `amount` and `maxAmountRequired` so v1-only x402 parsers like Coinbase awal can read v2 bodies), `settlement_override_header`, `dispatch_settlement_by_network`, `extract_payment_signer` (accepts positional `x402_payment_header` AND/OR `authorization_header=` kwarg; recovers signer from x402 EIP-3009 `payload.authorization.from` OR MPP `Authorization: Payment <base64>` `did:pkh:eip155:<chain>:<addr>` / `did:pkh:solana:<genesis>:<addr>` source DID), `detect_rail_from_headers` (returns `"x402"` / `"mpp"` / `None` from inbound headers), `register_x402_schemes_v1_v2`; drop-in x402 helpers: `validate_x402_network_config` (boot-time guard), `verify_x402_request` (parse + validate inbound X-Payment), `process_x402_settle` (verify-then-settle with one call), `classify_x402_settle_result` (maps the tagged settle result to a recommended HTTP status / code / next_steps so merchants get a controlled envelope without coupling to facilitator-specific error text), `classify_orchestration_error` (same `ClassifiedX402Error` shape but for uncaught exceptions thrown elsewhere in the orchestration; returns `None` for unknown errors so merchants rethrow instead of swallowing); `zero_amount_carve_out` (skip CDP / pympp upstream verify+settle for $0 settles where the upstream rejects value=0 payloads; parses the credential, lifts signer + network, returns a `ZeroSettleResult` shaped identically to the success path so callers branch on rail, not on result shape); `usd_to_atomic` (Decimal-based USD → atomic int, ROUND_HALF_UP — for Tempo / Solana / Base USDC amount construction). |
 | `agentscore_commerce.discovery` | `is_discovery_probe_request`, `build_discovery_probe_response` (with optional `x402_sample` for x402-aware crawlers like `awal x402 details`), `sample_x402_accept_for_network` (USDC sample-accept builder for known CAIP-2 networks), `build_well_known_mpp`, `build_llms_txt` + `llms_txt_identity_section` + `llms_txt_payment_section` (compact + verbose modes), `build_skill_md` (Claude-Skill-compatible `/skill.md` agent-discovery manifest; strictly agent-facing data only, no internal posture), `agentscore_openapi_snippets`, `build_bazaar_discovery_payload`, `NoindexNonDiscoveryMiddleware` (ASGI middleware that emits `X-Robots-Tag: noindex` on every path except the agent-discovery surfaces; defaults cover `/openapi.json`, `/llms.txt`, `/skill.md`, `/.well-known/{mpp.json,agent-card.json,ucp,jwks.json}`, `/favicon.{png,ico}`; pure helpers `is_discovery_path` + `DEFAULT_DISCOVERY_PATHS` for non-ASGI frameworks). |
 | `agentscore_commerce.challenge` | `build_402_body`, `build_accepted_methods`, `build_identity_metadata`, `build_how_to_pay`, `build_agent_instructions` (auto-emits per-rail `compatible_clients`: smoke-verified CLIs the agent should use; vendor override supported; pure helper `compatible_clients_by_rails(rails)` returns the same map for vendors building custom 402s), `build_pricing_block` (cents to dollar-string with optional shipping/tax), `first_encounter_agent_memory` (cross-merchant hint, returns the canonical block or `None` based on a per-merchant first-seen flag), `OrderReceipt` (dataclass for the post-settlement 200 response shape); `respond_402`, a drop-in 402 emit that preserves pympp's `WWW-Authenticate` and layers x402's `PAYMENT-REQUIRED`. `build_validation_error`: structured 4xx body builder (`{error: {code, message}, required_fields?, example_body?, next_steps?, ...extra}`) so vendors compose body shapes by name instead of inlining at every validation site. |

agentscore_commerce/identity/__init__.py

@@ -64,7 +64,6 @@
 )
 from agentscore_commerce.identity.ucp_jwks import (
     GeneratedUCPKey,
-    LoadUCPSigningKeyOptions,
     UCPVerificationError,
     build_jwks_response,
     generate_ucp_signing_key,
@@ -118,7 +117,6 @@ def _load_asgi_middleware() -> tuple[Any, Any]:
     "GateResult",
     "GeneratedUCPKey",
     "IdentityStatus",
-    "LoadUCPSigningKeyOptions",
     "OperatorVerification",
     "PolicyBlock",
     "SignerSanctions",

agentscore_commerce/identity/ucp_jwks.py

@@ -540,23 +540,6 @@ async def jwks():
 _logger = logging.getLogger(__name__)
 
 
-@dataclass(frozen=True)
-class LoadUCPSigningKeyOptions:
-    """Configuration for :func:`load_ucp_signing_key_from_env`.
-
-    Env-var names are overridable so a merchant can run multiple distinct signing
-    keys from different env namespaces (e.g. ``PROD_UCP_JWK`` vs ``STAGING_UCP_JWK``).
-    ``default_kid`` and ``default_alg`` are used when the env JWK is absent or
-    doesn't carry its own ``kid`` / can't dictate alg via kty+crv.
-    """
-
-    env_jwk_var: str = "UCP_SIGNING_KEY_JWK_PRIVATE"
-    env_kid_var: str = "UCP_SIGNING_KEY_KID"
-    env_alg_var: str = "UCP_SIGNING_KEY_ALG"
-    default_kid: str = "merchant-default"
-    default_alg: Literal["EdDSA", "ES256"] = "EdDSA"
-
-
 _env_loader_cache: dict[tuple[str, str, str, str, str], GeneratedUCPKey] = {}
 _env_loader_lock = threading.Lock()
 
@@ -585,34 +568,41 @@ def _detect_alg_from_jwk(jwk: dict[str, Any]) -> Literal["EdDSA", "ES256"] | Non
     return None
 
 
-def _build_env_signing_key(opts: LoadUCPSigningKeyOptions) -> GeneratedUCPKey:
+def _build_env_signing_key(
+    *,
+    env_jwk_var: str,
+    env_kid_var: str,
+    env_alg_var: str,
+    default_kid: str,
+    default_alg: Literal["EdDSA", "ES256"],
+) -> GeneratedUCPKey:
     """Load (or generate) one signing key per env state. No locking (caller wraps)."""
-    kid_default = _read_env_trimmed(opts.env_kid_var) or opts.default_kid
-    raw_alg = (_read_env_trimmed(opts.env_alg_var) or "").upper()
+    kid_default = _read_env_trimmed(env_kid_var) or default_kid
+    raw_alg = (_read_env_trimmed(env_alg_var) or "").upper()
     # Case-insensitive env-alg comparison: secret configs commonly carry casing
     # drift (``"es256"``, ``" ES256 "``, ``"eS256"``). Strict exact-match would
     # silently downgrade to the default and operators would publish a JWKS
     # containing the wrong key family.
-    alg_fallback: Literal["EdDSA", "ES256"] = "ES256" if raw_alg == "ES256" else opts.default_alg
+    alg_fallback: Literal["EdDSA", "ES256"] = "ES256" if raw_alg == "ES256" else default_alg
 
-    env_jwk = _read_env_trimmed(opts.env_jwk_var)
+    env_jwk = _read_env_trimmed(env_jwk_var)
     if env_jwk:
         from joserfc.jwk import ECKey, OKPKey  # type: ignore[import-not-found]
 
         try:
             jwk_dict = json.loads(env_jwk)
         except json.JSONDecodeError as exc:
-            msg = f"{opts.env_jwk_var} is not valid JSON: {exc}"
+            msg = f"{env_jwk_var} is not valid JSON: {exc}"
             raise ValueError(msg) from exc
 
         if not isinstance(jwk_dict, dict) or not jwk_dict:
-            msg = f"{opts.env_jwk_var} must be a non-empty JWK object; got {type(jwk_dict).__name__}."
+            msg = f"{env_jwk_var} must be a non-empty JWK object; got {type(jwk_dict).__name__}."
             raise ValueError(msg)
 
         detected_alg = _detect_alg_from_jwk(jwk_dict)
         if not detected_alg:
             msg = (
-                f"{opts.env_jwk_var} has unsupported kty/crv "
+                f"{env_jwk_var} has unsupported kty/crv "
                 f"(got kty={jwk_dict.get('kty')!r} crv={jwk_dict.get('crv')!r}); "
                 "expected OKP+Ed25519 or EC+P-256."
             )
@@ -625,7 +615,7 @@ def _build_env_signing_key(opts: LoadUCPSigningKeyOptions) -> GeneratedUCPKey:
             # back fields of the input JWK including private key material. Surface only
             # the exception class so logs never carry key bytes through stderr / CloudWatch.
             msg = (
-                f"{opts.env_jwk_var} has malformed key material ({type(exc).__name__}). "
+                f"{env_jwk_var} has malformed key material ({type(exc).__name__}). "
                 "Verify the JWK is well-formed and matches the declared kty/crv. "
                 "Underlying details suppressed to avoid leaking key bytes."
             )
@@ -656,60 +646,70 @@ def _build_env_signing_key(opts: LoadUCPSigningKeyOptions) -> GeneratedUCPKey:
             "Loaded persistent UCP signing key kid=%s alg=%s from %s",
             public_jwk["kid"],
             detected_alg,
-            opts.env_jwk_var,
+            env_jwk_var,
         )
         return GeneratedUCPKey(private_key=priv, public_jwk=public_jwk)
 
     _logger.error(
         "%s not set; generating ephemeral signing key. Verifier caches will break across restarts. "
         "NOT SAFE FOR PRODUCTION.",
-        opts.env_jwk_var,
+        env_jwk_var,
     )
     return generate_ucp_signing_key(kid=kid_default, alg=alg_fallback)
 
 
-def load_ucp_signing_key_from_env(opts: LoadUCPSigningKeyOptions | None = None) -> GeneratedUCPKey:
+def load_ucp_signing_key_from_env(
+    *,
+    env_jwk_var: str = "UCP_SIGNING_KEY_JWK_PRIVATE",
+    env_kid_var: str = "UCP_SIGNING_KEY_KID",
+    env_alg_var: str = "UCP_SIGNING_KEY_ALG",
+    default_kid: str = "merchant-default",
+    default_alg: Literal["EdDSA", "ES256"] = "EdDSA",
+) -> GeneratedUCPKey:
     """Load the merchant's UCP signing key from env, with concurrent-safe caching.
 
-    On first call (per ``opts``): reads ``opts.env_jwk_var``, parses it as a JWK,
+    On first call (per kwargs): reads ``env_jwk_var``, parses it as a JWK,
     validates kty/crv (OKP+Ed25519 or EC+P-256), and projects to a canonical
     public JWK. Falls back to an ephemeral keypair when the env var is missing
     or whitespace-only (dev-friendly; logs a loud warning).
 
-    Subsequent calls with the same ``opts`` return the cached key without
+    Subsequent calls with the same kwargs return the cached key without
     re-reading env. Concurrent first-callers serialize on a lock so only one
     key generation runs; the rest receive the cached result.
 
-    Different ``opts`` values get separate cache entries: a merchant running
-    one signing key per env namespace (e.g. prod vs staging) does not collide.
+    Different kwargs get separate cache entries: a merchant running one signing
+    key per env namespace (e.g. prod vs staging) does not collide. Env-var
+    names are overridable so a merchant can run multiple distinct signing keys
+    from different env namespaces (e.g. ``PROD_UCP_JWK`` vs ``STAGING_UCP_JWK``).
+    ``default_kid`` and ``default_alg`` are used when the env JWK is absent or
+    doesn't carry its own ``kid`` / can't dictate alg via kty+crv.
 
     Env-driven precedence:
 
-    * Embedded ``kid`` in the JWK wins over ``opts.env_kid_var`` env value;
-      empty-string ``kid`` in the env JWK falls through to ``opts.default_kid``.
-    * Structural ``kty``+``crv`` in the JWK wins over ``opts.env_alg_var`` env
+    * Embedded ``kid`` in the JWK wins over ``env_kid_var`` env value;
+      empty-string ``kid`` in the env JWK falls through to ``default_kid``.
+    * Structural ``kty``+``crv`` in the JWK wins over ``env_alg_var`` env
       value (which is only consulted in the ephemeral fallback path).
 
     Raises ``ValueError`` with a sanitized message for malformed env JWKs;
     raw exception detail is intentionally suppressed so key bytes can never
     reach logs.
     """
-    resolved = opts if opts is not None else LoadUCPSigningKeyOptions()
-    cache_key = (
-        resolved.env_jwk_var,
-        resolved.env_kid_var,
-        resolved.env_alg_var,
-        resolved.default_kid,
-        resolved.default_alg,
-    )
+    cache_key = (env_jwk_var, env_kid_var, env_alg_var, default_kid, default_alg)
     cached = _env_loader_cache.get(cache_key)
     if cached is not None:
         return cached
     with _env_loader_lock:
         cached = _env_loader_cache.get(cache_key)
         if cached is not None:
             return cached
-        result = _build_env_signing_key(resolved)
+        result = _build_env_signing_key(
+            env_jwk_var=env_jwk_var,
+            env_kid_var=env_kid_var,
+            env_alg_var=env_alg_var,
+            default_kid=default_kid,
+            default_alg=default_alg,
+        )
         _env_loader_cache[cache_key] = result
         return result
 
@@ -726,7 +726,6 @@ def _reset_ucp_signing_key_cache() -> None:
 
 __all__ = [
     "GeneratedUCPKey",
-    "LoadUCPSigningKeyOptions",
     "UCPVerificationError",
     "build_jwks_response",
     "generate_ucp_signing_key",

agentscore_commerce/payment/__init__.py

@@ -42,7 +42,6 @@
 )
 from agentscore_commerce.payment.x402 import register_x402_schemes_v1_v2
 from agentscore_commerce.payment.x402_server import (
-    CreateX402ServerOptions,
     CustomScheme,
     X402FacilitatorChoice,
     X402SymbolicRail,
@@ -80,7 +79,6 @@
     "USDC",
     "X402_SUPPORTED_BASE_NETWORKS",
     "ClassifiedX402Error",
-    "CreateX402ServerOptions",
     "CustomScheme",
     "MppxRails",
     "NetworkFamily",

agentscore_commerce/payment/x402_server.py

@@ -23,7 +23,7 @@
 
 import importlib
 import os
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal
 
 from agentscore_commerce.payment.networks import networks
@@ -51,38 +51,6 @@ class CustomScheme:
     scheme: Any
 
 
-@dataclass
-class CreateX402ServerOptions:
-    """Configuration for :func:`create_x402_server`."""
-
-    facilitator: X402FacilitatorChoice | Any = "http"
-    """Facilitator selection — ``"coinbase"`` (requires ``cdp-sdk`` peer dep + the
-    ``CDP_API_KEY_ID`` / ``CDP_API_KEY_SECRET`` env vars or explicit ``cdp_api_key_id``
-    / ``cdp_api_key_secret`` args), ``"http"`` (public testnet facilitator at
-    ``x402.org``), or any pre-built facilitator instance."""
-
-    rails: list[X402SymbolicRail] = field(default_factory=list)
-    """Symbolic rail names to register schemes for. Each gets v1+v2 dual-register
-    applied. Requires ``x402[evm]`` peer dep installed."""
-
-    schemes: list[CustomScheme] = field(default_factory=list)
-    """Advanced: register custom (network, scheme) pairs in addition to ``rails``."""
-
-    bazaar: bool = False
-    """Register the Bazaar discovery extension. Requires the extension peer dep installed."""
-
-    initialize: bool = True
-    """Initialize the server immediately (calls facilitator). Default ``True``."""
-
-    cdp_api_key_id: str | None = None
-    """CDP API key id for the Coinbase facilitator. Falls back to
-    ``CDP_API_KEY_ID`` env var. Only consulted when ``facilitator="coinbase"``."""
-
-    cdp_api_key_secret: str | None = None
-    """CDP API key secret for the Coinbase facilitator. Falls back to
-    ``CDP_API_KEY_SECRET`` env var. Only consulted when ``facilitator="coinbase"``."""
-
-
 def _import_optional(module_name: str) -> Any | None:
     """Try to import a module; return ``None`` if not installed."""
     try:
@@ -331,7 +299,6 @@ def build_x402_accepts_for_402(
 
 
 __all__ = [
-    "CreateX402ServerOptions",
     "CustomScheme",
     "X402FacilitatorChoice",
     "X402SymbolicRail",

examples/signed_ucp_merchant.py

@@ -36,7 +36,6 @@
 
 from agentscore_commerce.identity import (
     AgentScoreGatePolicy,
-    LoadUCPSigningKeyOptions,
     UCPServiceBinding,
     UCPSigningKey,
     UCPVerificationError,
@@ -50,20 +49,20 @@
 
 logger = logging.getLogger("signed_ucp_merchant")
 
-# Env-loader options pin the production kid + alg defaults for this example.
+# 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 = LoadUCPSigningKeyOptions(default_kid="merchant-2026-05")
+_SIGNING_KEY_OPTS = {"default_kid": "merchant-2026-05"}
 
 
 app = FastAPI()
 
 
 @app.get("/.well-known/ucp")
 async def well_known_ucp() -> JSONResponse:
-    key = load_ucp_signing_key_from_env(_SIGNING_KEY_OPTS)
+    key = load_ucp_signing_key_from_env(**_SIGNING_KEY_OPTS)
     profile = build_ucp_profile(
         name="My Agent Service",
         services={
@@ -98,14 +97,14 @@ async def well_known_ucp() -> JSONResponse:
         profile.to_dict(),
         signing_key=key.private_key,
         kid=key.public_jwk["kid"],
-        alg=key.public_jwk.get("alg", _SIGNING_KEY_OPTS.default_alg),
+        alg=key.public_jwk.get("alg", "EdDSA"),
     )
     return JSONResponse(signed, headers={"Cache-Control": "public, max-age=60"})
 
 
 @app.get("/.well-known/jwks.json")
 async def well_known_jwks() -> JSONResponse:
-    key = load_ucp_signing_key_from_env(_SIGNING_KEY_OPTS)
+    key = load_ucp_signing_key_from_env(**_SIGNING_KEY_OPTS)
     return JSONResponse(
         build_jwks_response([key.public_jwk]),
         headers={