AP2 mandate flows

samples/python/scenarios/a2a/human-present/cards/run.sh

@@ -130,6 +130,10 @@ echo "-> Starting the Card Processor Agent (port:8003 log:$LOG_DIR/mpp_agent.log
 $UV_RUN_CMD --package ap2-samples python -m roles.merchant_payment_processor_agent >"$LOG_DIR/mpp_agent.log" 2>&1 &
 pids+=($!)
 
+echo "-> Starting the OneID Identity Provider (port:8004 log:$LOG_DIR/oneid_identity_provider.log)..."
+$UV_RUN_CMD --package ap2-samples python -m roles.oneid_identity_provider >"$LOG_DIR/oneid_identity_provider.log" 2>&1 &
+pids+=($!)
+
 echo ""
 echo "All remote servers are starting."
 

samples/python/src/roles/credentials_provider_agent/tools.py

@@ -88,7 +88,25 @@ async def handle_search_payment_methods(
         "user_email is required for search_payment_methods"
     )
   if not method_data:
-    raise ValueError("method_data is required for search_payment_methods")
+    # AP2 EXTENSION (HOOTL flow): The AP2 spec does not define how a
+    # Credentials Provider should respond when no CartMandate is present
+    # (i.e. when the user is pre-authorising a payment method at delegation
+    # time rather than at checkout). We return all of the user's payment
+    # methods unfiltered so the shopping agent can present them for selection.
+    # In a production system a separate "list payment methods" endpoint would
+    # be more appropriate than overloading the merchant-filtered search.
+    # See: https://ap2-protocol.org/specification/#52-human-not-present-transaction
+    payment_methods = account_manager.get_account_payment_methods(user_email)
+    if os.environ.get("PAYMENT_METHOD") == "x402":
+      payment_methods = [m for m in payment_methods if m.get("brand") == "x402"]
+    else:
+      payment_methods = [m for m in payment_methods if m.get("brand") != "x402"]
+    aliases = {
+        "payment_method_aliases": _get_payment_method_aliases(payment_methods)
+    }
+    await updater.add_artifact([Part(root=DataPart(data=aliases))])
+    await updater.complete()
+    return
 
   merchant_method_data_list = [
       PaymentMethodData.model_validate(data) for data in method_data

samples/python/src/roles/merchant_agent/sub_agents/catalog_agent.py

@@ -46,6 +46,26 @@
 from common.system_utils import DEBUG_MODE_INSTRUCTIONS
 
 
+# Keywords used to detect age-restricted items in LLM-generated product labels.
+# This is a demo heuristic — in a production system the merchant's catalogue
+# would carry explicit age-restriction metadata rather than relying on keyword
+# matching.
+# AP2 EXTENSION: The AP2 spec defines no mechanism for merchants to signal
+# verification requirements within a cart. We set verification_requirements
+# on CartContents — the object the merchant cryptographically signs — so the
+# requirement is bound to the offer and travels with it. This follows the
+# precedent of user_cart_confirmation_required and is proposed as a natural
+# typed extension to the CartContents schema.
+# See: https://ap2-protocol.org/topics/core-concepts/#cart-mandate
+_AGE_RESTRICTED_KEYWORDS = frozenset({
+    "wine", "beer", "ale", "lager", "stout", "cider",
+    "whisky", "whiskey", "vodka", "gin", "rum", "brandy",
+    "champagne", "prosecco", "spirit", "liqueur", "bourbon",
+    "alcohol", "tobacco", "cigarette", "cigar", "vape",
+    "age 18+", "[age 18+ required]",
+})
+
+
 async def find_items_workflow(
     data_parts: list[dict[str, Any]],
     updater: TaskUpdater,
@@ -64,6 +84,10 @@ async def find_items_workflow(
 
         You MUST exclude all branding from the PaymentItem `label` field.
 
+        If any item is typically age-restricted (alcohol, wine, beer, spirits,
+        tobacco, or cigarettes), you MUST append ' [Age 18+ required]' to that
+        item's label field.
+
     %s
         """ % DEBUG_MODE_INSTRUCTIONS
 
@@ -77,8 +101,18 @@ async def find_items_workflow(
   )
   try:
     items: list[PaymentItem] = llm_response.parsed
-
     current_time = datetime.now(timezone.utc)
+
+    # Dual-layer age restriction detection: the LLM was instructed to append
+    # '[Age 18+ required]' to restricted item labels; we also verify here with
+    # a deterministic keyword check so the flag is reliable regardless of LLM
+    # compliance. Both checks run — the label provides user-visible context.
+    age_verification_required = any(
+        any(kw in item.label.lower() for kw in _AGE_RESTRICTED_KEYWORDS)
+        for item in items
+    )
+    verification_requirements = ["age_over_18"] if age_verification_required else None
+
     item_count = 0
     for item in items:
       item_count += 1
@@ -88,7 +122,9 @@ async def find_items_workflow(
           current_time,
           updater,
           os.environ.get("PAYMENT_METHOD", "CARD"),
+          verification_requirements,
       )
+
     risk_data = _collect_risk_data(updater)
     updater.add_artifact([
         Part(root=DataPart(data={"risk_data": risk_data})),
@@ -108,6 +144,7 @@ async def _create_and_add_cart_mandate_artifact(
     current_time: datetime,
     updater: TaskUpdater,
     payment_method: str,
+    verification_requirements: list[str] | None = None,
 ) -> None:
   """Creates a CartMandate and adds it as an artifact."""
   if payment_method == "x402":
@@ -154,6 +191,7 @@ async def _create_and_add_cart_mandate_artifact(
       payment_request=payment_request,
       cart_expiry=(current_time + timedelta(minutes=30)).isoformat(),
       merchant_name="Generic Merchant",
+      verification_requirements=verification_requirements,
   )
 
   cart_mandate = CartMandate(contents=cart_contents)

samples/python/src/roles/merchant_payment_processor_agent/tools.py

@@ -18,8 +18,10 @@
 shopping and purchasing process.
 """
 
+import base64
 from datetime import datetime
 from datetime import timezone
+import json
 import logging
 import os
 from typing import Any
@@ -93,8 +95,10 @@ async def _handle_payment_mandate(
     debug_mode: Whether the agent is in debug mode.
     payment_method: The payment method to use (e.g., 'CARD', 'x402').
   """
+  identity_assertion = _extract_identity_assertion(payment_mandate)
+
   if current_task is None:
-    await _raise_challenge(updater)
+    await _raise_challenge(updater, identity_assertion=identity_assertion)
     return
 
   if current_task.status.state == TaskState.input_required:
@@ -110,16 +114,56 @@ async def _handle_payment_mandate(
 
 async def _raise_challenge(
     updater: TaskUpdater,
+    identity_assertion: dict | None = None,
 ) -> None:
   """Raises a transaction challenge.
 
   This challenge would normally be raised by the issuer, but we don't
   have an issuer in the demo, so we raise the challenge here. For concreteness,
   we are using an OTP challenge in this sample.
 
+  If a OneID identity assertion is present in the PaymentMandate, it is
+  surfaced here to show what the payment processor / issuer can see and
+  how they might use it in a real authorisation decision.
+
   Args:
     updater: The task updater.
+    identity_assertion: Optional OneID identity credential extracted from
+      PaymentMandate.user_authorization, if present.
   """
+  parts = []
+
+  if identity_assertion:
+    subject = identity_assertion.get("credential_subject", {})
+    assurance = subject.get("assurance_level", "")
+    framework = subject.get("framework", "")
+    # Assess whether the trust signal is sufficient for frictionless auth.
+    # In a real bilateral agreement, the issuer's risk engine would make
+    # this determination. Here we simulate the decision criteria.
+    frictionless_eligible = assurance == "high" and framework == "UK_DIATF"
+    frictionless_note = (
+        "This transaction carries a verified identity credential with high "
+        "assurance under UK_DIATF. Under a bilateral agreement with OneID, "
+        "an issuer COULD choose to approve this transaction frictionlessly "
+        "without an OTP challenge. In this demo, standard OTP applies."
+        if frictionless_eligible
+        else (
+            "No OneID credential with sufficient assurance was found. "
+            "Standard OTP challenge required."
+        )
+    )
+    identity_data = {
+        "issuer": identity_assertion.get("issuer"),
+        "verified_scopes": subject.get("verified_scopes", []),
+        "assurance_level": assurance,
+        "framework": framework,
+        "frictionless_eligible": frictionless_eligible,
+        "frictionless_note": frictionless_note,
+    }
+    parts.append(
+        Part(root=DataPart(data={"oneid.IdentityAssertion": identity_data}))
+    )
+
   challenge_data = {
       "type": "otp",
       "display_text": (
@@ -129,13 +173,12 @@ async def _raise_challenge(
           "(Demo only hint: the code is 123)"
       ),
   }
-  text_part = TextPart(
+  parts.append(Part(root=TextPart(
       text="Please provide the challenge response to complete the payment."
-  )
-  data_part = DataPart(data={"challenge": challenge_data})
-  message = updater.new_agent_message(
-      parts=[Part(root=text_part), Part(root=data_part)]
-  )
+  )))
+  parts.append(Part(root=DataPart(data={"challenge": challenge_data})))
+
+  message = updater.new_agent_message(parts=parts)
   await updater.requires_input(message=message)
 
 
@@ -373,6 +416,27 @@ async def _send_payment_receipt_to_credentials_provider(
   await credentials_provider.send_a2a_message(message_builder.build())
 
 
+def _extract_identity_assertion(payment_mandate: PaymentMandate) -> dict | None:
+  """Extracts the OneID identity credential from PaymentMandate.user_authorization.
+
+  The user_authorization field carries a base64url-encoded JSON payload
+  produced by sign_mandates_on_user_device, binding the mandate hashes to
+  any OneID identity credential obtained during the shopping session.
+
+  Returns the identity_credential dict if present, or None.
+  """
+  user_auth = payment_mandate.user_authorization
+  if not user_auth:
+    return None
+  try:
+    # Restore stripped base64 padding before decoding.
+    padded = user_auth + "=" * (4 - len(user_auth) % 4)
+    payload = json.loads(base64.urlsafe_b64decode(padded))
+    return payload.get("identity_credential")
+  except Exception:
+    return None
+
+
 def _create_text_parts(*texts: str) -> list[Part]:
   """Helper to create text parts."""
   return [Part(root=TextPart(text=text)) for text in texts]

samples/python/src/roles/oneid_identity_provider/__main__.py

@@ -0,0 +1,23 @@
+"""OneID Identity Provider Agent — port 8004."""
+
+from collections.abc import Sequence
+
+from absl import app
+from roles.oneid_identity_provider.agent_executor import OneIDIdentityProviderExecutor
+from common import server
+
+AGENT_PORT = 8004
+
+
+def main(argv: Sequence[str]) -> None:
+  agent_card = server.load_local_agent_card(__file__)
+  server.run_agent_blocking(
+      port=AGENT_PORT,
+      agent_card=agent_card,
+      executor=OneIDIdentityProviderExecutor(agent_card.capabilities.extensions),
+      rpc_url="/a2a/oneid_identity_provider",
+  )
+
+
+if __name__ == "__main__":
+  app.run(main)

samples/python/src/roles/oneid_identity_provider/agent.json

@@ -0,0 +1,25 @@
+{
+  "name": "OneIDIdentityProvider",
+  "description": "An agent that verifies user identity using bank-verified data via OneID.",
+  "capabilities": {
+    "extensions": [
+      {
+        "uri": "https://github.com/google-agentic-commerce/ap2/v1",
+        "description": "Supports the Agent Payments Protocol.",
+        "required": true
+      }
+    ]
+  },
+  "skills": [
+    {
+      "id": "identity-verification",
+      "name": "Identity Verification",
+      "description": "Verifies user identity and age using UK bank authentication. Returns verified claims as a JWT Verifiable Credential.",
+      "tags": ["identity", "age-verification", "kyc"]
+    }
+  ],
+  "defaultInputModes": ["text/plain"],
+  "defaultOutputModes": ["application/json"],
+  "url": "http://localhost:8004/a2a/oneid_identity_provider",
+  "version": "1.0.0"
+}

samples/python/src/roles/oneid_identity_provider/agent_executor.py

@@ -0,0 +1,41 @@
+"""An A2A Agent Executor for the OneID Identity Provider agent.
+
+OneID verifies user identity via UK Open Banking: the user authenticates with
+their bank (biometrics/PIN) and OneID returns verified claims including age
+thresholds, as a JWT Verifiable Credential.
+
+In order to clearly demonstrate the use of the Agent Payments Protocol A2A
+extension, this agent was built directly using the A2A framework.
+"""
+
+from typing import Any
+
+from . import tools
+from common.base_server_executor import BaseServerExecutor
+from common.system_utils import DEBUG_MODE_INSTRUCTIONS
+
+
+class OneIDIdentityProviderExecutor(BaseServerExecutor):
+  """AgentExecutor for the OneID identity provider agent."""
+
+  _system_prompt = """
+    You are OneID, a UK identity verification provider. Your job is to verify
+    user identity using UK Open Banking authentication and return verified
+    claims as a Verifiable Credential.
+
+    Based on the user's request, identify their intent and select the single
+    correct tool to use. Your only output should be a tool call.
+    Do not engage in conversation.
+
+    %s
+  """ % DEBUG_MODE_INSTRUCTIONS
+
+  def __init__(self, supported_extensions: list[dict[str, Any]] = None):
+    """Initializes the OneIDIdentityProviderExecutor.
+
+    Args:
+        supported_extensions: A list of extension objects supported by the
+          agent.
+    """
+    agent_tools = [tools.handle_verify_identity]
+    super().__init__(supported_extensions, agent_tools, self._system_prompt)