fix(server): read the A2A profile's input member, and refuse parameters when the profile is activated

src/adcp/reporting/receipts/transport.py

@@ -158,7 +158,9 @@ def a2a_receipt_parameters(body: bytes | None, *, task: str = TASK) -> dict[str,
     if invocation is None:
         return None
     try:
-        params = invocation.get("parameters")
+        # The same member the A2A dispatcher reads: legacy ``parameters`` when
+        # present, else the AdCP A2A profile's ``input``; never a merge.
+        params = invocation.get("parameters" if "parameters" in invocation else "input")
         if type(params) is dict:
             return _parameters(params, task)
     except (ValueError, TypeError, RecursionError):

src/adcp/server/a2a_server.py

@@ -75,7 +75,9 @@
 A2A :class:`RequestContext`.
 
 The default parser handles a DataPart (``data`` oneof) carrying
-``{"skill": ..., "parameters": ...}`` plus a TextPart JSON fallback.
+``{"skill": ..., "input": ...}`` (the AdCP A2A profile) or the legacy
+``{"skill": ..., "parameters": ...}``, plus a TextPart JSON fallback; see
+:func:`_invocation_arguments` for which member is read.
 Override this hook to accept alternative wire shapes — JSON-RPC 2.0
 message bodies, vendor-specific DataPart schemas, or text-only skill
 encodings. Return ``(None, {})`` to signal "no parseable skill"; the
@@ -201,6 +203,9 @@ def _get_decisioning_adcp_error_types() -> tuple[type[BaseException], ...]:
     {"kind", "text", "data", "raw", "url", "file", "filename", "mediaType", "metadata"}
 )
 _DATA_MEMBERS = frozenset({"skill", "input", "parameters"})
+#: AdCP A2A profile (``a2a-profile-extension.mdx``). A request activates it by
+#: naming this URI in the ``A2A-Extensions`` service parameter.
+_ADCP_A2A_PROFILE_URI = "https://adcontextprotocol.org/extensions/adcp/v3"
 # A2A 1.0 Part content is a oneof of ``text``/``data``/``raw``/``url``; A2A 0.3
 # Parts carry ``text``/``data``/``file``. ``raw``/``url``/``file`` are FileParts.
 _PART_CONTENT_MEMBERS = ("text", "data", "raw", "url", "file")
@@ -605,6 +610,61 @@ def _normalize_a2a_parameters(params: Any) -> dict[str, Any]:
     return normalized
 
 
+def _invocation_arguments(data: dict[str, Any]) -> dict[str, Any]:
+    """The task arguments of one invocation object: ``parameters`` or ``input``.
+
+    One member is read, never a merge of both. ``parameters`` wins when the
+    key is present, so a caller that does not activate the AdCP A2A profile
+    (A2A 0.x clients, earlier SDKs) is dispatched exactly as before; otherwise
+    the profile's ``input`` member is read. The executor refuses a
+    profile-activated request that carries ``parameters`` at all
+    (:func:`_profile_invocation_violation`), so every request it accepts under
+    the profile is dispatched from ``input``. The choice depends only on the
+    invocation object, so the auth and signing middleware, which parse without
+    the request headers, see the same arguments as dispatch.
+    """
+    member = "parameters" if "parameters" in data else "input"
+    return _normalize_a2a_parameters(data.get(member, {}))
+
+
+def _profile_invocation_violation(context: RequestContext) -> str | None:
+    """Why a profile-activated message breaks the invocation shape, else ``None``.
+
+    Under the AdCP A2A profile a Message carries exactly one DataPart whose
+    object has an object ``input``. ``parameters`` "is not an alias for
+    ``input``", and the agent "MUST reject the invocation rather than choose
+    among ambiguous inputs" — so ``parameters``, alone or beside ``input``,
+    is refused. A request that does not activate the profile returns ``None``
+    and keeps the legacy shapes.
+
+    The profile is activated by the ``A2A-Extensions`` service parameter; the
+    Message's own ``extensions`` list naming the URI counts too. Member names
+    are read from the protobuf ``Value`` directly, so a number the JSON
+    printer refuses cannot fail this check.
+    """
+    message = context.message
+    # ``execute`` is public: a hand-built context (an adopter's executor test)
+    # may lack the a2a-sdk attributes, and then has not activated the profile.
+    requested = getattr(context, "requested_extensions", None) or ()
+    activated = _ADCP_A2A_PROFILE_URI in requested or (
+        message is not None and _ADCP_A2A_PROFILE_URI in getattr(message, "extensions", ())
+    )
+    if not activated:
+        return None
+    parts = message.parts if message is not None else []
+    data = [part.data for part in parts if part.WhichOneof("content") == "data"]
+    if len(data) != 1:
+        return "data_part_count"
+    if data[0].WhichOneof("kind") != "struct_value":
+        return "data_not_object"
+    fields = data[0].struct_value.fields
+    if "parameters" in fields:
+        return "parameters_member"
+    if "input" not in fields or fields["input"].WhichOneof("kind") != "struct_value":
+        return "input_missing"
+    return None
+
+
 def _default_parse_request(context: RequestContext) -> tuple[str | None, dict[str, Any]]:
     """Extract one unambiguous ADCP skill invocation from an A2A message.
 
@@ -623,7 +683,7 @@ def _default_parse_request(context: RequestContext) -> tuple[str | None, dict[st
         if data is not None:
             skill = data.get("skill")
             if skill:
-                parsed.append((str(skill), _normalize_a2a_parameters(data.get("parameters", {}))))
+                parsed.append((str(skill), _invocation_arguments(data)))
             continue
 
         text = _part_text(part)
@@ -634,12 +694,7 @@ def _default_parse_request(context: RequestContext) -> tuple[str | None, dict[st
         except (json.JSONDecodeError, TypeError):
             continue
         if isinstance(data, dict) and data.get("skill"):
-            parsed.append(
-                (
-                    str(data["skill"]),
-                    _normalize_a2a_parameters(data.get("parameters", {})),
-                )
-            )
+            parsed.append((str(data["skill"]), _invocation_arguments(data)))
 
     if len(parsed) != 1:
         return None, {}
@@ -712,8 +767,14 @@ class ADCPAgentExecutor(AgentExecutor):
     parameters, dispatched to the matching handler method, and the result
     is published back as A2A Task events.
 
-    Expects the explicit skill invocation format used by A2AAdapter:
+    Expects an explicit skill invocation DataPart, in the AdCP A2A profile's
+    shape or the legacy shape A2AAdapter sends:
+        Part(data={"skill": "get_products", "input": {...}})
         Part(data={"skill": "get_products", "parameters": {...}})
+
+    A request that activates the profile (``A2A-Extensions`` names
+    ``https://adcontextprotocol.org/extensions/adcp/v3``) must use ``input``;
+    one carrying ``parameters`` is refused before handler entry.
     """
 
     def __init__(
@@ -881,6 +942,29 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
             )
             return
 
