Module adcp.protocols

class A2AAdapter(ProtocolAdapter):
    """Adapter for A2A protocol using the official a2a-sdk 1.0 client."""

    # A2A task states in which the server is still expecting more from
    # the buyer on the same task (input-required, auth-required, and
    # in-flight states). While the adapter holds a task_id in one of
    # these states, the next outbound Message must echo it back so the
    # server resumes the same task rather than orphaning it and starting
    # a new one. Everything else — completed/failed/canceled/rejected
    # (terminal) and the defensive unknown state — clears the retained
    # task_id so subsequent calls start a fresh task. The frozenset
    # holds protobuf enum int values so a rename upstream is a load-time
    # error, not a silent behavior change.
    _NONTERMINAL_TASK_STATES: frozenset[int] = frozenset(
        {
            pb.TaskState.TASK_STATE_SUBMITTED,
            pb.TaskState.TASK_STATE_WORKING,
            pb.TaskState.TASK_STATE_INPUT_REQUIRED,
            pb.TaskState.TASK_STATE_AUTH_REQUIRED,
        }
    )

    def __init__(
        self,
        agent_config: AgentConfig,
        force_a2a_version: str | None = None,
    ):
        """Initialize A2A adapter with official A2A client.

        ``force_a2a_version`` pins the A2A wire version by filtering the
        peer's advertised ``supported_interfaces`` to only entries whose
        ``protocol_version`` matches. Intended for tests or for forcing
        a 0.3-speaking path against a dual-advertising peer. Raises
        :class:`ADCPConnectionError` on first use if no advertised
        interface matches. ``None`` lets the SDK's ``ClientFactory``
        pick the most capable transport the peer supports.
        """
        super().__init__(agent_config)
        self._httpx_client: httpx.AsyncClient | None = None
        self._a2a_client: Client | None = None
        self._cached_agent_card: pb.AgentCard | None = None
        self._force_a2a_version = force_a2a_version
        # A2A contextId for multi-turn conversations. First request sends
        # context_id=None → server mints one and returns it on Task.context_id;
        # we stash it here and echo it back on every subsequent send so the
        # server can scope state to the same session. Callers can seed this
        # via ADCPClient(context_id=...) to resume a session across process
        # restarts, or clear it via ADCPClient.reset_context() to start a
        # new conversation.
        self._context_id: str | None = None
        # A2A task_id retained across turns only while the prior task is
        # non-terminal (input-required, working, etc). On terminal states
        # this clears to None so the next call starts a new task under
        # the same context_id. Without this, resume of an input-required
        # task orphans the server-side in-flight task.
        self._active_task_id: str | None = None

    @property
    def context_id(self) -> str | None:
        """Current A2A conversation context_id, or None if not yet established.

        ``None`` means either (a) a fresh conversation where the server
        has not yet replied, or (b) the context was cleared via
        ``set_context_id(None)``. Callers that need to distinguish these
        must track their own state.

        Not thread-safe: the adapter mutates this on every response. For
        concurrent use, serialize calls on one adapter or construct one
        per conversation.
        """
        return self._context_id

    @property
    def active_task_id(self) -> str | None:
        """A2A task_id the next send must echo to resume the same task.

        Populated when the last response was non-terminal (e.g.
        ``input-required``, ``working``). Echoed on the next outbound
        message so the server continues the same task. Clears to None
        on terminal states (``completed``/``failed``/``canceled``/
        ``rejected``) — and defensively on ``unknown`` — so subsequent
        calls start a fresh task under the same context.
        """
        return self._active_task_id

    @property
    def a2a_protocol_versions(self) -> list[str] | None:
        """Sorted list of A2A ``protocol_version`` strings the peer advertises.

        Populated after the first call (or any operation that fetches
        the ``AgentCard`` — :meth:`list_tools`, :meth:`get_agent_info`,
        or an ``_call_a2a_tool`` invocation). Returns ``None`` before
        the card has been fetched so callers can distinguish "not yet
        known" from "peer advertises nothing" (empty list).

        Example::

            client = ADCPClient(a2a_config)
            await client.adapter.get_agent_info()
            print(client.a2a_protocol_versions)  # ['0.3', '1.0']
        """
        if self._cached_agent_card is None:
            return None
        return sorted(
            {iface.protocol_version for iface in self._cached_agent_card.supported_interfaces}
        )

    def set_context_id(self, context_id: str | None) -> None:
        """Set the A2A context_id for subsequent message sends.

        Pass ``None`` to clear — the server mints a fresh id on the next
        call — or a string to seed. Seeding is safe for *resume* (pass
        back an id the server previously returned). Seeding with a
        *self-generated* id is server-dependent: per the A2A spec,
        agents MAY accept or reject client-supplied ids, and some
        frameworks (notably ADK) rewrite the id into their own session
        format and return the rewritten value on the next response — at
        which point this adapter auto-adopts it.

        Also clears any retained ``active_task_id``: switching context
        always starts a fresh task under the new context.
        """
        self._context_id = context_id
        self._active_task_id = None

    def _restore_active_task_id(self, task_id: str) -> None:
        """Internal: rehydrate ``active_task_id`` from a persisted checkpoint.

        Separate from normal in-flight state updates so the checkpoint
        restore path is an explicit contract — a rename of the storage
        field fails loudly here instead of silently breaking resume.
        Intended for ``ADCPClient.from_checkpoint`` only.
        """
        self._active_task_id = task_id

    async def _get_httpx_client(self) -> httpx.AsyncClient:
        """Get or create the HTTP client with connection pooling."""
        if self._httpx_client is None:
            limits = httpx.Limits(
                max_keepalive_connections=10,
                max_connections=20,
                keepalive_expiry=30.0,
            )

            headers = {}
            if self.agent_config.auth_token:
                if self.agent_config.auth_type == "bearer":
                    headers["Authorization"] = f"Bearer {self.agent_config.auth_token}"
                else:
                    headers[self.agent_config.auth_header] = self.agent_config.auth_token

            if self.agent_config.extra_headers:
                headers.update(self.agent_config.extra_headers)

            # When ADCPClient installed a signing_request_hook, register it as
            # an httpx request event hook so RFC 9421 signature headers are
            # attached transparently to every outgoing request. The hook is
            # a bound method, so each call reads the owning client's live
            # state (signing config, cached capabilities) — it is *not* a
            # snapshot. Out-of-band calls (e.g. agent-card fetch) no-op
            # inside the hook because the autosign ContextVar isn't set.
            #
            # follow_redirects is forced off whenever signing is active: RFC
            # 9421 binds the signature to the original `@authority`, so a 302
            # would forward stale signature bytes to a new host. httpx's
            # current default is False already, but pinning it matches the
            # MCP factory's invariant and protects against future upstream
            # changes or a2a-sdk overrides.
            event_hooks: dict[str, list[Any]] = {}
            client_kwargs: dict[str, Any] = {
                "limits": limits,
                "headers": headers,
                "timeout": self.agent_config.timeout,
            }
            if self.signing_request_hook is not None:
                event_hooks["request"] = [self.signing_request_hook]
                client_kwargs["follow_redirects"] = False
            if event_hooks:
                client_kwargs["event_hooks"] = event_hooks

            self._httpx_client = httpx.AsyncClient(**client_kwargs)
            logger.debug(
                f"Created HTTP client with connection pooling for agent {self.agent_config.id}"
            )
        return self._httpx_client

    async def _get_a2a_client(self) -> Client:
        """Get or create the A2A client.

        Uses :class:`~a2a.client.ClientFactory` to build a transport-negotiated
        :class:`~a2a.client.Client` against the resolved
        :class:`~a2a.types.AgentCard`. The shared ``httpx.AsyncClient`` is
        passed into the :class:`~a2a.client.ClientConfig` so the signing
        request hook and connection pool are reused across every outbound
        send.
        """
        if self._a2a_client is None:
            httpx_client = await self._get_httpx_client()

            # Use A2ACardResolver to fetch the agent card
            card_resolver = A2ACardResolver(
                httpx_client=httpx_client,
                base_url=self.agent_config.agent_uri,
            )

            try:
                agent_card = await card_resolver.get_agent_card()
                logger.debug(f"Fetched agent card for {self.agent_config.id}")
            except httpx.HTTPStatusError as e:
                status_code = e.response.status_code
                if status_code in (401, 403):
                    raise ADCPAuthenticationError(
                        f"Authentication failed: HTTP {status_code}",
                        agent_id=self.agent_config.id,
                        agent_uri=self.agent_config.agent_uri,
                    ) from e
                else:
                    raise ADCPConnectionError(
                        f"Failed to fetch agent card: HTTP {status_code}",
                        agent_id=self.agent_config.id,
                        agent_uri=self.agent_config.agent_uri,
                    ) from e
            except httpx.TimeoutException as e:
                raise ADCPTimeoutError(
                    f"Timeout fetching agent card: {e}",
                    agent_id=self.agent_config.id,
                    agent_uri=self.agent_config.agent_uri,
                    timeout=self.agent_config.timeout,
                ) from e
            except httpx.HTTPError as e:
                raise ADCPConnectionError(
                    f"Failed to fetch agent card: {e}",
                    agent_id=self.agent_config.id,
                    agent_uri=self.agent_config.agent_uri,
                ) from e

            # Build a non-streaming client that reuses our httpx pool.
            # Streaming is disabled: the ADCP adapter surface is one
            # request in, one task out — streaming would require an
            # async iterator API that does not match the SDK contract.
            self._cached_agent_card = agent_card
            client_card = agent_card
            if self._force_a2a_version is not None:
                # Filter the advertised interfaces to the pinned version
                # before handing the card to ClientFactory; the factory
                # picks a transport from whatever remains. Raising here
                # is nicer than a cryptic "no transport available" deep
                # in the SDK.
                client_card = _filter_card_to_version(agent_card, self._force_a2a_version)
                if not client_card.supported_interfaces:
                    raise ADCPConnectionError(
                        f"Peer does not advertise A2A protocol_version="
                        f"{self._force_a2a_version!r}; advertised versions: "
                        f"{sorted({i.protocol_version for i in agent_card.supported_interfaces})}",
                        agent_id=self.agent_config.id,
                        agent_uri=self.agent_config.agent_uri,
                    )
            factory = ClientFactory(ClientConfig(httpx_client=httpx_client, streaming=False))
            self._a2a_client = factory.create(client_card)
            logger.debug(f"Created A2A client for agent {self.agent_config.id}")

        return self._a2a_client

    async def _send_and_aggregate(
        self, client: Client, request: pb.SendMessageRequest
    ) -> pb.StreamResponse:
        """Send a non-streaming request and return the terminal StreamResponse.

        The 1.0 :meth:`~a2a.client.Client.send_message` is an async
        generator that yields :class:`StreamResponse` events — with
        ``streaming=False`` it yields a single event carrying the final
        task. Pulls that event out so the ADCP adapter can stay
        request/response. Raises :class:`RuntimeError` if the generator
        yields nothing (should not happen: the SDK raises before
        yielding zero events).
        """
        last: pb.StreamResponse | None = None
        stream = client.send_message(request)
        async for event in stream:
            last = event
        if last is None:
            raise RuntimeError("A2A client yielded no response events")
        return last

    async def close(self) -> None:
        """Close the HTTP client and clean up resources."""
        if self._httpx_client is not None:
            logger.debug(f"Closing A2A adapter client for agent {self.agent_config.id}")
            # Close the A2A client first so it can drain any transport
            # state (grpc channel, streaming iterator) before we tear
            # down the shared httpx pool underneath it.
            if self._a2a_client is not None:
                try:
                    await self._a2a_client.close()
                except Exception:  # noqa: BLE001
                    logger.debug("A2A client close raised; ignoring", exc_info=True)
            await self._httpx_client.aclose()
            self._httpx_client = None
            self._a2a_client = None

    async def _call_a2a_tool(
        self, tool_name: str, params: dict[str, Any], use_explicit_skill: bool = True
    ) -> TaskResult[Any]:
        """
        Call a tool using A2A protocol via official a2a-sdk client.

        Args:
            tool_name: Name of the skill/tool to invoke
            params: Parameters to pass to the skill
            use_explicit_skill: If True, use explicit skill invocation (deterministic).
                               If False, use natural language (flexible).

        The default is explicit skill invocation for predictable, repeatable behavior.
        See: https://docs.adcontextprotocol.org/docs/protocols/a2a-guide
        """
        start_time = time.time() if self.agent_config.debug else None
        if _idempotency.is_mutating(tool_name) and self.idempotency_capability_check:
            await self.idempotency_capability_check()
        params, idempotency_key = _idempotency.inject_key(
            tool_name, params, client_token=self.idempotency_client_token
        )
        # Apply per-instance envelope enrichment (e.g. adcp_version pin).
        params = self._enrich_outgoing_params(params)

        # Pre-send schema validation. Matches the MCP adapter: strict mode
        # surfaces as TaskStatus.FAILED so the SDK's unified failure model
        # is preserved; warn mode logs and continues; off short-circuits.
        try:
            validate_outgoing_request(tool_name, params, self.request_validation_mode)
        except SchemaValidationError as exc:
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=str(exc),
                success=False,
                idempotency_key=idempotency_key,
            )

        a2a_client = await self._get_a2a_client()

        # Build A2A message
        message_id = str(uuid4())

        if use_explicit_skill:
            # Explicit skill invocation (deterministic): a single DataPart
            # carrying ``{"skill": tool_name, "parameters": params}``.
            parts = [_make_data_part({"skill": tool_name, "parameters": params})]
        else:
            # Natural language invocation (flexible): agent interprets
            # intent from text.
            parts = [_make_text_part(self._format_tool_request(tool_name, params))]

        message = pb.Message(
            message_id=message_id,
            role=pb.Role.ROLE_USER,
            parts=parts,
            context_id=self._context_id or "",
            task_id=self._active_task_id or "",
        )

        request = pb.SendMessageRequest(message=message)

        debug_info = None
        debug_request: dict[str, Any] = {}
        if self.agent_config.debug:
            debug_request = {
                "method": "send_message",
                "message_id": message_id,
                "tool": tool_name,
                "params": _idempotency.redact_params(params),
            }

        # Stamp the AdCP operation name so the httpx request event hook
        # installed by ADCPClient for RFC 9421 auto-signing can look up the
        # right signing policy. Set only around send_message so out-of-band
        # httpx calls (the agent-card fetch above, or unrelated work on
        # sibling tasks) stay outside the signing scope.
        signing_token = _signing_operation.set(tool_name)
        try:
            # Non-streaming send returns a single StreamResponse envelope.
            stream_event = await self._send_and_aggregate(a2a_client, request)

            payload_kind = stream_event.WhichOneof("payload")
            if payload_kind == "task":
                result_task = stream_event.task

                if self.agent_config.debug and start_time:
                    duration_ms = (time.time() - start_time) * 1000
                    debug_info = DebugInfo(
                        request=debug_request,
                        response=_idempotency.deep_redact(
                            {"result": _task_to_redacted_dict(result_task)}
                        ),
                        duration_ms=duration_ms,
                    )

                # Compute next-turn state from the response but do NOT
                # commit yet — _process_task_response and the idempotency
                # check below can raise, and leaving the adapter advanced
                # after an exception would orphan the legitimate in-flight
                # task on the next retry. Commit only after both succeed.
                next_context_id = result_task.context_id or None
                if result_task.status.state in self._NONTERMINAL_TASK_STATES:
                    next_active_task_id: str | None = result_task.id
                else:
                    # Terminal states (completed/failed/canceled/rejected)
                    # clear the retained task_id — subsequent calls start
                    # a new task under the same context. The defensive
                    # unspecified state falls here too; warn so operators
                    # notice if a server starts emitting it.
                    next_active_task_id = None
                    if result_task.status.state == pb.TaskState.TASK_STATE_UNSPECIFIED:
                        logger.warning(
                            "A2A agent %s returned TASK_STATE_UNSPECIFIED for "
                            "task_id=%s; clearing active_task_id and "
                            "starting a fresh task on next call",
                            self.agent_config.id,
                            result_task.id,
                        )

                task_result = self._process_task_response(result_task, debug_info)
                _idempotency.raise_for_idempotency_error(
                    tool_name, task_result.data, self.agent_config.id
                )
                # All raise-sites have passed; commit next-turn state so
                # the adapter reflects the response the caller is about
                # to receive.
                self._context_id = next_context_id
                self._active_task_id = next_active_task_id
                # Post-receive schema validation. Only runs when the task
                # carries data (terminal completion); async interim states
                # with ``data=None`` skip naturally. Strict mode flips the
                # TaskResult to FAILED; warn mode logs and passes through.
                # Runs after the state commit — a payload-schema failure
                # doesn't invalidate the A2A envelope ids, and the next
                # call in the same conversation should still target the
                # right session.
                if task_result.success and task_result.data is not None:
                    response_outcome = validate_incoming_response(
                        tool_name, task_result.data, self.response_validation_mode
                    )
                    if not response_outcome.valid and self.response_validation_mode == "strict":
                        task_result = TaskResult[Any](
                            status=TaskStatus.FAILED,
                            error=(
                                f"Schema validation failed for {tool_name}: "
                                f"{format_issues(response_outcome.issues)}"
                            ),
                            message=task_result.message,
                            success=False,
                            debug_info=task_result.debug_info,
                            idempotency_key=task_result.idempotency_key,
                        )
                return _idempotency.annotate_result(task_result, idempotency_key)

            if payload_kind == "message":
                # Message response (shouldn't happen for send_message with
                # skill invocation, but surface a graceful fallback).
                agent_id = self.agent_config.id
                logger.warning(f"Received Message instead of Task from A2A agent {agent_id}")
                return TaskResult[Any](
                    status=TaskStatus.COMPLETED,
                    data=None,
                    message="Received message response",
                    success=True,
                    debug_info=debug_info,
                )

            # Shouldn't reach here
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=f"Invalid response from A2A client (payload={payload_kind!r})",
                success=False,
                debug_info=debug_info,
                idempotency_key=idempotency_key,
            )

        except httpx.HTTPStatusError as e:
            status_code = e.response.status_code
            if self.agent_config.debug and start_time:
                duration_ms = (time.time() - start_time) * 1000
                debug_info = DebugInfo(
                    request=debug_request,
                    response={"error": str(e), "status_code": status_code},
                    duration_ms=duration_ms,
                )

            if status_code in (401, 403):
                error_msg = f"Authentication failed: HTTP {status_code}"
            else:
                error_msg = f"HTTP {status_code} error: {e}"

            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=error_msg,
                success=False,
                debug_info=debug_info,
                idempotency_key=idempotency_key,
            )
        except httpx.TimeoutException as e:
            if self.agent_config.debug and start_time:
                duration_ms = (time.time() - start_time) * 1000
                debug_info = DebugInfo(
                    request=debug_request,
                    response={"error": str(e)},
                    duration_ms=duration_ms,
                )
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=f"Timeout: {e}",
                success=False,
                debug_info=debug_info,
                idempotency_key=idempotency_key,
            )
        except (IdempotencyConflictError, IdempotencyExpiredError):
            # Propagate typed idempotency errors — callers MUST handle these
            # distinctly (mint fresh key / reconcile state). Other ADCPError
            # subclasses (connection, timeout, auth) continue to be converted
            # to TaskResult(failed) below, preserving the existing contract.
            raise
        except Exception as e:
            if self.agent_config.debug and start_time:
                duration_ms = (time.time() - start_time) * 1000
                debug_info = DebugInfo(
                    request=debug_request,
                    response={"error": str(e)},
                    duration_ms=duration_ms,
                )
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=str(e),
                success=False,
                debug_info=debug_info,
                idempotency_key=idempotency_key,
            )
        finally:
            _signing_operation.reset(signing_token)

    def _process_task_response(
        self, task: pb.Task, debug_info: DebugInfo | None
    ) -> TaskResult[Any]:
        """Process a Task response from A2A into our TaskResult format."""
        task_state = task.status.state

        if task_state == pb.TaskState.TASK_STATE_COMPLETED:
            # Extract the result from the artifacts array
            result_data = self._extract_result_from_task(task)

            # Check for task-level errors in the payload
            errors = result_data.get("errors", []) if isinstance(result_data, dict) else []
            has_errors = bool(errors)

            return TaskResult[Any](
                status=TaskStatus.COMPLETED,
                data=result_data,
                message=self._extract_text_from_task(task),
                success=not has_errors,
                metadata={
                    "task_id": task.id,
                    "context_id": task.context_id,
                },
                debug_info=debug_info,
            )
        elif task_state == pb.TaskState.TASK_STATE_FAILED:
            # Per transport-errors.mdx §A2A Binding: a failed task carries
            # an ``adcp_error`` DataPart alongside the human-readable
            # TextPart. The structured envelope lands on
            # ``TaskResult.adcp_error`` for programmatic branching; the
            # text stays on ``error`` for humans. When the seller omits
            # the TextPart, fall back to the structured envelope's
            # ``message`` / ``code`` so adopters don't see the
            # ``"Task failed"`` placeholder mask a real diagnostic.
            text_msg = self._extract_text_from_task(task)
            adcp_error = self._extract_adcp_error_from_task(task)
            if text_msg:
                error_msg: str | None = text_msg
            elif adcp_error:
                error_msg = adcp_error.get("message") or adcp_error.get("code")
            else:
                error_msg = "Task failed"
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=error_msg,
                adcp_error=adcp_error,
                success=False,
                debug_info=debug_info,
            )
        else:
            # Handle all interim states (submitted, working, input-required, etc.).
            # Metadata ``status`` stays in the 0.3-style lowercase spec form
            # (``working``, ``input-required``) so downstream consumers don't
            # need to learn the TaskState_ prefix.
            state_name = pb.TaskState.Name(task_state)
            if state_name.startswith("TASK_STATE_"):
                status_str = state_name[len("TASK_STATE_") :].lower().replace("_", "-")
            else:
                status_str = state_name.lower()
            return TaskResult[Any](
                status=TaskStatus.SUBMITTED,
                data=None,  # Interim responses may not have structured AdCP content
                message=self._extract_text_from_task(task),
                success=True,
                metadata={
                    "task_id": task.id,
                    "context_id": task.context_id,
                    "status": status_str,
                },
                debug_info=debug_info,
            )

    def _format_tool_request(self, tool_name: str, params: dict[str, Any]) -> str:
        """Format tool request as natural language for A2A."""
        import json

        return f"Execute tool: {tool_name}\nParameters: {json.dumps(params, indent=2)}"

    def _extract_result_from_task(self, task: pb.Task) -> Any:
        """
        Extract result data from A2A Task following canonical format.

        Per A2A response spec:
        - Responses MUST include at least one DataPart (``data`` oneof)
        - When multiple DataParts exist in an artifact, the last one is authoritative
        - When multiple artifacts exist, use the last one (most recent in streaming)
        - DataParts contain structured AdCP payload
        """
        if not task.artifacts:
            logger.warning("A2A Task missing required artifacts array")
            return {}

        # Use last artifact (most recent in streaming scenarios)
        target_artifact = task.artifacts[-1]

        if not target_artifact.parts:
            logger.warning("A2A Task artifact has no parts")
            return {}

        data_parts = [
            d for d in (_part_data_dict(p) for p in target_artifact.parts) if d is not None
        ]

        if not data_parts:
            logger.warning("A2A Task missing required DataPart (data oneof)")
            return {}

        # Use last DataPart as authoritative (handles streaming scenarios within an artifact)
        data = data_parts[-1]

        # Some A2A implementations (e.g., ADK) wrap the response in {"response": {...}}
        # Unwrap it to get the actual AdCP payload if present
        if isinstance(data, dict) and "response" in data:
            return data["response"]

        return data

    def _extract_adcp_error_from_task(self, task: pb.Task) -> dict[str, Any] | None:
        """Extract a spec-shaped ``adcp_error`` DataPart from a failed task.

        Per transport-errors.mdx §A2A Binding the failed task's artifact
        carries a DataPart wrapping ``{"adcp_error": {...}}``. Returns the
        validated envelope or ``None`` if no spec-shaped payload is present
        (spec-permitted: graceful sellers MAY omit the structured envelope,
        in which case adopters fall back to ``TaskResult.error``).
        """
        data = self._extract_result_from_task(task)
        if not isinstance(data, dict):
            return None
        return validate_adcp_error(data.get("adcp_error"))

    def _extract_text_from_task(self, task: pb.Task) -> str | None:
        """Extract human-readable message from TextPart if present."""
        if not task.artifacts:
            return None

        # Use last artifact (most recent in streaming scenarios)
        target_artifact = task.artifacts[-1]

        for part in target_artifact.parts:
            text = _part_text(part)
            if text is not None:
                return text

        return None

    # ========================================================================
    # ADCP Protocol Methods
    # ========================================================================

    async def get_products(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get advertising products."""
        return await self._call_a2a_tool("get_products", params)

    async def list_creative_formats(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List supported creative formats."""
        return await self._call_a2a_tool("list_creative_formats", params)

    async def sync_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync creatives."""
        return await self._call_a2a_tool("sync_creatives", params)

    async def list_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creatives."""
        return await self._call_a2a_tool("list_creatives", params)

    async def get_media_buy_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buy delivery."""
        return await self._call_a2a_tool("get_media_buy_delivery", params)

    async def get_media_buys(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buys with status, creative approval state, and optional delivery snapshots."""
        return await self._call_a2a_tool("get_media_buys", params)

    async def get_signals(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get signals."""
        return await self._call_a2a_tool("get_signals", params)

    async def activate_signal(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Activate signal."""
        return await self._call_a2a_tool("activate_signal", params)

    async def provide_performance_feedback(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Provide performance feedback."""
        return await self._call_a2a_tool("provide_performance_feedback", params)

    async def log_event(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Log event."""
        return await self._call_a2a_tool("log_event", params)

    async def sync_event_sources(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync event sources."""
        return await self._call_a2a_tool("sync_event_sources", params)

    async def sync_audiences(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync audiences."""
        return await self._call_a2a_tool("sync_audiences", params)

    async def sync_catalogs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync catalogs."""
        return await self._call_a2a_tool("sync_catalogs", params)

    async def preview_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Generate preview URLs for a creative manifest."""
        return await self._call_a2a_tool("preview_creative", params)

    async def create_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create media buy."""
        return await self._call_a2a_tool("create_media_buy", params)

    async def update_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update media buy."""
        return await self._call_a2a_tool("update_media_buy", params)

    async def build_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Build creative."""
        return await self._call_a2a_tool("build_creative", params)

    async def get_creative_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get creative delivery."""
        return await self._call_a2a_tool("get_creative_delivery", params)

    async def list_transformers(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creative transformers."""
        return await self._call_a2a_tool("list_transformers", params)

    async def list_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List accounts."""
        return await self._call_a2a_tool("list_accounts", params)

    async def sync_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync accounts."""
        return await self._call_a2a_tool("sync_accounts", params)

    async def get_account_financials(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get account financials."""
        return await self._call_a2a_tool("get_account_financials", params)

    async def report_usage(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report account usage."""
        return await self._call_a2a_tool("report_usage", params)

    async def list_tools(self) -> list[str]:
        """
        List available tools from A2A agent.

        Uses A2A client which already fetched the agent card during initialization.
        """
        # Ensure the A2A client (and cached agent card) is initialized.
        await self._get_a2a_client()

        if self._cached_agent_card is None:
            raise RuntimeError("Agent card cache was not populated by _get_a2a_client")
        agent_card: pb.AgentCard = self._cached_agent_card

        tool_names = [skill.name for skill in agent_card.skills if skill.name]
        logger.info(f"Found {len(tool_names)} tools from A2A agent {self.agent_config.id}")
        return tool_names

    async def get_agent_info(self) -> dict[str, Any]:
        """
        Get agent information including AdCP extension metadata from A2A agent card.

        Fetches the agent card via :class:`~a2a.client.A2ACardResolver` and
        extracts:

        - Basic agent info (name, description, version)
        - AdCP extension (extensions.adcp.adcp_version, extensions.adcp.protocols_supported)
        - Available skills/tools

        Returns:
            Dictionary with agent metadata
        """
        await self._get_a2a_client()

        logger.debug(f"Fetching A2A agent info for {self.agent_config.id}")

        if self._cached_agent_card is None:
            raise RuntimeError("Agent card cache was not populated by _get_a2a_client")
        agent_card: pb.AgentCard = self._cached_agent_card

        info: dict[str, Any] = {
            "name": agent_card.name,
            "description": agent_card.description,
            "version": agent_card.version,
            "protocol": "a2a",
            # A2A wire versions the peer advertises. Our server emits
            # both "0.3" and "1.0" so clients of either era interoperate;
            # this field lets buyers confirm what a given peer speaks.
            "a2a_protocol_versions": sorted(
                {iface.protocol_version for iface in agent_card.supported_interfaces}
            ),
        }

        tool_names = [skill.name for skill in agent_card.skills if skill.name]
        if tool_names:
            info["tools"] = tool_names

        # The 1.0 proto :class:`AgentCard` has no ``extensions`` map.
        # Sellers advertising AdCP capabilities must surface them via
        # ``skills`` entries or a follow-up
        # ``get_adcp_capabilities`` call rather than an out-of-band
        # extensions dict (which the 0.3 Pydantic card accepted).

        logger.info(f"Retrieved agent info for {self.agent_config.id}")
        return info

    # ========================================================================
    # V3 Protocol Methods - Protocol Discovery
    # ========================================================================

    async def get_adcp_capabilities(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get AdCP capabilities from the agent."""
        return await self._call_a2a_tool("get_adcp_capabilities", params)

    async def get_task_status(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get task status from the agent."""
        return await self._call_a2a_tool("get_task_status", params)

    async def list_tasks(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List tasks from the agent."""
        return await self._call_a2a_tool("list_tasks", params)

    # ========================================================================
    # V3 Protocol Methods - Content Standards
    # ========================================================================

    async def create_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create content standards configuration."""
        return await self._call_a2a_tool("create_content_standards", params)

    async def get_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get content standards configuration."""
        return await self._call_a2a_tool("get_content_standards", params)

    async def list_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List content standards configurations."""
        return await self._call_a2a_tool("list_content_standards", params)

    async def update_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update content standards configuration."""
        return await self._call_a2a_tool("update_content_standards", params)

    async def calibrate_content(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Calibrate content against standards."""
        return await self._call_a2a_tool("calibrate_content", params)

    async def validate_content_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate content delivery against standards."""
        return await self._call_a2a_tool("validate_content_delivery", params)

    async def get_media_buy_artifacts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get artifacts associated with a media buy."""
        return await self._call_a2a_tool("get_media_buy_artifacts", params)

    # ========================================================================
    # V3 Protocol Methods - Governance
    # ========================================================================

    async def get_creative_features(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Evaluate governance features for a creative."""
        return await self._call_a2a_tool("get_creative_features", params)

    async def sync_plans(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync campaign governance plans."""
        return await self._call_a2a_tool("sync_plans", params)

    async def check_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Check an action against campaign governance."""
        return await self._call_a2a_tool("check_governance", params)

    async def report_plan_outcome(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report the outcome of a governed action."""
        return await self._call_a2a_tool("report_plan_outcome", params)

    async def get_plan_audit_logs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Retrieve governance audit logs for plans."""
        return await self._call_a2a_tool("get_plan_audit_logs", params)

    # ========================================================================
    # V3 Protocol Methods - Sponsored Intelligence
    # ========================================================================

    async def si_get_offering(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get sponsored intelligence offering."""
        return await self._call_a2a_tool("si_get_offering", params)

    async def si_initiate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Initiate sponsored intelligence session."""
        return await self._call_a2a_tool("si_initiate_session", params)

    async def si_send_message(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Send message in sponsored intelligence session."""
        return await self._call_a2a_tool("si_send_message", params)

    async def si_terminate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Terminate sponsored intelligence session."""
        return await self._call_a2a_tool("si_terminate_session", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Property Lists)
    # ========================================================================

    async def create_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a property list for governance."""
        return await self._call_a2a_tool("create_property_list", params)

    async def get_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a property list with optional resolution."""
        return await self._call_a2a_tool("get_property_list", params)

    async def list_property_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List property lists."""
        return await self._call_a2a_tool("list_property_lists", params)

    async def update_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a property list."""
        return await self._call_a2a_tool("update_property_list", params)

    async def delete_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a property list."""
        return await self._call_a2a_tool("delete_property_list", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Collection Lists)
    # ========================================================================

    async def create_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a collection list for governance."""
        return await self._call_a2a_tool("create_collection_list", params)

    async def get_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a collection list with optional resolution."""
        return await self._call_a2a_tool("get_collection_list", params)

    async def list_collection_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List collection lists."""
        return await self._call_a2a_tool("list_collection_lists", params)

    async def update_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a collection list."""
        return await self._call_a2a_tool("update_collection_list", params)

    async def delete_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a collection list."""
        return await self._call_a2a_tool("delete_collection_list", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Sync Governance)
    # ========================================================================

    async def sync_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync governance agents attached to an account."""
        return await self._call_a2a_tool("sync_governance", params)

    # ========================================================================
    # V3 Protocol Methods - TMP
    # ========================================================================

    async def context_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match ad context to buyer packages."""
        return await self._call_a2a_tool("context_match", params)

    async def identity_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match user identity for package eligibility."""
        return await self._call_a2a_tool("identity_match", params)

    # ========================================================================
    # V3 Protocol Methods - Brand Rights
    # ========================================================================

    async def get_brand_identity(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get brand identity information."""
        return await self._call_a2a_tool("get_brand_identity", params)

    async def get_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get available rights for licensing."""
        return await self._call_a2a_tool("get_rights", params)

    async def acquire_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Acquire rights for brand content usage."""
        return await self._call_a2a_tool("acquire_rights", params)

    async def update_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update terms of an existing rights acquisition."""
        return await self._call_a2a_tool("update_rights", params)

    async def validate_input(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate creative input."""
        return await self._call_a2a_tool("validate_input", params)

    async def verify_brand_claim(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify a brand claim."""
        return await self._call_a2a_tool("verify_brand_claim", params)

    async def verify_brand_claims(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify brand claims."""
        return await self._call_a2a_tool("verify_brand_claims", params)

    # ========================================================================
    # V3 Protocol Methods - Compliance
    # ========================================================================

    async def comply_test_controller(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Compliance test controller (sandbox only)."""
        return await self._call_a2a_tool("comply_test_controller", params)

class MCPAdapter(ProtocolAdapter):
    """Adapter for MCP protocol using official Python MCP SDK."""

    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        if not MCP_AVAILABLE:
            raise ImportError(
                "MCP SDK not installed. Install with: pip install mcp (requires Python 3.10+)"
            )
        self._session: Any = None
        self._exit_stack: Any = None
        self._connected_url: str | None = None
        self._get_session_id: Callable[[], str | None] | None = None
        # True when the session was injected by ADCPClient.from_mcp_client().
        # Caller owns the lifecycle — close() is a no-op on injected adapters.
        self._session_is_injected: bool = False

    def _inject_session(self, session: ClientSession) -> None:
        """Pre-wire a caller-owned session, bypassing URL-based connection.

        Used by ADCPClient.from_mcp_client(). Once injected, _get_session()
        returns it immediately and close() is a no-op (caller owns lifecycle).
        """
        self._session = session
        self._session_is_injected = True

    def _http_headers(self) -> dict[str, str]:
        """Return transport headers for MCP HTTP requests."""
        headers: dict[str, str] = {}
        if self.agent_config.auth_token:
            # Support custom auth headers and types
            if self.agent_config.auth_type == "bearer":
                headers[self.agent_config.auth_header] = f"Bearer {self.agent_config.auth_token}"
            else:
                headers[self.agent_config.auth_header] = self.agent_config.auth_token

        if self.agent_config.extra_headers:
            headers.update(self.agent_config.extra_headers)
        return headers

    def _urls_to_try(self) -> list[str]:
        """Return MCP endpoint candidates, preserving the configured URL first."""
        uri = self.agent_config.agent_uri
        base = uri.rstrip("/")
        urls_to_try = [uri]
        if base.endswith("/mcp"):
            # User pointed at the MCP endpoint; also try the other slash form.
            urls_to_try.append(f"{base}/" if not uri.endswith("/") else base)
        else:
            urls_to_try.extend([f"{base}/mcp", f"{base}/mcp/"])
        return urls_to_try

    def _streamable_http_client_factory(self) -> Callable[..., httpx.AsyncClient]:
        """Return the HTTP client factory used for streamable-http requests."""
        if self.signing_request_hook is not None:
            return _make_signing_http_factory(self.signing_request_hook)
        return create_mcp_http_client

    def current_mcp_session_id(self) -> str | None:
        """Return the current SDK-managed MCP Streamable HTTP session id."""
        return self._get_session_id() if self._get_session_id is not None else None

    async def _cleanup_failed_connection(self, context: str) -> None:
        """
        Clean up resources after a failed connection attempt.

        This method handles cleanup without raising exceptions to avoid
        masking the original connection error.

        Args:
            context: Description of the context for logging (e.g., "during connection attempt")
        """
        if self._exit_stack is not None:
            old_stack = self._exit_stack
            self._exit_stack = None
            self._session = None
            self._connected_url = None
            self._get_session_id = None
            try:
                await old_stack.aclose()
            except BaseException as cleanup_error:
                # Handle all cleanup errors including ExceptionGroup
                # Re-raise KeyboardInterrupt and SystemExit immediately
                if isinstance(cleanup_error, (KeyboardInterrupt, SystemExit)):
                    raise

                if isinstance(cleanup_error, asyncio.CancelledError):
                    logger.debug(f"MCP session cleanup cancelled {context}")
                    return

                # Handle ExceptionGroup/BaseExceptionGroup from task group failures (Python 3.11+)
                # ExceptionGroup: for Exception subclasses (e.g., HTTPStatusError)
                # BaseExceptionGroup: for BaseException subclasses (e.g., CancelledError)
                # We need both because CancelledError is a BaseException, not an Exception
                is_exception_group = (
                    _ExceptionGroup is not None and isinstance(cleanup_error, _ExceptionGroup)
                ) or (
                    _BaseExceptionGroup is not None
                    and isinstance(cleanup_error, _BaseExceptionGroup)
                )

                if is_exception_group:
                    # Check if all exceptions in the group are CancelledError
                    # If so, treat the entire group as a cancellation
                    all_cancelled = all(
                        isinstance(exc, asyncio.CancelledError)
                        for exc in cleanup_error.exceptions  # type: ignore[attr-defined]
                    )
                    if all_cancelled:
                        logger.debug(f"MCP session cleanup cancelled {context}")
                        return

                    # Mixed group: skip CancelledErrors and log real errors
                    exceptions = cleanup_error.exceptions  # type: ignore[attr-defined]
                    cancelled_errors = [
                        exc for exc in exceptions if isinstance(exc, asyncio.CancelledError)
                    ]
                    cancelled_count = len(cancelled_errors)
                    if cancelled_count > 0:
                        logger.debug(
                            f"Skipping {cancelled_count} CancelledError(s) "
                            f"in mixed exception group {context}"
                        )

                    # Log each non-cancelled exception individually
                    for exc in exceptions:
                        if not isinstance(exc, asyncio.CancelledError):
                            self._log_cleanup_error(exc, context)
                else:
                    self._log_cleanup_error(cleanup_error, context)

    def _log_cleanup_error(self, exc: BaseException, context: str) -> None:
        """Log a cleanup error without raising."""
        # Check for known cleanup error patterns from httpx/anyio
        exc_str = str(exc).lower()

        # Common cleanup errors that are expected when connection fails
        is_known_cleanup_error = (
            isinstance(exc, RuntimeError)
            and ("cancel scope" in exc_str or "async context" in exc_str)
        ) or (
            # HTTP errors during cleanup (if httpx is available)
            HTTPX_AVAILABLE
            and isinstance(exc, _HTTP_STATUS_ERROR_TYPES)
        )

        if is_known_cleanup_error:
            # Expected cleanup errors - log at debug level without stack trace
            logger.debug(f"Ignoring expected cleanup error {context}: {exc}")
        else:
            # Truly unexpected cleanup errors - log at warning with full context
            logger.warning(f"Unexpected error during cleanup {context}: {exc}", exc_info=True)

    async def _get_session(self) -> ClientSession:
        """
        Get or create MCP client session with URL fallback handling.

        Raises:
            ADCPConnectionError: If connection to agent fails
        """
        if self._session is not None:
            return self._session  # type: ignore[no-any-return]

        logger.debug(f"Creating MCP session for agent {self.agent_config.id}")

        # Parse the agent URI to determine transport type
        parsed = urlparse(self.agent_config.agent_uri)

        # Use SSE transport for HTTP/HTTPS endpoints
        if parsed.scheme in ("http", "https"):
            self._exit_stack = AsyncExitStack()

            # Try the user's exact URL first, then the alternate slash form, then
            # /mcp discovery paths. MCP servers disagree on whether their endpoint
            # is at /mcp or /mcp/ — try both rather than silently normalizing.
            headers = self._http_headers()
            urls_to_try = self._urls_to_try()

            # RFC 9421 auto-signing: if ADCPClient installed a signing request
            # hook, wire it into streamable_http via a custom httpx client
            # factory. SSE transport has no equivalent knob — warn the user
            # and fall through to unsigned SSE.
            streamable_http_extra: dict[str, Any] = {}
            if self.signing_request_hook is not None:
                if self.agent_config.mcp_transport == "streamable_http":
                    streamable_http_extra["httpx_client_factory"] = (
                        self._streamable_http_client_factory()
                    )
                else:
                    logger.warning(
                        "RFC 9421 auto-signing is not supported on MCP SSE "
                        "transport for agent %s; use mcp_transport='streamable_http' "
                        "to sign outgoing requests.",
                        self.agent_config.id,
                    )

            last_error = None
            for url in urls_to_try:
                try:
                    get_session_id: Callable[[], str | None] | None = None
                    # Choose transport based on configuration
                    if self.agent_config.mcp_transport == "streamable_http":
                        # Use streamable HTTP transport (newer, bidirectional)
                        read, write, get_session_id = await self._exit_stack.enter_async_context(
                            streamablehttp_client(
                                url,
                                headers=headers,
                                timeout=self.agent_config.timeout,
                                **streamable_http_extra,
                            )
                        )
                    else:
                        # Use SSE transport (legacy, but widely supported)
                        read, write = await self._exit_stack.enter_async_context(
                            sse_client(url, headers=headers)
                        )

                    self._session = await self._exit_stack.enter_async_context(
                        _ClientSession(read, write)
                    )

                    # Initialize the session
                    await self._session.initialize()
                    self._connected_url = url
                    if self.agent_config.mcp_transport == "streamable_http":
                        self._get_session_id = get_session_id

                    logger.info(
                        f"Connected to MCP agent {self.agent_config.id} at {url} "
                        f"using {self.agent_config.mcp_transport} transport"
                    )
                    if url != self.agent_config.agent_uri:
                        logger.info(
                            f"Note: Connected using fallback URL {url} "
                            f"(configured: {self.agent_config.agent_uri})"
                        )

                    return self._session  # type: ignore[no-any-return]
                except BaseException as e:
                    # Catch BaseException to handle CancelledError from failed initialization
                    # Re-raise KeyboardInterrupt and SystemExit immediately
                    if isinstance(e, (KeyboardInterrupt, SystemExit)):
                        raise
                    last_error = e
                    # Clean up the exit stack on failure to avoid resource leaks
                    await self._cleanup_failed_connection("during connection attempt")

                    # If this isn't the last URL to try, create a new exit stack and continue
                    if url != urls_to_try[-1]:
                        logger.debug(f"Retrying with next URL after error: {last_error}")
                        self._exit_stack = AsyncExitStack()
                        continue
                    # If this was the last URL, raise the error
                    logger.error(
                        f"Failed to connect to MCP agent {self.agent_config.id} using "
                        f"{self.agent_config.mcp_transport} transport. "
                        f"Tried URLs: {', '.join(urls_to_try)}"
                    )

                    # Classify error type for better exception handling
                    error_str = str(last_error).lower()
                    if "401" in error_str or "403" in error_str or "unauthorized" in error_str:
                        from adcp.exceptions import ADCPAuthenticationError

                        raise ADCPAuthenticationError(
                            f"Authentication failed: {last_error}",
                            agent_id=self.agent_config.id,
                            agent_uri=self.agent_config.agent_uri,
                        ) from last_error
                    elif "timeout" in error_str:
                        raise ADCPTimeoutError(
                            f"Connection timeout: {last_error}",
                            agent_id=self.agent_config.id,
                            agent_uri=self.agent_config.agent_uri,
                            timeout=self.agent_config.timeout,
                        ) from last_error
                    else:
                        raise ADCPConnectionError(
                            f"Failed to connect: {last_error}",
                            agent_id=self.agent_config.id,
                            agent_uri=self.agent_config.agent_uri,
                        ) from last_error

            # This shouldn't be reached, but just in case
            raise RuntimeError(f"Failed to connect to MCP agent at {self.agent_config.agent_uri}")
        else:
            raise ValueError(f"Unsupported transport scheme: {parsed.scheme}")

    def _serialize_mcp_content(self, content: list[Any]) -> list[dict[str, Any]]:
        """
        Convert MCP SDK content objects to plain dicts.

        The MCP SDK returns Pydantic objects (TextContent, ImageContent, etc.)
        but the rest of the ADCP client expects protocol-agnostic dicts.
        This method handles the translation at the protocol boundary.

        Args:
            content: List of MCP content items (may be dicts or Pydantic objects)

        Returns:
            List of plain dicts representing the content
        """
        result = []
        for item in content:
            # Already a dict, pass through
            if isinstance(item, dict):
                result.append(item)
            # Pydantic v2 model with model_dump()
            elif hasattr(item, "model_dump"):
                result.append(item.model_dump())
            # Pydantic v1 model with dict()
            elif hasattr(item, "dict") and callable(item.dict):
                result.append(item.dict())
            # Fallback: try to access __dict__
            elif hasattr(item, "__dict__"):
                result.append(dict(item.__dict__))
            # Last resort: serialize as unknown type
            else:
                logger.warning(f"Unknown MCP content type: {type(item)}, serializing as string")
                result.append({"type": "unknown", "data": str(item)})
        return result

    async def _call_mcp_tool(self, tool_name: str, params: dict[str, Any]) -> TaskResult[Any]:
        """Call a tool using MCP protocol."""
        start_time = time.time() if self.agent_config.debug else None
        debug_info = None
        debug_request: dict[str, Any] = {}
        if _idempotency.is_mutating(tool_name) and self.idempotency_capability_check:
            await self.idempotency_capability_check()
        params, idempotency_key = _idempotency.inject_key(
            tool_name, params, client_token=self.idempotency_client_token
        )
        # Apply per-instance envelope enrichment (e.g. adcp_version pin).
        # Runs after idempotency injection so the enriched dict is the
        # one that's validated and sent.
        params = self._enrich_outgoing_params(params)

        try:
            # Pre-send schema validation — throws in strict, logs in warn,
            # skips in off. Runs before session setup so a drifted payload
            # doesn't even open a connection.
            try:
                validate_outgoing_request(tool_name, params, self.request_validation_mode)
            except SchemaValidationError as exc:
                return TaskResult[Any](
                    status=TaskStatus.FAILED,
                    error=str(exc),
                    success=False,
                    idempotency_key=idempotency_key,
                )

            session = await self._get_session()

            if self.agent_config.debug:
                debug_request = {
                    "protocol": "MCP",
                    "tool": tool_name,
                    "params": _idempotency.redact_params(params),
                    "transport": self.agent_config.mcp_transport,
                }

            # Stamp the AdCP operation name so the httpx request event hook
            # installed by ADCPClient (when a SigningConfig is present) can
            # look up the seller's signing policy for this call. Scoped
            # tightly around call_tool so session.initialize() above and
            # other out-of-band traffic stay outside the signing scope.
            signing_token = _signing_operation.set(tool_name)
            try:
                # Call the tool using MCP client session
                result = await session.call_tool(tool_name, params)
            finally:
                _signing_operation.reset(signing_token)

            # Check if this is an error response
            is_error = hasattr(result, "isError") and result.isError

            # Extract human-readable message from content
            message_text = None
            if hasattr(result, "content") and result.content:
                serialized_content = self._serialize_mcp_content(result.content)
                if isinstance(serialized_content, list):
                    for item in serialized_content:
                        is_text = isinstance(item, dict) and item.get("type") == "text"
                        if is_text and item.get("text"):
                            message_text = item["text"]
                            break

            # Handle error responses per transport-errors.mdx §Client Detection
            # Order. Extract the adcp_error object from structuredContent first,
            # then from text fallback — whichever is present.
            if is_error:
                adcp_error = extract_adcp_error(result)
                # Raise typed idempotency exceptions before building a generic
                # TaskResult(failed), so callers that catch them distinctly
                # don't lose the signal.
                if adcp_error and adcp_error.get("code") in (
                    "IDEMPOTENCY_CONFLICT",
                    "IDEMPOTENCY_EXPIRED",
                ):
                    from adcp.exceptions import classify_task_error

                    raise classify_task_error(
                        tool_name, [adcp_error], agent_id=self.agent_config.id
                    )
                # FastMCP-style is_error with plain-text content: text-match
                # fallback for the two idempotency codes.
                _idempotency.raise_for_idempotency_text(
                    tool_name, message_text, self.agent_config.id
                )
                error_message = (
                    (adcp_error.get("message") if adcp_error else None)
                    or message_text
                    or "Tool execution failed"
                )
                if self.agent_config.debug and start_time:
                    duration_ms = (time.time() - start_time) * 1000
                    debug_info = DebugInfo(
                        request=debug_request,
                        response={
                            "error": error_message,
                            "is_error": True,
                            "adcp_error": adcp_error,
                        },
                        duration_ms=duration_ms,
                    )
                return TaskResult[Any](
                    status=TaskStatus.FAILED,
                    error=error_message,
                    adcp_error=adcp_error,
                    success=False,
                    debug_info=debug_info,
                    idempotency_key=idempotency_key,
                )

            # Success extraction per mcp-response-extraction.mdx §Extraction
            # Algorithm: prefer structuredContent (MCP 2025-03-26+), fall back
            # to JSON-parsing content[].text for older servers (including the
            # AdCP reference training agent).
            data_to_return = extract_adcp_success(result)
            if data_to_return is None:
                raise ValueError(
                    f"MCP tool {tool_name} returned no structured AdCP data. "
                    f"Neither structuredContent nor content[].text yielded a "
                    f"parseable non-adcp_error JSON object. "
                    f"Got content: {result.content if hasattr(result, 'content') else 'none'}"
                )

            if self.agent_config.debug and start_time:
                duration_ms = (time.time() - start_time) * 1000
                debug_info = DebugInfo(
                    request=debug_request,
                    response=_idempotency.deep_redact(
                        {
                            "data": data_to_return,
                            "message": message_text,
                            "is_error": False,
                        }
                    ),
                    duration_ms=duration_ms,
                )

            _idempotency.raise_for_idempotency_error(
                tool_name, data_to_return, self.agent_config.id
            )

            # Post-receive schema validation — catches field-name drift from
            # agents. Strict mode fails the task; warn mode logs and returns
            # the data unchanged; off short-circuits without invoking the
            # validator. Never raises — mirrors the existing contract where
            # response-side failures surface as TaskStatus.FAILED.
            response_outcome = validate_incoming_response(
                tool_name, data_to_return, self.response_validation_mode
            )
            if not response_outcome.valid and self.response_validation_mode == "strict":
                return TaskResult[Any](
                    status=TaskStatus.FAILED,
                    error=(
                        f"Schema validation failed for {tool_name}: "
                        f"{format_issues(response_outcome.issues)}"
                    ),
                    message=message_text,
                    success=False,
                    debug_info=debug_info,
                    idempotency_key=idempotency_key,
                )

            # Return both the structured data and the human-readable message
            task_result = TaskResult[Any](
                status=TaskStatus.COMPLETED,
                data=data_to_return,
                message=message_text,
                success=True,
                debug_info=debug_info,
            )
            return _idempotency.annotate_result(task_result, idempotency_key)

        except (IdempotencyConflictError, IdempotencyExpiredError):
            # Propagate typed idempotency errors — callers MUST handle these
            # distinctly (mint fresh key / reconcile state). Other ADCPError
            # subclasses (connection, timeout) continue to be converted to
            # TaskResult(failed) below, preserving the existing contract.
            raise
        except Exception as e:
            if self.agent_config.debug and start_time:
                duration_ms = (time.time() - start_time) * 1000
                debug_info = DebugInfo(
                    request=debug_request,
                    response={"error": str(e)},
                    duration_ms=duration_ms,
                )
            return TaskResult[Any](
                status=TaskStatus.FAILED,
                error=str(e),
                success=False,
                debug_info=debug_info,
                idempotency_key=idempotency_key,
            )

    # ========================================================================
    # ADCP Protocol Methods
    # ========================================================================

    async def get_products(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get advertising products."""
        return await self._call_mcp_tool("get_products", params)

    async def list_creative_formats(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List supported creative formats."""
        return await self._call_mcp_tool("list_creative_formats", params)

    async def sync_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync creatives."""
        return await self._call_mcp_tool("sync_creatives", params)

    async def list_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creatives."""
        return await self._call_mcp_tool("list_creatives", params)

    async def get_media_buy_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buy delivery."""
        return await self._call_mcp_tool("get_media_buy_delivery", params)

    async def get_media_buys(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buys with status, creative approval state, and optional delivery snapshots."""
        return await self._call_mcp_tool("get_media_buys", params)

    async def get_signals(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get signals."""
        return await self._call_mcp_tool("get_signals", params)

    async def activate_signal(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Activate signal."""
        return await self._call_mcp_tool("activate_signal", params)

    async def provide_performance_feedback(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Provide performance feedback."""
        return await self._call_mcp_tool("provide_performance_feedback", params)

    async def log_event(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Log event."""
        return await self._call_mcp_tool("log_event", params)

    async def sync_event_sources(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync event sources."""
        return await self._call_mcp_tool("sync_event_sources", params)

    async def sync_audiences(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync audiences."""
        return await self._call_mcp_tool("sync_audiences", params)

    async def sync_catalogs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync catalogs."""
        return await self._call_mcp_tool("sync_catalogs", params)

    async def preview_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Generate preview URLs for a creative manifest."""
        return await self._call_mcp_tool("preview_creative", params)

    async def create_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create media buy."""
        return await self._call_mcp_tool("create_media_buy", params)

    async def update_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update media buy."""
        return await self._call_mcp_tool("update_media_buy", params)

    async def build_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Build creative."""
        return await self._call_mcp_tool("build_creative", params)

    async def get_creative_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get creative delivery."""
        return await self._call_mcp_tool("get_creative_delivery", params)

    async def list_transformers(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creative transformers."""
        return await self._call_mcp_tool("list_transformers", params)

    async def list_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List accounts."""
        return await self._call_mcp_tool("list_accounts", params)

    async def sync_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync accounts."""
        return await self._call_mcp_tool("sync_accounts", params)

    async def get_account_financials(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get account financials."""
        return await self._call_mcp_tool("get_account_financials", params)

    async def report_usage(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report account usage."""
        return await self._call_mcp_tool("report_usage", params)

    async def list_tools(self) -> list[str]:
        """List available tools from MCP agent."""
        session = await self._get_session()
        result = await session.list_tools()
        return [tool.name for tool in result.tools]

    async def get_agent_info(self) -> dict[str, Any]:
        """
        Get agent information including AdCP extension metadata from MCP server.

        MCP servers may expose metadata through:
        - Server capabilities exposed during initialization
        - extensions.adcp in server info (if supported)
        - Tool list

        Returns:
            Dictionary with agent metadata
        """
        session = await self._get_session()

        # Extract basic MCP server info
        info: dict[str, Any] = {
            "name": getattr(session, "server_name", None),
            "version": getattr(session, "server_version", None),
            "protocol": "mcp",
        }

        # Get available tools
        try:
            tools_result = await session.list_tools()
            tool_names = [tool.name for tool in tools_result.tools]
            if tool_names:
                info["tools"] = tool_names
        except Exception as e:
            logger.warning(f"Failed to list tools for {self.agent_config.id}: {e}")

        # Try to extract AdCP extension metadata from server capabilities
        # MCP servers may expose this in their initialization response
        capabilities = getattr(session, "_server_capabilities", None)
        if capabilities is not None:
            if isinstance(capabilities, dict):
                extensions = capabilities.get("extensions", {})
                adcp_ext = extensions.get("adcp", {})
                if adcp_ext:
                    info["adcp_version"] = adcp_ext.get("adcp_version")
                    info["protocols_supported"] = adcp_ext.get("protocols_supported")

        logger.info(f"Retrieved agent info for {self.agent_config.id}")
        return info

    async def close(self) -> None:
        """Close the MCP session and clean up resources."""
        if self._session_is_injected:
            return  # caller owns lifecycle; never close an injected session
        await self._cleanup_failed_connection("during close")

    async def close_mcp_session(self, session_id: str | None = None) -> None:
        """Terminate a stateful Streamable HTTP MCP session by id."""
        if self._session_is_injected:
            raise RuntimeError(
                "close_mcp_session is unavailable for from_mcp_client() sessions; "
                "the caller owns the injected transport lifecycle."
            )
        if self.agent_config.mcp_transport != "streamable_http":
            raise TypeError(
                "close_mcp_session is only supported for MCP streamable_http transport; "
                f"got {self.agent_config.mcp_transport!r}."
            )
        if session_id is None:
            session_id = self.current_mcp_session_id()
            if session_id is None:
                raise ValueError(
                    "No active MCP session id is available; pass session_id explicitly "
                    "or call after this client has initialized a Streamable HTTP session."
                )
        if not session_id or any(ch in session_id for ch in ("\r", "\n", "\x00")):
            raise ValueError("session_id must be a non-empty MCP session id header value")
        if not HTTPX_AVAILABLE:
            raise ImportError("httpx is required to close MCP Streamable HTTP sessions")

        headers = self._http_headers()
        headers[MCP_SESSION_ID] = session_id
        timeout = _httpx.Timeout(self.agent_config.timeout)
        httpx_client_factory = self._streamable_http_client_factory()
        urls_to_try = (
            [self._connected_url] if self._connected_url is not None else self._urls_to_try()
        )

        last_error: BaseException | None = None
        for url in urls_to_try:
            try:
                async with httpx_client_factory(headers=headers, timeout=timeout) as client:
                    response = await client.delete(url)
                if response.is_redirect:
                    location = response.headers.get("location")
                    suffix = f" to {location}" if location else ""
                    raise HTTPStatusError(
                        f"Unexpected redirect while closing MCP session{suffix}",
                        request=response.request,
                        response=response,
                    )
                response.raise_for_status()

                current_session_id = self._get_session_id() if self._get_session_id else None
                if current_session_id == session_id:
                    await self._cleanup_failed_connection("after explicit MCP session close")
                return
            except _HTTP_STATUS_ERROR_TYPES as exc:
                last_error = exc
                # Keep fallback behavior symmetrical with session initialization:
                # a 404/405 on one candidate usually means "try the slash variant".
                exc_response = getattr(exc, "response", None)
                status_code = getattr(exc_response, "status_code", None)
                if status_code in (404, 405) and url != urls_to_try[-1]:
                    continue
                break
            except Exception as exc:
                last_error = exc
                if url != urls_to_try[-1]:
                    continue
                break

        raise ADCPConnectionError(
            f"Failed to close MCP session {session_id!r}: {last_error}",
            agent_id=self.agent_config.id,
            agent_uri=self.agent_config.agent_uri,
        ) from last_error

    # ========================================================================
    # V3 Protocol Methods - Protocol Discovery
    # ========================================================================

    async def get_adcp_capabilities(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get AdCP capabilities from the agent."""
        return await self._call_mcp_tool("get_adcp_capabilities", params)

    async def get_task_status(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get task status from the agent."""
        return await self._call_mcp_tool("get_task_status", params)

    async def list_tasks(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List tasks from the agent."""
        return await self._call_mcp_tool("list_tasks", params)

    # ========================================================================
    # V3 Protocol Methods - Content Standards
    # ========================================================================

    async def create_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create content standards configuration."""
        return await self._call_mcp_tool("create_content_standards", params)

    async def get_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get content standards configuration."""
        return await self._call_mcp_tool("get_content_standards", params)

    async def list_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List content standards configurations."""
        return await self._call_mcp_tool("list_content_standards", params)

    async def update_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update content standards configuration."""
        return await self._call_mcp_tool("update_content_standards", params)

    async def calibrate_content(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Calibrate content against standards."""
        return await self._call_mcp_tool("calibrate_content", params)

    async def validate_content_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate content delivery against standards."""
        return await self._call_mcp_tool("validate_content_delivery", params)

    async def get_media_buy_artifacts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get artifacts associated with a media buy."""
        return await self._call_mcp_tool("get_media_buy_artifacts", params)

    # ========================================================================
    # V3 Protocol Methods - Governance
    # ========================================================================

    async def get_creative_features(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Evaluate governance features for a creative."""
        return await self._call_mcp_tool("get_creative_features", params)

    async def sync_plans(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync campaign governance plans."""
        return await self._call_mcp_tool("sync_plans", params)

    async def check_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Check an action against campaign governance."""
        return await self._call_mcp_tool("check_governance", params)

    async def report_plan_outcome(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report the outcome of a governed action."""
        return await self._call_mcp_tool("report_plan_outcome", params)

    async def get_plan_audit_logs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Retrieve governance audit logs for plans."""
        return await self._call_mcp_tool("get_plan_audit_logs", params)

    # ========================================================================
    # V3 Protocol Methods - Sponsored Intelligence
    # ========================================================================

    async def si_get_offering(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get sponsored intelligence offering."""
        return await self._call_mcp_tool("si_get_offering", params)

    async def si_initiate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Initiate sponsored intelligence session."""
        return await self._call_mcp_tool("si_initiate_session", params)

    async def si_send_message(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Send message in sponsored intelligence session."""
        return await self._call_mcp_tool("si_send_message", params)

    async def si_terminate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Terminate sponsored intelligence session."""
        return await self._call_mcp_tool("si_terminate_session", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Property Lists)
    # ========================================================================

    async def create_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a property list for governance."""
        return await self._call_mcp_tool("create_property_list", params)

    async def get_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a property list with optional resolution."""
        return await self._call_mcp_tool("get_property_list", params)

    async def list_property_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List property lists."""
        return await self._call_mcp_tool("list_property_lists", params)

    async def update_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a property list."""
        return await self._call_mcp_tool("update_property_list", params)

    async def delete_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a property list."""
        return await self._call_mcp_tool("delete_property_list", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Collection Lists)
    # ========================================================================

    async def create_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a collection list for governance."""
        return await self._call_mcp_tool("create_collection_list", params)

    async def get_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a collection list with optional resolution."""
        return await self._call_mcp_tool("get_collection_list", params)

    async def list_collection_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List collection lists."""
        return await self._call_mcp_tool("list_collection_lists", params)

    async def update_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a collection list."""
        return await self._call_mcp_tool("update_collection_list", params)

    async def delete_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a collection list."""
        return await self._call_mcp_tool("delete_collection_list", params)

    # ========================================================================
    # V3 Protocol Methods - Governance (Sync Governance)
    # ========================================================================

    async def sync_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync governance agents attached to an account."""
        return await self._call_mcp_tool("sync_governance", params)

    # ========================================================================
    # V3 Protocol Methods - TMP
    # ========================================================================

    async def context_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match ad context to buyer packages."""
        return await self._call_mcp_tool("context_match", params)

    async def identity_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match user identity for package eligibility."""
        return await self._call_mcp_tool("identity_match", params)

    # ========================================================================
    # V3 Protocol Methods - Brand Rights
    # ========================================================================

    async def get_brand_identity(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get brand identity information."""
        return await self._call_mcp_tool("get_brand_identity", params)

    async def get_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get available rights for licensing."""
        return await self._call_mcp_tool("get_rights", params)

    async def acquire_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Acquire rights for brand content usage."""
        return await self._call_mcp_tool("acquire_rights", params)

    async def update_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update terms of an existing rights acquisition."""
        return await self._call_mcp_tool("update_rights", params)

    async def validate_input(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate creative input."""
        return await self._call_mcp_tool("validate_input", params)

    async def verify_brand_claim(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify a brand claim."""
        return await self._call_mcp_tool("verify_brand_claim", params)

    async def verify_brand_claims(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify brand claims."""
        return await self._call_mcp_tool("verify_brand_claims", params)

    # ========================================================================
    # V3 Protocol Methods - Compliance
    # ========================================================================

    async def comply_test_controller(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Compliance test controller (sandbox only)."""
        return await self._call_mcp_tool("comply_test_controller", params)

class ProtocolAdapter(ABC):
    """
    Base class for protocol adapters.

    Each adapter implements the ADCP protocol methods and handles
    protocol-specific translation (MCP/A2A) while returning properly
    typed responses.
    """

    def __init__(self, agent_config: AgentConfig):
        """Initialize adapter with agent configuration."""
        self.agent_config = agent_config
        # Optional hook; ADCPClient sets this when strict_idempotency is enabled.
        # Invoked before each mutating tool call to verify the seller declared
        # adcp.idempotency.replay_ttl_seconds in capabilities. None = no check.
        self.idempotency_capability_check: Callable[[], Awaitable[None]] | None = None
        # Unique token for this adapter's owning client — used to scope
        # ``use_idempotency_key`` so a key pinned on one client does not bleed
        # to sibling clients (cross-seller correlation risk per AdCP #2315).
        self.idempotency_client_token: str | None = None
        # Optional httpx request event hook. ADCPClient installs one when a
        # SigningConfig is present; the hook attaches RFC 9421 Signature-Input
        # / Signature / Content-Digest headers to outgoing requests that the
        # seller's capability policy says should be signed. A2A consumes this
        # via its httpx client's event_hooks; MCP consumes it via a custom
        # httpx_client_factory passed to streamablehttp_client.
        self.signing_request_hook: Callable[[httpx.Request], Awaitable[None]] | None = None
        # Schema validation modes — resolved by the owning ADCPClient via
        # ``configure_validation``. Class defaults match the TS port: warn
        # on requests (don't block partial payloads in error-path tests),
        # strict on responses (agent drift fails the task on first call).
        # Adapters instantiated directly without an ADCPClient inherit
        # these defaults; production callers flip responses to warn via
        # ``ADCPClient(validation=...)`` or an env override.
        self.request_validation_mode: ValidationMode = "warn"
        self.response_validation_mode: ValidationMode = "strict"
        # Optional hook applied to every outbound request params dict
        # before validation/send. The owning ADCPClient installs one to
        # auto-inject ``adcp_version`` from the per-instance pin. Returns
        # a new dict (the original is not mutated). Caller-supplied
        # values on the original dict win — the enricher is the default,
        # not an override.
        #
        # Contract: the validator runs on the enriched dict, so any field
        # the enricher injects must be either (a) declared in the request
        # schema, or (b) tolerated by the schema's ``additionalProperties``
        # policy. Top-level Request models in this SDK declare
        # ``extra="allow"`` (see ``AdCPBaseModel`` overrides in generated
        # types) — flipping any of them to ``extra="forbid"`` would break
        # this assumption silently.
        self.envelope_enricher: Callable[[dict[str, Any]], dict[str, Any]] | None = None

    def _enrich_outgoing_params(self, params: Any) -> Any:
        """Apply ``envelope_enricher`` to an outbound params dict.

        No-op for non-dict params (rare — most tool methods pass dicts
        from ``model_dump()``) and when no enricher is installed.
        """
        if self.envelope_enricher is None or not isinstance(params, dict):
            return params
        return self.envelope_enricher(params)

    def configure_validation(self, config: ValidationHookConfig | None) -> None:
        """Apply a client's :class:`ValidationHookConfig` to this adapter."""
        from adcp.validation.client_hooks import resolve_validation_modes

        req, resp = resolve_validation_modes(config)
        self.request_validation_mode = req
        self.response_validation_mode = resp

    # ========================================================================
    # Helper methods for response parsing
    # ========================================================================

    def _parse_response(
        self, raw_result: TaskResult[Any], response_type: type[T] | Any
    ) -> TaskResult[T]:
        """
        Parse raw TaskResult into typed TaskResult.

        Handles both MCP content arrays and A2A dict responses.
        Supports both single types and Union types (for oneOf discriminated unions).

        Args:
            raw_result: Raw TaskResult from adapter
            response_type: Expected Pydantic response type (can be a Union type)

        Returns:
            Typed TaskResult
        """
        # Handle failed results or interim states without data
        # For A2A: interim states (submitted/working) have data=None but success=True
        # For MCP: completed tasks always have data, missing data indicates failure
        if not raw_result.success or raw_result.data is None:
            # If already marked as unsuccessful, preserve that
            # If successful but no data (A2A interim state), preserve success=True
            return TaskResult[T](
                status=raw_result.status,
                data=None,
                message=raw_result.message,
                success=raw_result.success,  # Preserve original success state
                error=raw_result.error,  # Only use error if one was set
                metadata=raw_result.metadata,
                debug_info=raw_result.debug_info,
                idempotency_key=raw_result.idempotency_key,
                replayed=raw_result.replayed,
            )

        try:
            # Handle MCP content arrays
            if isinstance(raw_result.data, list):
                parsed_data = parse_mcp_content(raw_result.data, response_type)
            else:
                # Handle A2A or direct responses
                parsed_data = parse_json_or_text(raw_result.data, response_type)

            return TaskResult[T](
                status=raw_result.status,
                data=parsed_data,
                message=raw_result.message,  # Preserve human-readable message from protocol
                success=raw_result.success,
                error=raw_result.error,
                metadata=raw_result.metadata,
                debug_info=raw_result.debug_info,
                idempotency_key=raw_result.idempotency_key,
                replayed=raw_result.replayed,
            )
        except ValueError as e:
            # Parsing failed - return error result. Preserve idempotency_key
            # and replayed so callers can still correlate/suppress side-effects
            # even when response parsing fails.
            return TaskResult[T](
                status=TaskStatus.FAILED,
                error=f"Failed to parse response: {e}",
                message=raw_result.message,
                success=False,
                debug_info=raw_result.debug_info,
                idempotency_key=raw_result.idempotency_key,
                replayed=raw_result.replayed,
            )

    # ========================================================================
    # ADCP Protocol Methods - Type-safe, spec-aligned interface
    # Each adapter MUST implement these methods explicitly.
    # ========================================================================

    @abstractmethod
    async def get_products(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get advertising products."""
        pass

    @abstractmethod
    async def list_creative_formats(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List supported creative formats."""
        pass

    @abstractmethod
    async def sync_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync creatives."""
        pass

    @abstractmethod
    async def list_creatives(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creatives."""
        pass

    @abstractmethod
    async def get_media_buy_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buy delivery."""
        pass

    @abstractmethod
    async def get_media_buys(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get media buys with status, creative approval state, and optional delivery snapshots."""
        pass

    @abstractmethod
    async def get_signals(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get signals."""
        pass

    @abstractmethod
    async def activate_signal(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Activate signal."""
        pass

    @abstractmethod
    async def provide_performance_feedback(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Provide performance feedback."""
        pass

    @abstractmethod
    async def log_event(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Log event."""
        pass

    @abstractmethod
    async def sync_event_sources(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync event sources."""
        pass

    @abstractmethod
    async def sync_audiences(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync audiences."""
        pass

    @abstractmethod
    async def sync_catalogs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync catalogs."""
        pass

    @abstractmethod
    async def create_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create media buy."""
        pass

    @abstractmethod
    async def update_media_buy(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update media buy."""
        pass

    @abstractmethod
    async def build_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Build creative."""
        pass

    @abstractmethod
    async def preview_creative(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Preview creative."""
        pass

    @abstractmethod
    async def validate_input(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate creative input against a format declaration."""
        pass

    @abstractmethod
    async def get_creative_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get creative delivery."""
        pass

    @abstractmethod
    async def list_transformers(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List creative transformers."""
        pass

    @abstractmethod
    async def list_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List accounts."""
        pass

    @abstractmethod
    async def sync_accounts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync accounts."""
        pass

    @abstractmethod
    async def get_account_financials(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get account financials."""
        pass

    @abstractmethod
    async def report_usage(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report account usage."""
        pass

    @abstractmethod
    async def list_tools(self) -> list[str]:
        """
        List available tools from the agent.

        Returns:
            List of tool names
        """
        pass

    @abstractmethod
    async def get_agent_info(self) -> dict[str, Any]:
        """
        Get agent information including AdCP extension metadata.

        Returns agent card information including:
        - Agent name, description, version
        - AdCP version (from extensions.adcp.adcp_version)
        - Supported protocols (from extensions.adcp.protocols_supported)
        - Available tools/skills

        Returns:
            Dictionary with agent metadata including AdCP extension fields
        """
        pass

    @abstractmethod
    async def close(self) -> None:
        """
        Close the adapter and clean up resources.

        Implementations should close any open connections, clients, or other resources.
        """
        pass

    # ========================================================================
    # V3 Protocol Methods - Protocol Discovery
    # ========================================================================

    @abstractmethod
    async def get_adcp_capabilities(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get AdCP capabilities from the agent."""
        pass

    @abstractmethod
    async def get_task_status(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get task status from the agent."""
        pass

    @abstractmethod
    async def list_tasks(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List tasks from the agent."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Content Standards
    # ========================================================================

    @abstractmethod
    async def create_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create content standards configuration."""
        pass

    @abstractmethod
    async def get_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get content standards configuration."""
        pass

    @abstractmethod
    async def list_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List content standards configurations."""
        pass

    @abstractmethod
    async def update_content_standards(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update content standards configuration."""
        pass

    @abstractmethod
    async def calibrate_content(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Calibrate content against standards."""
        pass

    @abstractmethod
    async def validate_content_delivery(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Validate content delivery against standards."""
        pass

    @abstractmethod
    async def get_media_buy_artifacts(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get artifacts associated with a media buy."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Governance
    # ========================================================================

    @abstractmethod
    async def get_creative_features(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Evaluate governance features for a creative."""
        pass

    @abstractmethod
    async def sync_plans(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync campaign governance plans."""
        pass

    @abstractmethod
    async def check_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Check an action against campaign governance."""
        pass

    @abstractmethod
    async def report_plan_outcome(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Report the outcome of a governed action."""
        pass

    @abstractmethod
    async def get_plan_audit_logs(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Retrieve governance audit logs for plans."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Sponsored Intelligence
    # ========================================================================

    @abstractmethod
    async def si_get_offering(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get sponsored intelligence offering."""
        pass

    @abstractmethod
    async def si_initiate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Initiate sponsored intelligence session."""
        pass

    @abstractmethod
    async def si_send_message(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Send message in sponsored intelligence session."""
        pass

    @abstractmethod
    async def si_terminate_session(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Terminate sponsored intelligence session."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Governance (Property Lists)
    # ========================================================================

    @abstractmethod
    async def create_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a property list for governance."""
        pass

    @abstractmethod
    async def get_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a property list with optional resolution."""
        pass

    @abstractmethod
    async def list_property_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List property lists."""
        pass

    @abstractmethod
    async def update_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a property list."""
        pass

    @abstractmethod
    async def delete_property_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a property list."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Governance (Collection Lists)
    # ========================================================================

    @abstractmethod
    async def create_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Create a collection list for governance."""
        pass

    @abstractmethod
    async def get_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get a collection list with optional resolution."""
        pass

    @abstractmethod
    async def list_collection_lists(self, params: dict[str, Any]) -> TaskResult[Any]:
        """List collection lists."""
        pass

    @abstractmethod
    async def update_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update a collection list."""
        pass

    @abstractmethod
    async def delete_collection_list(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Delete a collection list."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Governance (Sync Governance)
    # ========================================================================

    @abstractmethod
    async def sync_governance(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Sync governance agents attached to an account."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Temporal Matching Protocol (TMP)
    # ========================================================================

    @abstractmethod
    async def context_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match ad context to buyer packages."""
        pass

    @abstractmethod
    async def identity_match(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Match user identity for package eligibility."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Brand Rights
    # ========================================================================

    @abstractmethod
    async def get_brand_identity(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get brand identity information."""
        pass

    @abstractmethod
    async def get_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Get available rights for licensing."""
        pass

    @abstractmethod
    async def acquire_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Acquire rights for brand content usage."""
        pass

    @abstractmethod
    async def update_rights(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Update terms of an existing rights acquisition."""
        pass

    @abstractmethod
    async def verify_brand_claim(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify a single brand claim."""
        pass

    @abstractmethod
    async def verify_brand_claims(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Verify multiple brand claims."""
        pass

    # ========================================================================
    # V3 Protocol Methods - Compliance
    # ========================================================================

    @abstractmethod
    async def comply_test_controller(self, params: dict[str, Any]) -> TaskResult[Any]:
        """Compliance test controller (sandbox only)."""
        pass