diff --git a/agentscore_commerce/challenge/__init__.py b/agentscore_commerce/challenge/__init__.py index 311a213..71bc7d7 100644 --- a/agentscore_commerce/challenge/__init__.py +++ b/agentscore_commerce/challenge/__init__.py @@ -1,38 +1,15 @@ """402-body builders + pricing/receipt/agent-memory helpers.""" -from agentscore_commerce.challenge.accepted_methods import ( - BuildAcceptedMethodsInput, - SolanaMppConfig, - StripeConfig, - TempoConfig, - X402BaseConfig, - build_accepted_methods, -) -from agentscore_commerce.challenge.agent_instructions import ( - BuildAgentInstructionsInput, - build_agent_instructions, -) +from agentscore_commerce.challenge.accepted_methods import build_accepted_methods +from agentscore_commerce.challenge.agent_instructions import build_agent_instructions, compatible_clients_by_rails from agentscore_commerce.challenge.agent_memory import ( AgentMemoryHint, build_agent_memory_hint, first_encounter_agent_memory, ) -from agentscore_commerce.challenge.body import Build402BodyInput, X402PaymentRequired, build_402_body -from agentscore_commerce.challenge.how_to_pay import ( - BuildHowToPayInput, - HowToPayRails, - SolanaMppRailConfig, - StripeRailConfig, - TempoRailConfig, - X402BaseRailConfig, - build_how_to_pay, -) -from agentscore_commerce.challenge.identity import ( - IdentityMetadataInput, - IdentityMode, - SignerMatchResult, - build_identity_metadata, -) +from agentscore_commerce.challenge.body import X402PaymentRequired, build_402_body +from agentscore_commerce.challenge.how_to_pay import build_how_to_pay +from agentscore_commerce.challenge.identity import IdentityMode, SignerMatchResult, build_identity_metadata from agentscore_commerce.challenge.order_receipt import ( OrderNextSteps, OrderProductInfo, @@ -40,38 +17,19 @@ ShippingAddress, ) from agentscore_commerce.challenge.pricing import PricingBlock, build_pricing_block -from agentscore_commerce.challenge.respond_402 import Respond402Input, Respond402Result, respond_402 -from agentscore_commerce.challenge.validation_error import ( - BuildValidationErrorInput, - build_validation_error, -) +from agentscore_commerce.challenge.respond_402 import Respond402Result, respond_402 +from agentscore_commerce.challenge.validation_error import build_validation_error __all__ = [ "AgentMemoryHint", - "Build402BodyInput", - "BuildAcceptedMethodsInput", - "BuildAgentInstructionsInput", - "BuildHowToPayInput", - "BuildValidationErrorInput", - "HowToPayRails", - "IdentityMetadataInput", "IdentityMode", "OrderNextSteps", "OrderProductInfo", "OrderReceipt", "PricingBlock", - "Respond402Input", "Respond402Result", "ShippingAddress", "SignerMatchResult", - "SolanaMppConfig", - "SolanaMppRailConfig", - "StripeConfig", - "StripeRailConfig", - "TempoConfig", - "TempoRailConfig", - "X402BaseConfig", - "X402BaseRailConfig", "X402PaymentRequired", "build_402_body", "build_accepted_methods", @@ -81,6 +39,7 @@ "build_identity_metadata", "build_pricing_block", "build_validation_error", + "compatible_clients_by_rails", "first_encounter_agent_memory", "respond_402", ] diff --git a/agentscore_commerce/challenge/accepted_methods.py b/agentscore_commerce/challenge/accepted_methods.py index 7d28c5b..3b02b95 100644 --- a/agentscore_commerce/challenge/accepted_methods.py +++ b/agentscore_commerce/challenge/accepted_methods.py @@ -1,90 +1,85 @@ """accepted_methods[] builder for enriched 402 bodies.""" -from dataclasses import dataclass, field from typing import Any +_DEFAULT_TEMPO = { + "network": "tempo-mainnet", + "chain_id": 4217, + "token": "0x20C000000000000000000000b9537d11c60E8b50", + "symbol": "USDC.e", + "decimals": 6, +} +_DEFAULT_X402_BASE = { + "network": "eip155:8453", + "chain_id": 8453, + "token": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", + "symbol": "USDC", + "decimals": 6, +} +_DEFAULT_SOLANA_MPP = { + "network": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp", + "token": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", + "symbol": "USDC", + "decimals": 6, +} +_DEFAULT_STRIPE_RAILS = ["card", "link", "shared_payment_token"] -@dataclass -class TempoConfig: - recipient: str - network: str = "tempo-mainnet" - chain_id: int = 4217 - token: str = "0x20C000000000000000000000b9537d11c60E8b50" - symbol: str = "USDC.e" - decimals: int = 6 +def build_accepted_methods( + *, + tempo: dict[str, Any] | None = None, + x402_base: dict[str, Any] | None = None, + solana_mpp: dict[str, Any] | None = None, + stripe: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: + """Build the accepted_methods[] array. Each rail entry conditionally included if vendor passed it. -@dataclass -class X402BaseConfig: - recipient: str - network: str = "eip155:8453" - chain_id: int = 8453 - token: str = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" - symbol: str = "USDC" - decimals: int = 6 - - -@dataclass -class SolanaMppConfig: - recipient: str - network: str = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp" - token: str = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" - symbol: str = "USDC" - decimals: int = 6 - - -@dataclass -class StripeConfig: - profile_id: str | None = None - rails: list[str] = field(default_factory=lambda: ["card", "link", "shared_payment_token"]) - - -@dataclass -class BuildAcceptedMethodsInput: - tempo: TempoConfig | None = None - x402_base: X402BaseConfig | None = None - solana_mpp: SolanaMppConfig | None = None - stripe: StripeConfig | None = None - - -def build_accepted_methods(input: BuildAcceptedMethodsInput) -> list[dict[str, Any]]: - """Build the accepted_methods[] array. Each rail entry conditionally included if vendor passed it.""" + Each rail value is a plain dict. Required key: ``recipient`` (or ``profile_id`` for stripe). + Optional keys override the rail's protocol defaults: ``network``, ``chain_id``, ``token``, + ``symbol``, ``decimals`` (for chain rails) or ``rails`` (for stripe). + """ out: list[dict[str, Any]] = [] - if input.tempo: + if tempo: out.append( { "method": "tempo/charge", - "network": input.tempo.network, - "chain_id": input.tempo.chain_id, - "token": input.tempo.token, - "symbol": input.tempo.symbol, - "decimals": input.tempo.decimals, - "pay_to": input.tempo.recipient, + "network": tempo.get("network", _DEFAULT_TEMPO["network"]), + "chain_id": tempo.get("chain_id", _DEFAULT_TEMPO["chain_id"]), + "token": tempo.get("token", _DEFAULT_TEMPO["token"]), + "symbol": tempo.get("symbol", _DEFAULT_TEMPO["symbol"]), + "decimals": tempo.get("decimals", _DEFAULT_TEMPO["decimals"]), + "pay_to": tempo["recipient"], } ) - if input.x402_base: + if x402_base: out.append( { "method": "x402/exact", - "network": input.x402_base.network, - "chain_id": input.x402_base.chain_id, - "token": input.x402_base.token, - "symbol": input.x402_base.symbol, - "decimals": input.x402_base.decimals, - "pay_to": input.x402_base.recipient, + "network": x402_base.get("network", _DEFAULT_X402_BASE["network"]), + "chain_id": x402_base.get("chain_id", _DEFAULT_X402_BASE["chain_id"]), + "token": x402_base.get("token", _DEFAULT_X402_BASE["token"]), + "symbol": x402_base.get("symbol", _DEFAULT_X402_BASE["symbol"]), + "decimals": x402_base.get("decimals", _DEFAULT_X402_BASE["decimals"]), + "pay_to": x402_base["recipient"], } ) - if input.solana_mpp: + if solana_mpp: out.append( { "method": "x402/exact", - "network": input.solana_mpp.network, - "token": input.solana_mpp.token, - "symbol": input.solana_mpp.symbol, - "decimals": input.solana_mpp.decimals, - "pay_to": input.solana_mpp.recipient, + "network": solana_mpp.get("network", _DEFAULT_SOLANA_MPP["network"]), + "token": solana_mpp.get("token", _DEFAULT_SOLANA_MPP["token"]), + "symbol": solana_mpp.get("symbol", _DEFAULT_SOLANA_MPP["symbol"]), + "decimals": solana_mpp.get("decimals", _DEFAULT_SOLANA_MPP["decimals"]), + "pay_to": solana_mpp["recipient"], + } + ) + if stripe: + out.append( + { + "method": "stripe/charge", + "rails": stripe.get("rails", _DEFAULT_STRIPE_RAILS), + "profile_id": stripe.get("profile_id"), } ) - if input.stripe: - out.append({"method": "stripe/charge", "rails": input.stripe.rails, "profile_id": input.stripe.profile_id}) return out diff --git a/agentscore_commerce/challenge/agent_instructions.py b/agentscore_commerce/challenge/agent_instructions.py index 865cb85..ab6b43a 100644 --- a/agentscore_commerce/challenge/agent_instructions.py +++ b/agentscore_commerce/challenge/agent_instructions.py @@ -1,7 +1,6 @@ """agent_instructions block builder for the 402 body.""" from collections.abc import Iterable -from dataclasses import dataclass, field from typing import Any, Literal _TEMPO_WARNING = ( @@ -74,7 +73,7 @@ def compatible_clients_by_rails(rails: Iterable[str]) -> dict[str, list[str]] | def _default_compatible_clients(how_to_pay: dict[str, Any]) -> dict[str, list[str]] | None: """Default ``compatible_clients`` derived from the rails declared in ``how_to_pay``. - Vendors override this in ``BuildAgentInstructionsInput(compatible_clients=...)`` + Vendors override via the ``compatible_clients`` kwarg of ``build_agent_instructions`` to add their own tested clients or remove entries that don't fit their endpoint. Verified state as of the SDK release. """ @@ -90,55 +89,44 @@ def _default_compatible_clients(how_to_pay: dict[str, Any]) -> dict[str, list[st return compatible_clients_by_rails(rails) -@dataclass -class BuildAgentInstructionsInput: - how_to_pay: dict[str, Any] - recommended_tools: list[str] | None = None - wallet_compatibility: str | None = None - timeout_seconds: int = 300 - warnings: list[str] | None = None +def build_agent_instructions( + *, + how_to_pay: dict[str, Any], + recommended_tools: list[str] | None = None, + wallet_compatibility: str | None = None, + timeout_seconds: int = 300, + warnings: list[str] | None = None, # Appended to the default protocol-footgun warnings. Use this to keep the SDK's # protocol warnings AND add merchant-specific notes. Ignored when ``warnings`` is set. - extra_warnings: list[str] | None = None - recommended: str | None = None + extra_warnings: list[str] | None = None, + recommended: str | None = None, # Per-rail list of client names the merchant has verified work end-to-end. # Vendors set this from their own smoke matrix — defaults to None, in which case # the field is not emitted (avoids vouching for clients the merchant has not tested). # Keys are rail identifiers (e.g. "x402_base", "tempo_mpp"); values are display labels. - compatible_clients: dict[str, list[str]] | None = None - extra: dict[str, Any] = field(default_factory=dict) - - -def build_agent_instructions(input: BuildAgentInstructionsInput) -> dict[str, Any]: + compatible_clients: dict[str, list[str]] | None = None, + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: """Build the agent_instructions block — combines how_to_pay with tools, warnings, compat note, timeout. Defaults adapt to the rails declared in ``how_to_pay``: only tempo-relevant warnings/tools appear if ``how_to_pay["tempo"]`` is set, only x402-relevant ones if ``x402_base``/ ``solana_mpp`` are set. Vendors override ``warnings``/``recommended_tools`` for full control. """ - recommended_tools = ( - input.recommended_tools if input.recommended_tools is not None else _default_recommended_tools(input.how_to_pay) - ) - warnings = ( - input.warnings - if input.warnings is not None - else [*_default_warnings(input.how_to_pay), *(input.extra_warnings or [])] - ) - compatible_clients = ( - input.compatible_clients - if input.compatible_clients is not None - else _default_compatible_clients(input.how_to_pay) - ) + resolved_tools = recommended_tools if recommended_tools is not None else _default_recommended_tools(how_to_pay) + resolved_warnings = warnings if warnings is not None else [*_default_warnings(how_to_pay), *(extra_warnings or [])] + resolved_clients = compatible_clients if compatible_clients is not None else _default_compatible_clients(how_to_pay) out: dict[str, Any] = { - "how_to_pay": input.how_to_pay, - "recommended_tools": recommended_tools, - "wallet_compatibility": input.wallet_compatibility or DEFAULT_WALLET_COMPATIBILITY, - "timeout_seconds": input.timeout_seconds, - "warnings": warnings, + "how_to_pay": how_to_pay, + "recommended_tools": resolved_tools, + "wallet_compatibility": wallet_compatibility or DEFAULT_WALLET_COMPATIBILITY, + "timeout_seconds": timeout_seconds, + "warnings": resolved_warnings, } - if input.recommended: - out["recommended"] = input.recommended - if compatible_clients: - out["compatible_clients"] = compatible_clients - out.update(input.extra) + if recommended: + out["recommended"] = recommended + if resolved_clients: + out["compatible_clients"] = resolved_clients + if extra: + out.update(extra) return out diff --git a/agentscore_commerce/challenge/agent_memory.py b/agentscore_commerce/challenge/agent_memory.py index f396142..e614925 100644 --- a/agentscore_commerce/challenge/agent_memory.py +++ b/agentscore_commerce/challenge/agent_memory.py @@ -29,14 +29,14 @@ def first_encounter_agent_memory( Use directly with the ``agent_memory`` field of :func:`build_402_body`:: - body = build_402_body(Build402BodyInput( + body = build_402_body( accepted_methods=accepted, agent_instructions=instructions, pricing=pricing, agent_memory=first_encounter_agent_memory( first_encounter=not has_seen_operator(operator_token), ), - )) + ) Returning ``None`` means ``build_402_body`` cleanly skips the field instead of emitting ``agent_memory: null`` (which would imply "I tried but failed" rather than diff --git a/agentscore_commerce/challenge/body.py b/agentscore_commerce/challenge/body.py index 656f9a7..519888e 100644 --- a/agentscore_commerce/challenge/body.py +++ b/agentscore_commerce/challenge/body.py @@ -1,6 +1,6 @@ """build_402_body — full enriched 402 response body builder.""" -from dataclasses import asdict, dataclass, field, is_dataclass +from dataclasses import asdict, dataclass, is_dataclass from typing import Any, Literal from agentscore_commerce.challenge.pricing import PricingBlock @@ -13,55 +13,52 @@ class X402PaymentRequired: version: Literal[1, 2] = 2 -@dataclass -class Build402BodyInput: - accepted_methods: list[dict[str, Any]] - agent_instructions: dict[str, Any] | None = None - identity_metadata: dict[str, Any] | None = None - agent_memory: Any = None - pricing: PricingBlock | None = None - amount_usd: str | None = None - currency: str | None = None - order_id: str | None = None - product: dict[str, str] | None = None - retry_body: Any = None - recommended: str | None = None - x402: X402PaymentRequired | None = None - extra: dict[str, Any] = field(default_factory=dict) - - -def build_402_body(input: Build402BodyInput) -> dict[str, Any]: +def build_402_body( + *, + accepted_methods: list[dict[str, Any]], + agent_instructions: dict[str, Any] | None = None, + identity_metadata: dict[str, Any] | None = None, + agent_memory: Any = None, + pricing: PricingBlock | None = None, + amount_usd: str | None = None, + currency: str | None = None, + order_id: str | None = None, + product: dict[str, str] | None = None, + retry_body: Any = None, + recommended: str | None = None, + x402: X402PaymentRequired | None = None, + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: """Assemble the full enriched 402 response body. Each section conditionally included if vendor passed it.""" - body: dict[str, Any] = {"payment_required": True, "accepted_methods": input.accepted_methods} - if input.x402: - body["x402Version"] = input.x402.version - body["accepts"] = alias_amount_fields(input.x402.accepts) - if input.amount_usd is not None: - body["amount_usd"] = input.amount_usd - if input.currency: - body["currency"] = input.currency - if input.pricing: - body["pricing"] = input.pricing.to_dict() - if input.order_id is not None: - body["order_id"] = input.order_id - if input.product: - body["product"] = input.product - if input.recommended: - body["recommended"] = input.recommended - if input.retry_body is not None: - body["retry_body"] = input.retry_body - if input.identity_metadata: - body.update(input.identity_metadata) - if input.agent_instructions: - body["agent_instructions"] = input.agent_instructions - if input.agent_memory is not None: + body: dict[str, Any] = {"payment_required": True, "accepted_methods": accepted_methods} + if x402: + body["x402Version"] = x402.version + body["accepts"] = alias_amount_fields(x402.accepts) + if amount_usd is not None: + body["amount_usd"] = amount_usd + if currency: + body["currency"] = currency + if pricing: + body["pricing"] = pricing.to_dict() + if order_id is not None: + body["order_id"] = order_id + if product: + body["product"] = product + if recommended: + body["recommended"] = recommended + if retry_body is not None: + body["retry_body"] = retry_body + if identity_metadata: + body.update(identity_metadata) + if agent_instructions: + body["agent_instructions"] = agent_instructions + if agent_memory is not None: # AgentMemoryHint is a dataclass; merchants pass it directly via # first_encounter_agent_memory(...). Convert here so JSONResponse / # json.dumps can serialise without per-merchant boilerplate. body["agent_memory"] = ( - asdict(input.agent_memory) - if is_dataclass(input.agent_memory) and not isinstance(input.agent_memory, type) - else input.agent_memory + asdict(agent_memory) if is_dataclass(agent_memory) and not isinstance(agent_memory, type) else agent_memory ) - body.update(input.extra) + if extra: + body.update(extra) return body diff --git a/agentscore_commerce/challenge/how_to_pay.py b/agentscore_commerce/challenge/how_to_pay.py index 1ff13fc..9d1e6c6 100644 --- a/agentscore_commerce/challenge/how_to_pay.py +++ b/agentscore_commerce/challenge/how_to_pay.py @@ -1,53 +1,11 @@ """how_to_pay block builder — per-rail setup/command/what_it_does for 402 agent_instructions.""" import math -from dataclasses import dataclass -from typing import Any, Literal - - -@dataclass -class TempoRailConfig: - recipient: str - network_name: str = "tempo-mainnet" - chain_id: int = 4217 - recommend: Literal["tempo", "agentscore-pay", "both"] = "both" - - -@dataclass -class X402BaseRailConfig: - recipient: str - network: str = "eip155:8453" - - -@dataclass -class SolanaMppRailConfig: - recipient: str - network: str = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp" - - -@dataclass -class StripeRailConfig: - profile_id: str | None = None - product_name: str | None = None - - -@dataclass -class HowToPayRails: - tempo: TempoRailConfig | None = None - x402_base: X402BaseRailConfig | None = None - solana_mpp: SolanaMppRailConfig | None = None - stripe: StripeRailConfig | None = None - - -@dataclass -class BuildHowToPayInput: - url: str - retry_body_json: str - total_usd: float | str - rails: HowToPayRails - op_token_placeholder: str = "" - max_spend: float | str | None = None +from typing import Any +_DEFAULT_TEMPO = {"network_name": "tempo-mainnet", "chain_id": 4217, "recommend": "both"} +_DEFAULT_X402_BASE = {"network": "eip155:8453"} +_DEFAULT_SOLANA_MPP = {"network": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp"} TEMPO_SETUP = [ "curl -fsSL https://tempo.xyz/install | bash", @@ -67,56 +25,72 @@ class BuildHowToPayInput: ] -def build_how_to_pay(input: BuildHowToPayInput) -> dict[str, Any]: +def build_how_to_pay( + *, + url: str, + retry_body_json: str, + total_usd: float | str, + rails: dict[str, dict[str, Any]], + op_token_placeholder: str = "", # noqa: S107 — literal placeholder, not a secret + max_spend: float | str | None = None, +) -> dict[str, Any]: """Build the agent_instructions.how_to_pay block. Generates per-rail setup/command/what_it_does so agents see concrete commands per rail in the 402 body. + ``rails`` is a dict keyed by rail name (``"tempo"``, ``"x402_base"``, ``"solana_mpp"``, ``"stripe"``); + each value is a plain dict carrying the rail's config (``recipient`` for chain rails, ``profile_id`` + + ``product_name`` for stripe, plus the protocol-default overrides ``network`` / ``network_name`` / + ``chain_id`` / ``recommend``). Pass ``rails={}`` to emit no per-rail block. """ - total_num = float(input.total_usd) if isinstance(input.total_usd, str) else input.total_usd - max_spend = str(input.max_spend) if input.max_spend is not None else f"{math.ceil(total_num) + 1:.2f}" - op_token = input.op_token_placeholder + total_num = float(total_usd) if isinstance(total_usd, str) else total_usd + max_spend_str = str(max_spend) if max_spend is not None else f"{math.ceil(total_num) + 1:.2f}" + op_token = op_token_placeholder block: dict[str, Any] = {} - if input.rails.tempo: - t = input.rails.tempo + tempo = rails.get("tempo") + if tempo: + network_name = tempo.get("network_name", _DEFAULT_TEMPO["network_name"]) + chain_id = tempo.get("chain_id", _DEFAULT_TEMPO["chain_id"]) + recommend = tempo.get("recommend", _DEFAULT_TEMPO["recommend"]) tempo_command = ( f"tempo request -X POST -H 'X-Operator-Token: {op_token}' -H 'Content-Type: application/json' " - f"--json '{input.retry_body_json}' --max-spend {max_spend} {input.url}" + f"--json '{retry_body_json}' --max-spend {max_spend_str} {url}" ) pay_command = ( - f"agentscore-pay pay POST {input.url} --chain tempo -H 'X-Operator-Token: {op_token}' " - f"-H 'Content-Type: application/json' -d '{input.retry_body_json}' --max-spend {max_spend}" + f"agentscore-pay pay POST {url} --chain tempo -H 'X-Operator-Token: {op_token}' " + f"-H 'Content-Type: application/json' -d '{retry_body_json}' --max-spend {max_spend_str}" ) entry: dict[str, Any] = { "setup": TEMPO_SETUP, "prerequisite": ( - f"Run `tempo wallet whoami` and confirm USDC.e balance on {t.network_name} (chain {t.chain_id}) " - f"is at least ${max_spend}. If the tempo CLI is not installed, run the setup commands above first." + f"Run `tempo wallet whoami` and confirm USDC.e balance on {network_name} (chain {chain_id}) " + f"is at least ${max_spend_str}. If the tempo CLI is not installed, run the setup commands above first." ), - "command": pay_command if t.recommend == "agentscore-pay" else tempo_command, + "command": pay_command if recommend == "agentscore-pay" else tempo_command, "what_it_does": ( - f"Hits this endpoint, receives this same 402, signs the MPP challenge on {t.network_name}, and " + f"Hits this endpoint, receives this same 402, signs the MPP challenge on {network_name}, and " "submits the credential back via Authorization: Payment. Either client (tempo request or " "agentscore-pay pay --chain tempo) works — both run the full MPP handshake." ), } - if t.recommend == "both": + if recommend == "both": entry["alternative_command"] = pay_command - elif t.recommend == "agentscore-pay": + elif recommend == "agentscore-pay": entry["alternative_command"] = tempo_command block["tempo"] = entry - if input.rails.x402_base: - b = input.rails.x402_base + x402_base = rails.get("x402_base") + if x402_base: + network = x402_base.get("network", _DEFAULT_X402_BASE["network"]) block["x402_base"] = { "setup": PAY_SETUP_BASE, "prerequisite": ( - f"Run `agentscore-pay balance --chain base` and confirm USDC balance on Base ({b.network}) is at " - f"least ${max_spend}. If the CLI is not installed, run the setup commands above first." + f"Run `agentscore-pay balance --chain base` and confirm USDC balance on Base ({network}) is at " + f"least ${max_spend_str}. If the CLI is not installed, run the setup commands above first." ), "command": ( - f"agentscore-pay pay POST {input.url} --chain base -H 'X-Operator-Token: {op_token}' " - f"-H 'Content-Type: application/json' -d '{input.retry_body_json}' --max-spend {max_spend}" + f"agentscore-pay pay POST {url} --chain base -H 'X-Operator-Token: {op_token}' " + f"-H 'Content-Type: application/json' -d '{retry_body_json}' --max-spend {max_spend_str}" ), "what_it_does": ( "Hits this endpoint, receives this same 402, signs an EIP-3009 USDC TransferWithAuthorization " @@ -125,17 +99,18 @@ def build_how_to_pay(input: BuildHowToPayInput) -> dict[str, Any]: ), } - if input.rails.solana_mpp: - s = input.rails.solana_mpp + solana_mpp = rails.get("solana_mpp") + if solana_mpp: + network = solana_mpp.get("network", _DEFAULT_SOLANA_MPP["network"]) block["solana_mpp"] = { "setup": PAY_SETUP_SOLANA, "prerequisite": ( - f"Run `agentscore-pay balance --chain solana` and confirm USDC balance on Solana ({s.network}) " - f"is at least ${max_spend}. If the CLI is not installed, run the setup commands above first." + f"Run `agentscore-pay balance --chain solana` and confirm USDC balance on Solana ({network}) " + f"is at least ${max_spend_str}. If the CLI is not installed, run the setup commands above first." ), "command": ( - f"agentscore-pay pay POST {input.url} --chain solana -H 'X-Operator-Token: {op_token}' " - f"-H 'Content-Type: application/json' -d '{input.retry_body_json}' --max-spend {max_spend}" + f"agentscore-pay pay POST {url} --chain solana -H 'X-Operator-Token: {op_token}' " + f"-H 'Content-Type: application/json' -d '{retry_body_json}' --max-spend {max_spend_str}" ), "what_it_does": ( "Hits this endpoint, receives this same 402, signs an SPL Token TransferChecked transaction on " @@ -144,11 +119,12 @@ def build_how_to_pay(input: BuildHowToPayInput) -> dict[str, Any]: ), } - if input.rails.stripe: - cfg = input.rails.stripe + stripe = rails.get("stripe") + if stripe: + profile_id = stripe.get("profile_id") + product_name = stripe.get("product_name") or "this purchase" amount_cents = round(total_num * 100) link_cli_blocked = amount_cents > 50000 - product_name = cfg.product_name or "this purchase" spt_context = ( f'Purchasing "{product_name}" via the agent commerce API. The user authorized this purchase ' f"through their AI agent for ${total_num}; charge to be settled via shared payment token over the " @@ -164,7 +140,7 @@ def build_how_to_pay(input: BuildHowToPayInput) -> dict[str, Any]: "via Authorization: Payment MPP header with method=stripe/charge." ), } - if cfg.profile_id and not link_cli_blocked: + if profile_id and not link_cli_blocked: stripe_block["setup_link_cli"] = [ "npm install -g @stripe/link-cli # or use npx -y @stripe/link-cli for one-shot", "link-cli auth login # one-time, opens your Link wallet", @@ -174,13 +150,13 @@ def build_how_to_pay(input: BuildHowToPayInput) -> dict[str, Any]: ( "SPEND_ID=$(link-cli spend-request create " "--payment-method-id " - f"--credential-type shared_payment_token --network-id {cfg.profile_id} " + f"--credential-type shared_payment_token --network-id {profile_id} " f"--amount {amount_cents} " f'--context "{spt_context}" --request-approval --output-json | jq -r .id)' ), ( - f"link-cli mpp pay {input.url} --spend-request-id $SPEND_ID --method POST " - f"--data '{input.retry_body_json}' --header 'X-Operator-Token: {op_token}' --output-json" + f"link-cli mpp pay {url} --spend-request-id $SPEND_ID --method POST " + f"--data '{retry_body_json}' --header 'X-Operator-Token: {op_token}' --output-json" ), ] stripe_block["what_it_does_link_cli"] = ( diff --git a/agentscore_commerce/challenge/identity.py b/agentscore_commerce/challenge/identity.py index ca8d55c..45d90f3 100644 --- a/agentscore_commerce/challenge/identity.py +++ b/agentscore_commerce/challenge/identity.py @@ -14,29 +14,27 @@ class SignerMatchResult: linked_wallets: list[str] | None = None -@dataclass -class IdentityMetadataInput: - mode: IdentityMode - wallet: str | None = None - signer_match_result: SignerMatchResult | None = None - linked_wallets: list[str] | None = None - signer_constraint: str | None = None - - -def build_identity_metadata(input: IdentityMetadataInput) -> dict[str, Any]: +def build_identity_metadata( + *, + mode: IdentityMode, + wallet: str | None = None, + signer_match_result: SignerMatchResult | None = None, + linked_wallets: list[str] | None = None, + signer_constraint: str | None = None, +) -> dict[str, Any]: """Build the identity-metadata block. Echoes wallet-mode signer requirements so agents can self-correct.""" - block: dict[str, Any] = {"identity_mode": input.mode} - if input.mode != "wallet": + block: dict[str, Any] = {"identity_mode": mode} + if mode != "wallet": return block - if input.wallet: + if wallet: block["required_signer"] = ( - input.signer_match_result.expected_signer - if input.signer_match_result and input.signer_match_result.expected_signer - else input.wallet + signer_match_result.expected_signer + if signer_match_result and signer_match_result.expected_signer + else wallet ) - if input.linked_wallets: - block["linked_wallets"] = input.linked_wallets - block["signer_constraint"] = input.signer_constraint or ( + if linked_wallets: + block["linked_wallets"] = linked_wallets + block["signer_constraint"] = signer_constraint or ( "Payment must be signed with the claimed wallet OR any same-operator linked wallet listed in linked_wallets." ) return block diff --git a/agentscore_commerce/challenge/respond_402.py b/agentscore_commerce/challenge/respond_402.py index 35a25ab..0e92d77 100644 --- a/agentscore_commerce/challenge/respond_402.py +++ b/agentscore_commerce/challenge/respond_402.py @@ -17,38 +17,20 @@ Usage:: - from agentscore_commerce.challenge import respond_402, Respond402Input + from agentscore_commerce.challenge import respond_402 - result = respond_402(Respond402Input( + result = respond_402( mppx_challenge_headers=dict(challenge_response.headers), - body=Build402BodyInput(accepted_methods=..., ...), - x402=PaymentRequiredHeaderInput(x402_version=2, accepts=..., resource=...), - )) + body={"accepted_methods": ..., ...}, + x402={"x402_version": 2, "accepts": [...], "resource": {...}}, + ) return JSONResponse(result.body, status_code=result.status, headers=result.headers) """ from dataclasses import dataclass +from typing import Any -from agentscore_commerce.challenge.body import Build402BodyInput, build_402_body -from agentscore_commerce.payment.wwwauthenticate import ( - PaymentRequiredHeaderInput, - payment_required_header, -) - - -@dataclass -class Respond402Input: - """Input for :func:`respond_402`.""" - - #: Headers from the pympp ``compose()`` 402 response. The ``www-authenticate`` - #: header is preserved verbatim — pympp's server-side validator matches credentials - #: to the directive ids it generated, so overwriting breaks the round-trip. - mppx_challenge_headers: dict[str, str] - #: Inputs to :func:`build_402_body` — the rich JSON body sent to the agent. - body: Build402BodyInput - #: When set, layers on the x402 PAYMENT-REQUIRED header (base64-encoded JSON). - #: Omit for merchants that don't accept x402 (Base/Solana) — pympp-only setups. - x402: PaymentRequiredHeaderInput | None = None +from agentscore_commerce.payment.wwwauthenticate import PaymentRequiredHeaderInput, payment_required_header @dataclass @@ -60,15 +42,27 @@ class Respond402Result: status: int = 402 -def respond_402(input: Respond402Input) -> Respond402Result: +def respond_402( + *, + mppx_challenge_headers: dict[str, str], + body: dict[str, Any], + x402: dict[str, Any] | None = None, +) -> Respond402Result: """Compose the rich body + preserved-mppx WWW-Auth + optional x402 PAYMENT-REQUIRED. The merchant wraps the returned ``Respond402Result`` in their framework's response shape (``JSONResponse`` for FastAPI, ``flask.Response`` for Flask, etc.). + + ``body`` is the already-built dict from :func:`build_402_body`. ``x402``, when + set, carries the PAYMENT-REQUIRED header inputs (``x402_version``, ``accepts``, + ``resource``); omit for merchants that don't accept x402 (Base / Solana) — pympp-only + setups. """ - body = build_402_body(input.body) - headers = {k.lower(): v for k, v in input.mppx_challenge_headers.items()} + headers = {k.lower(): v for k, v in mppx_challenge_headers.items()} headers["content-type"] = "application/json" - if input.x402 is not None: - headers["payment-required"] = payment_required_header(input.x402) + if x402 is not None: + # PaymentRequiredHeaderInput still exists pending the wwwauthenticate + # flatten in a subsequent PR; respond_402's public API takes a dict now + # and we adapt internally so the wrapper deletion is invisible to callers. + headers["payment-required"] = payment_required_header(PaymentRequiredHeaderInput(**x402)) return Respond402Result(body=body, headers=headers, status=402) diff --git a/agentscore_commerce/challenge/validation_error.py b/agentscore_commerce/challenge/validation_error.py index a48de0d..bd8945c 100644 --- a/agentscore_commerce/challenge/validation_error.py +++ b/agentscore_commerce/challenge/validation_error.py @@ -9,40 +9,22 @@ for 400/404/409/422. """ -from dataclasses import dataclass, field from typing import Any +_NO_EXAMPLE: Any = object() -@dataclass -class BuildValidationErrorInput: - """Inputs for ``build_validation_error``. - Attributes: - code: Machine-readable error code (e.g. ``'bad_request'``, ``'not_found'``, - ``'out_of_stock'``). - message: Human-readable message — surfaced directly to the user via the agent. - required_fields: Optional schema description of required body fields, keyed by - field name. Surfaced so agents can self-correct without fetching docs. - example_body: Optional concrete example body. Pairs with ``required_fields`` - for max self-serve. Use the ``_HAS_NO_EXAMPLE`` sentinel to omit; pass - ``None`` to emit a literal ``"example_body": null``. - next_steps: Optional next-step hint block (``{action, user_message?, - ...vendor_extras}``). - extra: Vendor-specific top-level fields merged into the body (e.g. ``available``, - ``blocked_states``, ``max_length``). - """ - - code: str - message: str - required_fields: dict[str, str] | None = None +def build_validation_error( + *, + code: str, + message: str, + required_fields: dict[str, str] | None = None, # Sentinel: distinguish "no example provided" from "explicit null in body". - example_body: Any = None - has_example_body: bool = False - next_steps: dict[str, Any] | None = None - extra: dict[str, Any] = field(default_factory=dict) - - -def build_validation_error(input: BuildValidationErrorInput) -> dict[str, Any]: + # Pass any value (including None) to emit it; omit to suppress the field entirely. + example_body: Any = _NO_EXAMPLE, + next_steps: dict[str, Any] | None = None, + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: """Compose a 4xx body that vendors return via their framework's response helper. Combine with the merchant's chosen HTTP status (400 for body shape errors, @@ -50,22 +32,23 @@ def build_validation_error(input: BuildValidationErrorInput) -> dict[str, Any]: Example:: - body = build_validation_error(BuildValidationErrorInput( + body = build_validation_error( code='bad_request', message='product_id, email, and shipping are required', required_fields={'product_id': 'uuid', 'email': 'string', 'shipping': 'object'}, next_steps={'action': 'retry_with_complete_body'}, - )) + ) return JSONResponse(body, status_code=400) """ body: dict[str, Any] = { - "error": {"code": input.code, "message": input.message}, + "error": {"code": code, "message": message}, } - if input.required_fields is not None: - body["required_fields"] = input.required_fields - if input.has_example_body: - body["example_body"] = input.example_body - if input.next_steps is not None: - body["next_steps"] = input.next_steps - body.update(input.extra) + if required_fields is not None: + body["required_fields"] = required_fields + if example_body is not _NO_EXAMPLE: + body["example_body"] = example_body + if next_steps is not None: + body["next_steps"] = next_steps + if extra: + body.update(extra) return body diff --git a/examples/multi_rail_merchant.py b/examples/multi_rail_merchant.py index 46e3cb5..ca02173 100644 --- a/examples/multi_rail_merchant.py +++ b/examples/multi_rail_merchant.py @@ -39,21 +39,7 @@ from fastapi.responses import JSONResponse from agentscore_commerce.challenge import ( - Build402BodyInput, - BuildAcceptedMethodsInput, - BuildAgentInstructionsInput, - BuildHowToPayInput, - BuildValidationErrorInput, - HowToPayRails, - Respond402Input, - SolanaMppConfig, - SolanaMppRailConfig, - StripeConfig, - StripeRailConfig, - TempoConfig, - TempoRailConfig, - X402BaseConfig, - X402BaseRailConfig, + build_402_body, build_accepted_methods, build_agent_instructions, build_how_to_pay, @@ -65,7 +51,6 @@ from agentscore_commerce.identity.fastapi import AgentScoreGate, get_agentscore_data from agentscore_commerce.payment import ( USDC, - PaymentRequiredHeaderInput, ProcessX402SettleInput, ValidateX402NetworkConfigInput, VerifyX402RequestInput, @@ -179,12 +164,10 @@ async def purchase(request: Request, assess: dict = Depends(get_agentscore_data) if not settle.success: return JSONResponse( build_validation_error( - BuildValidationErrorInput( - 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}, - ) + 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, ) @@ -215,73 +198,67 @@ async def purchase(request: Request, assess: dict = Depends(get_agentscore_data) pympx_challenge_headers = {"www-authenticate": 'Payment id="..."'} # from pympp.compose deposit_addresses = {"tempo": "0x...", "base": "0x...", "solana": "..."} # from create_multichain_payment_intent accepted = build_accepted_methods( - BuildAcceptedMethodsInput( - tempo=TempoConfig(recipient=deposit_addresses["tempo"]), - x402_base=X402BaseConfig(recipient=deposit_addresses["base"]), - solana_mpp=SolanaMppConfig(recipient=deposit_addresses["solana"]), - stripe=StripeConfig(profile_id=os.environ["STRIPE_PROFILE_ID"]), - ) + tempo={"recipient": deposit_addresses["tempo"]}, + x402_base={"recipient": deposit_addresses["base"]}, + solana_mpp={"recipient": deposit_addresses["solana"]}, + stripe={"profile_id": os.environ["STRIPE_PROFILE_ID"]}, ) how_to_pay = build_how_to_pay( - BuildHowToPayInput( - url=APP_URL, - retry_body_json=str(body), - total_usd=total_usd, - rails=HowToPayRails( - tempo=TempoRailConfig(recipient=deposit_addresses["tempo"]), - x402_base=X402BaseRailConfig(recipient=deposit_addresses["base"]), - solana_mpp=SolanaMppRailConfig(recipient=deposit_addresses["solana"]), - stripe=StripeRailConfig(profile_id=os.environ["STRIPE_PROFILE_ID"]), - ), - ) + url=APP_URL, + retry_body_json=str(body), + total_usd=total_usd, + rails={ + "tempo": {"recipient": deposit_addresses["tempo"]}, + "x402_base": {"recipient": deposit_addresses["base"]}, + "solana_mpp": {"recipient": deposit_addresses["solana"]}, + "stripe": {"profile_id": os.environ["STRIPE_PROFILE_ID"]}, + }, ) result = respond_402( - Respond402Input( - mppx_challenge_headers=pympx_challenge_headers, - body=Build402BodyInput( - accepted_methods=accepted, - agent_instructions=build_agent_instructions(BuildAgentInstructionsInput(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=PaymentRequiredHeaderInput( - 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, + 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 ), - { - "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"}, - ), - ) + "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) diff --git a/tests/test_challenge.py b/tests/test_challenge.py index b9197f6..1948a38 100644 --- a/tests/test_challenge.py +++ b/tests/test_challenge.py @@ -1,19 +1,6 @@ from agentscore_commerce.challenge import ( - Build402BodyInput, - BuildAcceptedMethodsInput, - BuildAgentInstructionsInput, - BuildHowToPayInput, - HowToPayRails, - IdentityMetadataInput, PricingBlock, SignerMatchResult, - SolanaMppConfig, - StripeConfig, - StripeRailConfig, - TempoConfig, - TempoRailConfig, - X402BaseConfig, - X402BaseRailConfig, X402PaymentRequired, build_402_body, build_accepted_methods, @@ -25,10 +12,8 @@ def test_build_accepted_methods_includes_only_provided_rails(): out = build_accepted_methods( - BuildAcceptedMethodsInput( - tempo=TempoConfig(recipient="0xT"), - stripe=StripeConfig(profile_id="acct_x"), - ) + tempo={"recipient": "0xT"}, + stripe={"profile_id": "acct_x"}, ) methods = [e["method"] for e in out] assert "tempo/charge" in methods @@ -38,20 +23,27 @@ def test_build_accepted_methods_includes_only_provided_rails(): def test_build_accepted_methods_full_set(): out = build_accepted_methods( - BuildAcceptedMethodsInput( - tempo=TempoConfig(recipient="0xT"), - x402_base=X402BaseConfig(recipient="0xB"), - solana_mpp=SolanaMppConfig(recipient="solanaaddr"), - stripe=StripeConfig(profile_id="acct_x"), - ) + tempo={"recipient": "0xT"}, + x402_base={"recipient": "0xB"}, + solana_mpp={"recipient": "solanaaddr"}, + stripe={"profile_id": "acct_x"}, ) assert len(out) == 4 assert out[1]["pay_to"] == "0xB" assert out[2]["network"].startswith("solana:") +def test_build_accepted_methods_overrides_defaults(): + """Per-rail dict values override the rail's protocol defaults inline.""" + out = build_accepted_methods( + tempo={"recipient": "0xT", "network": "tempo-testnet", "chain_id": 42431}, + ) + assert out[0]["network"] == "tempo-testnet" + assert out[0]["chain_id"] == 42431 + + def test_build_identity_metadata_wallet_mode_emits_signer_constraint(): - md = build_identity_metadata(IdentityMetadataInput(mode="wallet", wallet="0xClaim", linked_wallets=["0xSibling"])) + md = build_identity_metadata(mode="wallet", wallet="0xClaim", linked_wallets=["0xSibling"]) assert md["identity_mode"] == "wallet" assert md["required_signer"] == "0xClaim" assert md["linked_wallets"] == ["0xSibling"] @@ -60,32 +52,28 @@ def test_build_identity_metadata_wallet_mode_emits_signer_constraint(): def test_build_identity_metadata_signer_match_overrides_required(): md = build_identity_metadata( - IdentityMetadataInput( - mode="wallet", - wallet="0xClaim", - signer_match_result=SignerMatchResult(kind="pass", expected_signer="0xExpected"), - ) + mode="wallet", + wallet="0xClaim", + signer_match_result=SignerMatchResult(kind="pass", expected_signer="0xExpected"), ) assert md["required_signer"] == "0xExpected" def test_build_identity_metadata_token_mode_only_returns_mode(): - md = build_identity_metadata(IdentityMetadataInput(mode="operator_token")) + md = build_identity_metadata(mode="operator_token") assert md == {"identity_mode": "operator_token"} def test_build_how_to_pay_emits_per_rail_blocks(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com/buy", - retry_body_json='{"x":1}', - total_usd=10.0, - rails=HowToPayRails( - tempo=TempoRailConfig(recipient="0xT"), - x402_base=X402BaseRailConfig(recipient="0xB"), - stripe=StripeRailConfig(profile_id="acct_x"), - ), - ) + url="https://ex.com/buy", + retry_body_json='{"x":1}', + total_usd=10.0, + rails={ + "tempo": {"recipient": "0xT"}, + "x402_base": {"recipient": "0xB"}, + "stripe": {"profile_id": "acct_x"}, + }, ) assert "tempo" in out assert out["tempo"]["command"].startswith("tempo request") @@ -97,19 +85,29 @@ def test_build_how_to_pay_emits_per_rail_blocks(): def test_build_how_to_pay_blocks_link_cli_above_500(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com/buy", - retry_body_json="{}", - total_usd=750.0, - rails=HowToPayRails(stripe=StripeRailConfig(profile_id="acct_x")), - ) + url="https://ex.com/buy", + retry_body_json="{}", + total_usd=750.0, + rails={"stripe": {"profile_id": "acct_x"}}, ) assert "note" in out["stripe"] assert "setup_link_cli" not in out["stripe"] +def test_build_how_to_pay_recommend_kwarg_switches_command(): + """`recommend='agentscore-pay'` puts the pay CLI command as primary; `'tempo'` uses tempo request.""" + out = build_how_to_pay( + url="https://ex.com/buy", + retry_body_json="{}", + total_usd=5.0, + rails={"tempo": {"recipient": "0xT", "recommend": "agentscore-pay"}}, + ) + assert "agentscore-pay pay POST" in out["tempo"]["command"] + assert "tempo request" in out["tempo"]["alternative_command"] + + def test_build_agent_instructions_uses_defaults(): - out = build_agent_instructions(BuildAgentInstructionsInput(how_to_pay={"tempo": {}})) + out = build_agent_instructions(how_to_pay={"tempo": {}}) assert out["timeout_seconds"] == 300 assert any("agentscore-pay" in t for t in out["recommended_tools"]) assert any("tempo wallet transfer" in w for w in out["warnings"]) @@ -117,16 +115,16 @@ def test_build_agent_instructions_uses_defaults(): def test_build_agent_instructions_warnings_match_rails(): """Defaults adapt to which rails are present in how_to_pay.""" - x402_only = build_agent_instructions(BuildAgentInstructionsInput(how_to_pay={"x402_base": {}})) + x402_only = build_agent_instructions(how_to_pay={"x402_base": {}}) assert not any("tempo wallet transfer" in w for w in x402_only["warnings"]) assert any("x402 deposit addresses" in w for w in x402_only["warnings"]) assert not any("tempo request" in t for t in x402_only["recommended_tools"]) assert any("agentscore-pay" in t for t in x402_only["recommended_tools"]) - tempo_only = build_agent_instructions(BuildAgentInstructionsInput(how_to_pay={"tempo": {}})) + tempo_only = build_agent_instructions(how_to_pay={"tempo": {}}) assert not any("x402 deposit addresses" in w for w in tempo_only["warnings"]) - stripe_only = build_agent_instructions(BuildAgentInstructionsInput(how_to_pay={"stripe": {}})) + stripe_only = build_agent_instructions(how_to_pay={"stripe": {}}) assert stripe_only["warnings"] == [] assert stripe_only["recommended_tools"] == [] @@ -134,10 +132,8 @@ def test_build_agent_instructions_warnings_match_rails(): def test_build_agent_instructions_appends_extra_warnings(): """extra_warnings is appended to the rail-derived defaults.""" out = build_agent_instructions( - BuildAgentInstructionsInput( - how_to_pay={"tempo": {}, "x402_base": {}}, - extra_warnings=["Solana unavailable for this order; use base or tempo."], - ) + how_to_pay={"tempo": {}, "x402_base": {}}, + extra_warnings=["Solana unavailable for this order; use base or tempo."], ) assert len(out["warnings"]) == 3 assert "tempo wallet transfer" in out["warnings"][0] @@ -147,26 +143,22 @@ def test_build_agent_instructions_appends_extra_warnings(): def test_build_agent_instructions_extra_warnings_ignored_when_warnings_set(): """Explicit warnings override defaults AND extra_warnings.""" out = build_agent_instructions( - BuildAgentInstructionsInput( - how_to_pay={"tempo": {}}, - warnings=["custom only"], - extra_warnings=["ignored"], - ) + how_to_pay={"tempo": {}}, + warnings=["custom only"], + extra_warnings=["ignored"], ) assert out["warnings"] == ["custom only"] def test_build_402_body_assembles_full_response(): body = build_402_body( - Build402BodyInput( - accepted_methods=[{"method": "tempo/charge"}], - agent_instructions={"how_to_pay": {}}, - identity_metadata={"identity_mode": "wallet"}, - pricing=PricingBlock(subtotal="100", tax="8", tax_rate=0.08, tax_state="CA", total="108"), - amount_usd="108", - order_id="ord_1", - x402=X402PaymentRequired(accepts=[{}]), - ) + accepted_methods=[{"method": "tempo/charge"}], + agent_instructions={"how_to_pay": {}}, + identity_metadata={"identity_mode": "wallet"}, + pricing=PricingBlock(subtotal="100", tax="8", tax_rate=0.08, tax_state="CA", total="108"), + amount_usd="108", + order_id="ord_1", + x402=X402PaymentRequired(accepts=[{}]), ) assert body["payment_required"] is True assert body["x402Version"] == 2 @@ -178,13 +170,11 @@ def test_build_402_body_assembles_full_response(): def test_build_402_body_emits_v1_alias_on_accepts_entries(): """Each accepts entry carries both `amount` (v2) and `maxAmountRequired` (v1).""" body = build_402_body( - Build402BodyInput( - accepted_methods=[], - x402=X402PaymentRequired( - accepts=[{"scheme": "exact", "network": "eip155:84532", "amount": "110000"}], - version=2, - ), - ) + accepted_methods=[], + x402=X402PaymentRequired( + accepts=[{"scheme": "exact", "network": "eip155:84532", "amount": "110000"}], + version=2, + ), ) entry = body["accepts"][0] assert entry["amount"] == "110000" diff --git a/tests/test_coverage_fillers.py b/tests/test_coverage_fillers.py index c18e8d7..db935e8 100644 --- a/tests/test_coverage_fillers.py +++ b/tests/test_coverage_fillers.py @@ -1,15 +1,6 @@ """Targeted tests covering optional-field branches across discovery + challenge builders.""" from agentscore_commerce.challenge import ( - Build402BodyInput, - BuildAcceptedMethodsInput, - BuildAgentInstructionsInput, - BuildHowToPayInput, - HowToPayRails, - SolanaMppConfig, - SolanaMppRailConfig, - StripeRailConfig, - TempoRailConfig, build_402_body, build_accepted_methods, build_agent_instructions, @@ -116,19 +107,17 @@ def test_llms_txt_payment_section_includes_all_rails(): def test_build_accepted_methods_includes_solana_only(): - out = build_accepted_methods(BuildAcceptedMethodsInput(solana_mpp=SolanaMppConfig(recipient="solanaaddr"))) + out = build_accepted_methods(solana_mpp={"recipient": "solanaaddr"}) assert out[0]["network"].startswith("solana:") assert out[0]["pay_to"] == "solanaaddr" def test_build_how_to_pay_solana_only(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com", - retry_body_json="{}", - total_usd=5.0, - rails=HowToPayRails(solana_mpp=SolanaMppRailConfig(recipient="solanaaddr")), - ) + url="https://ex.com", + retry_body_json="{}", + total_usd=5.0, + rails={"solana_mpp": {"recipient": "solanaaddr"}}, ) assert "solana_mpp" in out assert "agentscore-pay pay POST" in out["solana_mpp"]["command"] @@ -136,12 +125,10 @@ def test_build_how_to_pay_solana_only(): def test_build_how_to_pay_tempo_recommend_pay(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com", - retry_body_json="{}", - total_usd=5.0, - rails=HowToPayRails(tempo=TempoRailConfig(recipient="0xT", recommend="agentscore-pay")), - ) + url="https://ex.com", + retry_body_json="{}", + total_usd=5.0, + rails={"tempo": {"recipient": "0xT", "recommend": "agentscore-pay"}}, ) assert out["tempo"]["command"].startswith("agentscore-pay pay POST") assert out["tempo"]["alternative_command"].startswith("tempo request") @@ -149,24 +136,20 @@ def test_build_how_to_pay_tempo_recommend_pay(): def test_build_how_to_pay_tempo_recommend_tempo_only(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com", - retry_body_json="{}", - total_usd=5.0, - rails=HowToPayRails(tempo=TempoRailConfig(recipient="0xT", recommend="tempo")), - ) + url="https://ex.com", + retry_body_json="{}", + total_usd=5.0, + rails={"tempo": {"recipient": "0xT", "recommend": "tempo"}}, ) assert "alternative_command" not in out["tempo"] def test_build_how_to_pay_stripe_no_profile_id_skips_link_cli(): out = build_how_to_pay( - BuildHowToPayInput( - url="https://ex.com", - retry_body_json="{}", - total_usd=5.0, - rails=HowToPayRails(stripe=StripeRailConfig(profile_id=None)), - ) + url="https://ex.com", + retry_body_json="{}", + total_usd=5.0, + rails={"stripe": {"profile_id": None}}, ) assert "setup_link_cli" not in out["stripe"] assert "note" not in out["stripe"] @@ -174,11 +157,9 @@ def test_build_how_to_pay_stripe_no_profile_id_skips_link_cli(): def test_build_agent_instructions_with_recommended_and_extra(): out = build_agent_instructions( - BuildAgentInstructionsInput( - how_to_pay={"tempo": {}}, - recommended="tempo", - extra={"vendor_field": "value"}, - ) + how_to_pay={"tempo": {}}, + recommended="tempo", + extra={"vendor_field": "value"}, ) assert out["recommended"] == "tempo" assert out["vendor_field"] == "value" @@ -186,15 +167,13 @@ def test_build_agent_instructions_with_recommended_and_extra(): def test_build_402_body_includes_all_optional_blocks(): body = build_402_body( - Build402BodyInput( - accepted_methods=[], - agent_memory={"pattern": "agentscore-shared-identity"}, - currency="USD", - product={"id": "p_1", "name": "Wine"}, - recommended="tempo", - retry_body={"product_id": "p_1"}, - extra={"vendor_field": "value"}, - ) + accepted_methods=[], + agent_memory={"pattern": "agentscore-shared-identity"}, + currency="USD", + product={"id": "p_1", "name": "Wine"}, + recommended="tempo", + retry_body={"product_id": "p_1"}, + extra={"vendor_field": "value"}, ) assert body["agent_memory"] == {"pattern": "agentscore-shared-identity"} assert body["currency"] == "USD" diff --git a/tests/test_lifted_helpers.py b/tests/test_lifted_helpers.py index 7e17203..fbb1a79 100644 --- a/tests/test_lifted_helpers.py +++ b/tests/test_lifted_helpers.py @@ -10,14 +10,9 @@ import pytest -from agentscore_commerce.challenge import ( - Build402BodyInput, - Respond402Input, - respond_402, -) +from agentscore_commerce.challenge import build_402_body, respond_402 from agentscore_commerce.payment import ( X402_SUPPORTED_BASE_NETWORKS, - PaymentRequiredHeaderInput, ProcessX402SettleFailure, ProcessX402SettleInput, ProcessX402SettleSuccess, @@ -140,13 +135,11 @@ def test_stripe_test_tx_hashes_documented(): def test_respond_402_preserves_mppx_www_authenticate(): result = respond_402( - Respond402Input( - mppx_challenge_headers={ - "WWW-Authenticate": 'Payment id="ord_x", method="tempo", request="..."', - "Content-Type": "application/json", - }, - body=Build402BodyInput(accepted_methods=[{"method": "tempo/charge"}]), - ) + mppx_challenge_headers={ + "WWW-Authenticate": 'Payment id="ord_x", method="tempo", request="..."', + "Content-Type": "application/json", + }, + body=build_402_body(accepted_methods=[{"method": "tempo/charge"}]), ) assert result.status == 402 assert "tempo" in result.headers["www-authenticate"] @@ -158,15 +151,13 @@ def test_respond_402_preserves_mppx_www_authenticate(): def test_respond_402_layers_payment_required_when_x402_set(): result = respond_402( - Respond402Input( - mppx_challenge_headers={"www-authenticate": 'Payment id="ord_y"'}, - body=Build402BodyInput(accepted_methods=[]), - x402=PaymentRequiredHeaderInput( - x402_version=2, - accepts=[{"scheme": "exact", "network": "eip155:84532"}], - resource={"url": "https://x.example/y", "mimeType": "application/json"}, - ), - ) + mppx_challenge_headers={"www-authenticate": 'Payment id="ord_y"'}, + body=build_402_body(accepted_methods=[]), + x402={ + "x402_version": 2, + "accepts": [{"scheme": "exact", "network": "eip155:84532"}], + "resource": {"url": "https://x.example/y", "mimeType": "application/json"}, + }, ) assert "payment-required" in result.headers decoded = json.loads(base64.b64decode(result.headers["payment-required"]).decode()) diff --git a/tests/test_validation_error.py b/tests/test_validation_error.py index ce688f1..3f17b38 100644 --- a/tests/test_validation_error.py +++ b/tests/test_validation_error.py @@ -1,13 +1,10 @@ """Tests for the lifted ``build_validation_error`` helper.""" -from agentscore_commerce.challenge import ( - BuildValidationErrorInput, - build_validation_error, -) +from agentscore_commerce.challenge import build_validation_error def test_minimal_body_only_code_and_message() -> None: - body = build_validation_error(BuildValidationErrorInput(code="bad_request", message="Missing fields")) + body = build_validation_error(code="bad_request", message="Missing fields") assert body == {"error": {"code": "bad_request", "message": "Missing fields"}} assert "required_fields" not in body assert "next_steps" not in body @@ -16,13 +13,10 @@ def test_minimal_body_only_code_and_message() -> None: def test_includes_required_fields_and_example_body() -> None: body = build_validation_error( - BuildValidationErrorInput( - code="bad_request", - message="product_id and email are required", - required_fields={"product_id": "uuid", "email": "string"}, - example_body={"product_id": "abc", "email": "a@b.c"}, - has_example_body=True, - ) + code="bad_request", + message="product_id and email are required", + required_fields={"product_id": "uuid", "email": "string"}, + example_body={"product_id": "abc", "email": "a@b.c"}, ) assert body["required_fields"] == {"product_id": "uuid", "email": "string"} assert body["example_body"] == {"product_id": "abc", "email": "a@b.c"} @@ -30,35 +24,32 @@ def test_includes_required_fields_and_example_body() -> None: def test_includes_next_steps_with_arbitrary_keys() -> None: body = build_validation_error( - BuildValidationErrorInput( - code="not_found", - message="Product not found", - next_steps={"action": "fetch_catalog", "catalog_url": "https://example.com/catalog"}, - ) + code="not_found", + message="Product not found", + next_steps={"action": "fetch_catalog", "catalog_url": "https://example.com/catalog"}, ) assert body["next_steps"] == {"action": "fetch_catalog", "catalog_url": "https://example.com/catalog"} def test_merges_extra_top_level_fields() -> None: body = build_validation_error( - BuildValidationErrorInput( - code="out_of_stock", - message="Insufficient quantity", - extra={"available": 3, "max_length": 300}, - ) + code="out_of_stock", + message="Insufficient quantity", + extra={"available": 3, "max_length": 300}, ) assert body["available"] == 3 assert body["max_length"] == 300 -def test_omits_example_body_when_has_example_body_false_default() -> None: - body = build_validation_error(BuildValidationErrorInput(code="x", message="y")) +def test_omits_example_body_when_not_passed() -> None: + """When example_body kwarg is omitted, the field is suppressed in the body.""" + body = build_validation_error(code="x", message="y") assert "example_body" not in body -def test_emits_explicit_null_example_body_when_has_example_body_true() -> None: - body = build_validation_error( - BuildValidationErrorInput(code="x", message="y", example_body=None, has_example_body=True) - ) +def test_emits_explicit_null_example_body_when_passed_as_none() -> None: + """Passing example_body=None explicitly emits a literal null in the body + (distinguished from "field omitted" via the sentinel default).""" + body = build_validation_error(code="x", message="y", example_body=None) assert "example_body" in body assert body["example_body"] is None