+        violation = _profile_invocation_violation(context) if self._message_parser is None else None
+        if violation is not None:
+            # The AdCP A2A profile refuses an ambiguous or legacy-shaped
+            # activated invocation before handler entry. Nothing is echoed:
+            # echoing ``context`` from either member would choose between them.
+            from adcp.decisioning.types import AdcpError
+
+            logger.info("a2a dispatch rejected", extra={"reason": violation})
+            await self._send_adcp_error(
+                event_queue,
+                context,
+                AdcpError(
+                    "INVALID_REQUEST",
+                    message=(
+                        "An invocation that activates the AdCP A2A profile must carry "
+                        "exactly one DataPart with an `input` object and no `parameters`"
+                    ),
+                    recovery="correctable",
+                ),
+                skill_name=skill_name or "",
+            )
+            return
+
         if skill_name is None:
             await self._send_error(event_queue, context, "No skill specified in message")
             return
@@ -1120,7 +1204,8 @@ def _default_parse_request(self, context: RequestContext) -> tuple[str | None, d
         """Built-in parser. Supports two formats:
 
         1. Explicit skill invocation via a DataPart:
-           ``Part(data={"skill": "get_products", "parameters": {...}})``
+           ``Part(data={"skill": "get_products", "input": {...}})``, or the
+           legacy ``"parameters"`` member (see :func:`_invocation_arguments`)
         2. Natural language fallback via TextPart (best-effort parse)
 
         Exposed as a module-level method so custom parsers can compose
@@ -1134,7 +1219,7 @@ def _parse_text_request(self, text: str) -> tuple[str | None, dict[str, Any]]:
         try:
             data = json.loads(text)
             if isinstance(data, dict) and "skill" in data:
-                return str(data["skill"]), _normalize_a2a_parameters(data.get("parameters", {}))
+                return str(data["skill"]), _invocation_arguments(data)
         except (json.JSONDecodeError, TypeError):
             pass
         return None, {}
@@ -1769,8 +1854,12 @@ def create_a2a_server(
             audit hooks need.
         message_parser: Optional :data:`MessageParser` for alternative
             wire shapes. The default parser handles a DataPart carrying
+            ``{"skill": ..., "input": ...}`` or the legacy
             ``{"skill": ..., "parameters": ...}`` plus a TextPart JSON
-            fallback. Supply this to accept JSON-RPC 2.0 message bodies,
+            fallback, and the executor enforces the AdCP A2A profile's
+            invocation shape on requests that activate it. A custom parser
+            owns its wire shapes, so that check is skipped. Supply this to
+            accept JSON-RPC 2.0 message bodies,
             vendor-specific DataPart schemas, or other layouts. The
             callable returns ``(skill_name, params)`` or ``(None, {})``
             for "no parseable skill"; see :data:`MessageParser` and