From 6065d23e1e0d71b8aa065528e8d85343af6a9771 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 24 Aug 2026 15:48:10 +0200 Subject: [PATCH] feat(observability): add optional OpenTelemetry tracing --- README.md | 37 ++ pyproject.toml | 8 + src/adcp/__init__.py | 9 + src/adcp/client.py | 85 ++++- src/adcp/observability.py | 308 ++++++++++++++++ src/adcp/protocols/a2a.py | 9 +- src/adcp/protocols/base.py | 4 + src/adcp/protocols/mcp.py | 79 ++-- tests/fixtures/public_api_snapshot.json | 3 + tests/test_observability.py | 469 ++++++++++++++++++++++++ tests/test_security_hardening.py | 6 +- tests/type_checks/observability.py | 12 + 12 files changed, 978 insertions(+), 51 deletions(-) create mode 100644 src/adcp/observability.py create mode 100644 tests/test_observability.py create mode 100644 tests/type_checks/observability.py diff --git a/README.md b/README.md index 17ce9dfa2..f22489cd1 100644 --- a/README.md +++ b/README.md @@ -828,6 +828,43 @@ Multi-agent fan-out intentionally does not yet accept it because one timeout exception cannot safely represent several sellers' independent mutation outcomes and idempotency keys. +### Optional OpenTelemetry tracing + +AdCP client calls create one OpenTelemetry `CLIENT` span when an application +has configured an OpenTelemetry SDK provider. The library configures no SDK, +exporter, endpoint, or credentials itself, so its default behavior remains a +non-recording no-op. + +```bash +pip install "adcp[observability]" opentelemetry-sdk +``` + +Configure the provider/exporter once in application startup using the normal +OpenTelemetry Python APIs. The SDK then emits `adcp.mcp.call_tool` or +`adcp.a2a.call_tool` spans with bounded attributes for the agent ID, protocol, +tool name (`adcp.tool` plus the `adcp.tool.name` compatibility alias), standard +RPC system/method fields, task status, and success flag. Multi-call helpers use +an `adcp.client.workflow` parent with one child CLIENT span per wire call. +Request parameters, response bodies, credentials, idempotency keys, and remote +error prose are never attached; invalid or oversized identifiers become +`unknown` instead of being truncated into telemetry. + +Active W3C `traceparent`/`tracestate` values propagate on actual tool requests; +agent-card and connection discovery requests are excluded. AdCP deliberately +does not propagate OpenTelemetry baggage across the agent boundary. + +```python +from adcp import ADCPClient, is_tracing_available + +client = ADCPClient(config) +assert is_tracing_available() # API installed; exporting still needs a provider +result = await client.get_products(request) +``` + +The `get_tracer()` and `inject_trace_headers()` exports are available for +custom integrations. Libraries should depend only on `opentelemetry-api`; +applications own the SDK and exporter configuration. + ### Error Handling The library provides a comprehensive exception hierarchy with helpful error messages: diff --git a/pyproject.toml b/pyproject.toml index 450ffc84e..8635c6ed6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -145,6 +145,9 @@ dev = [ # src/app.py and seed.py. "alembic>=1.13.0", "pre-commit>=4.4.0", + # In-memory span exporter used by the optional observability tests. The + # library itself imports only opentelemetry-api and never configures an SDK. + "opentelemetry-sdk>=1.25,<2", ] docs = [ "pdoc3>=0.10.0", @@ -157,6 +160,11 @@ pg = [ "psycopg[binary]>=3.1.0", "psycopg-pool>=3.2.0", ] +observability = [ + # API-only library instrumentation: without an application-configured SDK + # provider/exporter, OpenTelemetry remains a non-recording no-op. + "opentelemetry-api>=1.25,<2", +] [project.urls] Homepage = "https://github.com/adcontextprotocol/adcp-client-python" diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 3ffb272a0..7b00d1cf8 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -144,6 +144,11 @@ def _resolve_version() -> str: "TaskOptions", "TaskRecoveryMetadata", ), + "adcp.observability": ( + "get_tracer", + "inject_trace_headers", + "is_tracing_available", + ), "adcp.exceptions": ( "AdagentsAccessBlockedError", "AdagentsNotFoundError", @@ -849,6 +854,9 @@ def get_adcp_version() -> str: "Checkpoint", "TaskOptions", "TaskRecoveryMetadata", + "get_tracer", + "inject_trace_headers", + "is_tracing_available", "RegistryClient", "PropertyRegistry", "RegistrySync", @@ -1551,6 +1559,7 @@ def get_adcp_version() -> str: FeedStateStore, RefreshResult, ) + from adcp.observability import get_tracer, inject_trace_headers, is_tracing_available from adcp.property_registry import PropertyRegistry from adcp.registry import RegistryClient from adcp.registry_sync import ( diff --git a/src/adcp/client.py b/src/adcp/client.py index f29a0ef55..baff8717d 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -6,6 +6,7 @@ import contextlib import hashlib import hmac +import inspect import json import logging import os @@ -49,6 +50,15 @@ refine_proposals_verified as _refine_proposals_verified, ) from adcp.negotiation import verify_refinement_result as _verify_refinement_result +from adcp.observability import ( + client_task_span, + inject_trace_context, + is_tracing_available, + mcp_trace_headers_from_payload, + safe_tool_name, + set_request_trace_headers, + set_task_result_attributes, +) from adcp.protocols.a2a import A2AAdapter from adcp.protocols.base import ProtocolAdapter from adcp.protocols.mcp import MCPAdapter, MCPHttpxClientFactory @@ -411,18 +421,41 @@ async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R: client = cast("ADCPClient", args[0]) if options is not None and not isinstance(options, TaskOptions): raise TypeError("options must be a TaskOptions instance") - if options is None: - return await method(*args, **kwargs) - task_name = method.__name__ - if task_name in {"execute_task", "execute_task_legacy"}: + method_name = method.__name__ + workflow = method_name in _TRACE_WORKFLOW_METHODS + task_name = method_name + if method_name in {"execute_task", "execute_task_legacy"}: requested_name = args[1] if len(args) > 1 else call_kwargs.get("task_name") - if isinstance(requested_name, str): + allowed_names = ( + _CANONICAL_EXECUTE_TASKS + if method_name == "execute_task" + else _LEGACY_CREATIVE_TASKS + ) + if isinstance(requested_name, str) and requested_name in allowed_names: task_name = requested_name - return await client._run_with_task_options( - task_name, - lambda: method(*args, **kwargs), - options, - ) + else: + task_name = "unknown" + elif method_name.endswith("_legacy"): + task_name = method_name.removesuffix("_legacy") + task_name = safe_tool_name(task_name) + protocol = client.agent_config.protocol.value + with client_task_span( + client._task_options_token, + protocol=protocol, + task_name=task_name, + agent_id=client.agent_config.id, + workflow=workflow, + ) as span: + if options is None: + result = await method(*args, **kwargs) + else: + result = await client._run_with_task_options( + task_name, + lambda: method(*args, **kwargs), + options, + ) + set_task_result_attributes(span, result) + return result return wrapped @@ -445,6 +478,15 @@ async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R: _LEGACY_ONLY_CREATIVE_TASKS = frozenset( {"build_creative", "list_creative_formats", "preview_creative"} ) +_TRACE_WORKFLOW_METHODS = frozenset({"refine_proposals_verified", "wait_for_refinement_verified"}) +_PROTOCOL_NON_TASK_METHODS = frozenset({"close", "get_agent_info", "list_tools"}) +_CANONICAL_EXECUTE_TASKS = frozenset( + name + for name, member in inspect.getmembers(ProtocolAdapter, inspect.iscoroutinefunction) + if not name.startswith("_") + and name not in _PROTOCOL_NON_TASK_METHODS + and name not in _LEGACY_ONLY_CREATIVE_TASKS +) class Checkpoint(TypedDict): @@ -739,6 +781,8 @@ def __init__( else: raise ValueError(f"Unsupported protocol: {agent_config.protocol}") self.adapter.task_options_client_token = self._task_options_token + if is_tracing_available(): + self.adapter.tracing_request_hook = self._inject_outgoing_trace_context self.adapter.idempotency_client_token = self._idempotency_client_token if strict_idempotency: @@ -1138,6 +1182,23 @@ async def _prepare_signing_capabilities(self) -> None: """Populate signing policy before a transport writer sends a request.""" await self.fetch_capabilities() + async def _inject_outgoing_trace_context(self, request: httpx.Request) -> None: + """Inject trace context into tool requests, excluding discovery traffic.""" + + try: + payload = json.loads(request.content) + except (json.JSONDecodeError, TypeError, UnicodeDecodeError): + payload = None + bridged_headers = mcp_trace_headers_from_payload(payload) + if bridged_headers: + set_request_trace_headers(request, bridged_headers) + return + mcp_operation = self._mcp_operation_from_request(request) + operation = mcp_operation or _signing_current_operation.get() + if operation is None: + return + inject_trace_context(request) + @staticmethod def _mcp_operation_from_request(request: httpx.Request) -> str | None: """Extract one MCP ``tools/call`` name from its JSON-RPC body.""" @@ -5278,9 +5339,9 @@ async def execute_task( f"{task_name} request contains legacy creative identity; generic execute_task " "is canonical-only" ) - method = getattr(self, task_name, None) - if method is None or not callable(method) or task_name.endswith("_legacy"): + if task_name not in _CANONICAL_EXECUTE_TASKS: raise ValueError(f"Unknown canonical AdCP task: {task_name}") + method = getattr(self, task_name) return cast(TaskResult[Any], await method(request)) @_task_options_method diff --git a/src/adcp/observability.py b/src/adcp/observability.py new file mode 100644 index 000000000..1b24e26e4 --- /dev/null +++ b/src/adcp/observability.py @@ -0,0 +1,308 @@ +"""Optional OpenTelemetry tracing for AdCP client calls. + +The module imports :mod:`opentelemetry-api` lazily. Without the optional +dependency every helper is a no-op; with the API but no SDK/provider installed, +OpenTelemetry's own non-recording provider keeps the same no-op behavior. + +Only W3C Trace Context is propagated. Baggage is intentionally excluded from +the cross-agent boundary because application baggage can contain sensitive or +high-cardinality values. +""" + +from __future__ import annotations + +import re +import threading +from collections.abc import Iterator, Mapping, MutableMapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Any + +_INSTRUMENTATION_NAME = "adcp" +MCP_TRACEPARENT_META_KEY = "io.adcontextprotocol/traceparent" +MCP_TRACESTATE_META_KEY = "io.adcontextprotocol/tracestate" +_TRACEPARENT_RE = re.compile( + r"^(?!ff)[0-9a-f]{2}-(?!0{32})[0-9a-f]{32}-(?!0{16})[0-9a-f]{16}-[0-9a-f]{2}$" +) +_SAFE_TOOL_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +_SAFE_AGENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") + + +@dataclass(frozen=True, slots=True) +class _OpenTelemetryBindings: + trace: Any + span_kind: Any + status: Any + status_code: Any + trace_context_propagator: Any + + +_bindings: _OpenTelemetryBindings | None = None +_bindings_checked = False +_bindings_lock = threading.Lock() + + +def _import_bindings() -> _OpenTelemetryBindings | None: + try: + from opentelemetry import trace + from opentelemetry.trace import SpanKind, Status, StatusCode + from opentelemetry.trace.propagation.tracecontext import ( + TraceContextTextMapPropagator, + ) + except ImportError: + return None + return _OpenTelemetryBindings( + trace=trace, + span_kind=SpanKind, + status=Status, + status_code=StatusCode, + trace_context_propagator=TraceContextTextMapPropagator, + ) + + +def _load_bindings() -> _OpenTelemetryBindings | None: + """Load the OpenTelemetry API once, returning ``None`` when absent.""" + + global _bindings, _bindings_checked + if _bindings_checked: + return _bindings + with _bindings_lock: + if not _bindings_checked: + # Publish ``checked`` only after the value is complete. Client + # construction is synchronous but may occur on several threads. + _bindings = _import_bindings() + _bindings_checked = True + return _bindings + + +def is_tracing_available() -> bool: + """Return whether the optional OpenTelemetry API is installed. + + Availability does not imply that spans are exported. Applications remain + responsible for installing/configuring an OpenTelemetry SDK and exporter. + """ + + return _load_bindings() is not None + + +def get_tracer() -> Any | None: + """Return the AdCP tracer, or ``None`` without OpenTelemetry installed.""" + + bindings = _load_bindings() + if bindings is None: + return None + return bindings.trace.get_tracer(_INSTRUMENTATION_NAME) + + +def inject_trace_headers( + carrier: MutableMapping[str, str] | None = None, +) -> dict[str, str]: + """Inject the active W3C trace context into a string header mapping. + + The returned dictionary contains the full resulting carrier. Existing + ``traceparent`` and ``tracestate`` values are replaced only when an active, + valid OpenTelemetry span produces new values. W3C baggage is never + propagated by this helper. + """ + + result = {name: value for name, value in (carrier or {}).items() if name.lower() != "baggage"} + bindings = _load_bindings() + if bindings is None: + return result + + injected: dict[str, str] = {} + bindings.trace_context_propagator().inject(injected) + if not injected: + return result + for existing_name in tuple(result): + if existing_name.lower() in {"traceparent", "tracestate"}: + result.pop(existing_name) + for name in ("traceparent", "tracestate"): + value = injected.get(name) + if value is not None: + result[name] = value + return result + + +_active_client_span: ContextVar[tuple[object, str, str] | None] = ContextVar( + "adcp_active_client_span", default=None +) + + +def _safe_attributes(attributes: Mapping[str, object]) -> dict[str, str | int | float | bool]: + allowed: dict[str, str | int | float | bool] = {} + for key, value in attributes.items(): + if isinstance(value, str): + allowed[key] = value if len(value) <= 128 and value.isprintable() else "unknown" + elif isinstance(value, (int, float, bool)): + allowed[key] = value + return allowed + + +def safe_tool_name(value: object) -> str: + """Return a bounded telemetry-safe wire task name.""" + + return value if isinstance(value, str) and _SAFE_TOOL_RE.fullmatch(value) else "unknown" + + +def _safe_agent_id(value: object) -> str: + return value if isinstance(value, str) and _SAFE_AGENT_RE.fullmatch(value) else "unknown" + + +@contextmanager +def client_task_span( + client_token: object, + *, + protocol: str, + task_name: str, + agent_id: str, + workflow: bool = False, +) -> Iterator[Any | None]: + """Start one safe CLIENT span for an outermost client task. + + Nested public methods on the same client share the existing span. Internal + exception recording deliberately stores only the exception type, never its + message, response body, request parameters, credentials, or remote prose. + """ + + span_type = "workflow" if workflow else "wire" + safe_tool = safe_tool_name(task_name) + active = _active_client_span.get() + if active is not None and active[0] is client_token: + # Generic execute_task delegates to the same public wire method: retain + # one span. Distinct nested tasks are separate logical RPCs (for + # example a cold capability preflight), so each gets a child CLIENT + # span. Workflows remain one INTERNAL parent around their wire calls. + if workflow or (active[1] == "wire" and active[2] == safe_tool): + yield None + return + + bindings = _load_bindings() + tracer = get_tracer() + if bindings is None or tracer is None: + yield None + return + + token = _active_client_span.set((client_token, span_type, safe_tool)) + safe_agent = _safe_agent_id(agent_id) + if workflow: + span_name = "adcp.client.workflow" + span_kind = bindings.span_kind.INTERNAL + attributes = _safe_attributes( + { + "adcp.agent.id": safe_agent, + "adcp.protocol": protocol, + "adcp.workflow.name": safe_tool, + } + ) + else: + span_name = f"adcp.{protocol}.call_tool" + span_kind = bindings.span_kind.CLIENT + attributes = _safe_attributes( + { + "rpc.system.name": "adcp", + "rpc.method": safe_tool, + # `adcp.tool` matches the released JS SDK. Keep the more + # explicit Python key as an additive compatibility alias. + "adcp.tool": safe_tool, + "adcp.tool.name": safe_tool, + "adcp.agent.id": safe_agent, + "adcp.protocol": protocol, + } + ) + try: + with tracer.start_as_current_span( + span_name, + kind=span_kind, + attributes=attributes, + record_exception=False, + set_status_on_exception=False, + ) as span: + try: + yield span + except BaseException as exc: + span.set_attribute("error.type", safe_tool_name(type(exc).__name__.lower())) + span.set_status(bindings.status(bindings.status_code.ERROR)) + raise + finally: + _active_client_span.reset(token) + + +def set_task_result_attributes(span: Any | None, result: object) -> None: + """Annotate a task span from public, bounded result metadata only.""" + + if span is None: + return + success = getattr(result, "success", None) + if isinstance(success, bool): + span.set_attribute("adcp.task.success", success) + status = getattr(result, "status", None) + status_value = getattr(status, "value", status) + if isinstance(status_value, str): + span.set_attribute("adcp.task.status", status_value) + if success is False: + span.set_attribute("error.type", "adcp.task.failed") + bindings = _load_bindings() + if bindings is not None: + span.set_status(bindings.status(bindings.status_code.ERROR)) + + +def set_request_trace_headers(request: Any, carrier: Mapping[str, str]) -> None: + """Apply validated W3C trace headers to an httpx-compatible request.""" + + if "traceparent" not in carrier: + return + request.headers.pop("traceparent", None) + request.headers.pop("tracestate", None) + for name in ("traceparent", "tracestate"): + value = carrier.get(name) + if value is not None: + request.headers.pop(name, None) + request.headers[name] = value + + +def inject_trace_context(request: Any) -> None: + """Inject active W3C trace headers into an httpx-compatible request.""" + + set_request_trace_headers(request, inject_trace_headers()) + + +def mcp_trace_meta() -> dict[str, str] | None: + """Capture active trace context for MCP's long-lived writer task.""" + + headers = inject_trace_headers() + if "traceparent" not in headers: + return None + meta = {MCP_TRACEPARENT_META_KEY: headers["traceparent"]} + if "tracestate" in headers: + meta[MCP_TRACESTATE_META_KEY] = headers["tracestate"] + return meta + + +def mcp_trace_headers_from_payload(payload: object) -> dict[str, str]: + """Read the SDK's bounded trace bridge from an MCP JSON-RPC payload.""" + + if not isinstance(payload, dict): + return {} + params = payload.get("params") + if not isinstance(params, dict): + return {} + meta = params.get("_meta") + if not isinstance(meta, dict): + return {} + traceparent = meta.get(MCP_TRACEPARENT_META_KEY) + if not isinstance(traceparent, str) or _TRACEPARENT_RE.fullmatch(traceparent) is None: + return {} + result = {"traceparent": traceparent} + tracestate = meta.get(MCP_TRACESTATE_META_KEY) + if ( + isinstance(tracestate, str) + and len(tracestate) <= 512 + and all(0x20 <= ord(character) <= 0x7E for character in tracestate) + ): + result["tracestate"] = tracestate + return result + + +__all__ = ["get_tracer", "inject_trace_headers", "is_tracing_available"] diff --git a/src/adcp/protocols/a2a.py b/src/adcp/protocols/a2a.py index ee90b45b9..063fb8069 100644 --- a/src/adcp/protocols/a2a.py +++ b/src/adcp/protocols/a2a.py @@ -260,9 +260,16 @@ async def _get_httpx_client(self) -> httpx.AsyncClient: "timeout": self.agent_config.timeout, "trust_env": False, } + request_hooks: list[Any] = [] + if self.tracing_request_hook is not None: + # Trace context must exist before signing so deployments that + # cover it with RFC 9421 sign the bytes actually sent. + request_hooks.append(self.tracing_request_hook) if self.signing_request_hook is not None: - event_hooks["request"] = [self.signing_request_hook] + request_hooks.append(self.signing_request_hook) client_kwargs["follow_redirects"] = False + if request_hooks: + event_hooks["request"] = request_hooks if event_hooks: client_kwargs["event_hooks"] = event_hooks diff --git a/src/adcp/protocols/base.py b/src/adcp/protocols/base.py index 9aabcd1c7..f6c2b6198 100644 --- a/src/adcp/protocols/base.py +++ b/src/adcp/protocols/base.py @@ -40,6 +40,10 @@ def __init__(self, agent_config: AgentConfig): self.idempotency_client_token: str | None = None # Owning client identity for task-local deadline/recovery attribution. self.task_options_client_token: object | None = None + # Optional OpenTelemetry request hook. The owning client installs it + # when the OTel API is importable; it injects only active W3C trace + # context for actual tool calls (never discovery traffic). + self.tracing_request_hook: Callable[[httpx.Request], Awaitable[None]] | 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 diff --git a/src/adcp/protocols/mcp.py b/src/adcp/protocols/mcp.py index a6eb2720d..a426e7330 100644 --- a/src/adcp/protocols/mcp.py +++ b/src/adcp/protocols/mcp.py @@ -6,9 +6,9 @@ import contextlib import logging import time -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Sequence from contextlib import AsyncExitStack, asynccontextmanager -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from urllib.parse import urlparse # ExceptionGroup and BaseExceptionGroup are available in Python 3.11+ @@ -25,6 +25,7 @@ if TYPE_CHECKING: from mcp import ClientSession + from mcp.types import RequestParamsMeta try: import anyio @@ -71,6 +72,7 @@ IdempotencyConflictError, IdempotencyExpiredError, ) +from adcp.observability import mcp_trace_meta from adcp.protocols._adcp_errors import validate_adcp_error as _validate_adcp_error from adcp.protocols.base import ProtocolAdapter from adcp.signing.autosign import current_operation as _signing_operation @@ -97,7 +99,9 @@ """ -def _make_hardened_mcp_http_factory() -> Callable[..., Any]: +def _make_hardened_mcp_http_factory( + request_hooks: Sequence[Callable[[Any], Awaitable[None]]] = (), +) -> Callable[..., Any]: """Build an MCP HTTP client factory with fail-closed network defaults.""" def factory( @@ -114,6 +118,8 @@ def factory( "follow_redirects": False, "trust_env": False, } + if request_hooks: + kwargs["event_hooks"] = {"request": list(request_hooks)} kwargs["timeout"] = _coerce_mcp_timeout(timeout) if headers is not None: kwargs["headers"] = headers @@ -149,33 +155,12 @@ def _make_signing_http_factory( the original ``@authority``. """ - def factory( - headers: dict[str, str] | None = None, - timeout: Any = None, - auth: Any = None, - **extra: Any, - ) -> Any: - # Forward any future MCP-SDK kwargs (e.g. verify=, cert=) verbatim - # so adding a new factory parameter upstream doesn't break signing. - kwargs: dict[str, Any] = { - **extra, - "follow_redirects": False, - "event_hooks": {"request": [hook]}, - "trust_env": False, - } - kwargs["timeout"] = _coerce_mcp_timeout(timeout) - if headers is not None: - kwargs["headers"] = headers - if auth is not None: - kwargs["auth"] = auth - return _mcp_httpx.AsyncClient(**kwargs) - - return factory + return _make_hardened_mcp_http_factory((hook,)) def _make_custom_mcp_http_factory( custom_factory: MCPHttpxClientFactory, - signing_hook: Callable[[Any], Awaitable[None]] | None = None, + request_hooks: Sequence[Callable[[Any], Awaitable[None]]] = (), ) -> MCPHttpxClientFactory: """Wrap an adopter factory with mandatory MCP transport invariants.""" @@ -209,15 +194,16 @@ def factory( "MCP SDK v2 uses httpx2, so a plain httpx.AsyncClient is not compatible" ) - if signing_hook is not None: + if request_hooks: event_hooks = getattr(client, "event_hooks", None) if not isinstance(event_hooks, dict): raise TypeError( "httpx_client_factory must return a client exposing an event_hooks dict" ) - request_hooks = event_hooks.setdefault("request", []) - if signing_hook not in request_hooks: - request_hooks.append(signing_hook) + installed_hooks = event_hooks.setdefault("request", []) + for hook in request_hooks: + if hook not in installed_hooks: + installed_hooks.append(hook) return client return factory @@ -434,14 +420,17 @@ def _urls_to_try(self) -> list[str]: def _streamable_http_client_factory(self) -> MCPHttpxClientFactory: """Return the HTTP client factory used for MCP HTTP requests.""" + request_hooks = tuple( + hook + for hook in (self.tracing_request_hook, self.signing_request_hook) + if hook is not None + ) if self._httpx_client_factory is not None: return _make_custom_mcp_http_factory( self._httpx_client_factory, - self.signing_request_hook, + request_hooks, ) - if self.signing_request_hook is not None: - return _make_signing_http_factory(self.signing_request_hook) - return _make_hardened_mcp_http_factory() + return _make_hardened_mcp_http_factory(request_hooks) def current_mcp_session_id(self) -> str | None: """Return the current SDK-managed MCP Streamable HTTP session id.""" @@ -599,11 +588,19 @@ async def _get_session(self) -> ClientSession: ) ) else: + sse_request_hooks = ( + (self.tracing_request_hook,) + if self.tracing_request_hook is not None + else () + ) # Use SSE transport (legacy, but widely supported) sse_http_factory = ( - _make_custom_mcp_http_factory(self._httpx_client_factory) + _make_custom_mcp_http_factory( + self._httpx_client_factory, + sse_request_hooks, + ) if self._httpx_client_factory is not None - else _make_hardened_mcp_http_factory() + else _make_hardened_mcp_http_factory(sse_request_hooks) ) read, write = await self._exit_stack.enter_async_context( sse_client( @@ -784,7 +781,15 @@ async def _call_mcp_tool(self, tool_name: str, params: dict[str, Any]) -> TaskRe mutating=_idempotency.is_mutating(tool_name), idempotency_key=idempotency_key, ) - result = await session.call_tool(tool_name, params) + trace_meta = mcp_trace_meta() + if trace_meta is None: + result = await session.call_tool(tool_name, params) + else: + result = await session.call_tool( + tool_name, + params, + meta=cast("RequestParamsMeta", trace_meta), + ) finally: _signing_operation.reset(signing_token) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 0b0c83102..2227a070e 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -568,8 +568,11 @@ "get_properties_by_agent", "get_repeatable_groups", "get_required_assets", + "get_tracer", "has_assets", "identifiers_match", + "inject_trace_headers", + "is_tracing_available", "normalize_assets_required", "resolve_properties_for_agent", "sign_legacy_webhook", diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 000000000..cacc5d8d8 --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,469 @@ +"""Optional OpenTelemetry span and propagation contracts.""" + +from __future__ import annotations + +import re +import threading +from concurrent.futures import ThreadPoolExecutor +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from pydantic import TypeAdapter + +import adcp +from adcp import ADCPClient +from adcp import observability as obs +from adcp.protocols.a2a import A2AAdapter +from adcp.protocols.mcp import MCPAdapter +from adcp.signing.autosign import current_operation +from adcp.types import GetAdcpCapabilitiesResponse, GetProductsRequest, SyncCatalogsRequest +from adcp.types.core import AgentConfig, Protocol, TaskResult, TaskStatus + + +def _config(protocol: Protocol = Protocol.MCP) -> AgentConfig: + path = "/mcp" if protocol is Protocol.MCP else "/agent" + return AgentConfig( + id="seller-safe-id", + agent_uri=f"https://seller.example{path}", + protocol=protocol, + ) + + +def _request() -> GetProductsRequest: + return TypeAdapter(GetProductsRequest).validate_python( + {"buying_mode": "brief", "brief": "observability test"} + ) + + +@pytest.fixture +def span_exporter(monkeypatch: pytest.MonkeyPatch) -> InMemorySpanExporter: + """Route this module's tracer through an isolated in-memory provider.""" + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + bindings = obs._load_bindings() + assert bindings is not None + monkeypatch.setattr(bindings.trace, "get_tracer", provider.get_tracer) + return exporter + + +@pytest.mark.asyncio +async def test_client_task_creates_one_bounded_span( + span_exporter: InMemorySpanExporter, +) -> None: + client = ADCPClient(_config()) + + async def completed(_params: dict[str, Any]) -> TaskResult[Any]: + return TaskResult( + status=TaskStatus.COMPLETED, + success=True, + data={"products": []}, + ) + + with patch.object(client.adapter, "get_products", new=completed): + result = await client.get_products(_request()) + + assert result.success + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "adcp.mcp.call_tool" + assert dict(span.attributes or {}) == { + "rpc.system.name": "adcp", + "rpc.method": "get_products", + "adcp.tool": "get_products", + "adcp.tool.name": "get_products", + "adcp.agent.id": "seller-safe-id", + "adcp.protocol": "mcp", + "adcp.task.success": True, + "adcp.task.status": "completed", + } + assert span.events == () + + +@pytest.mark.asyncio +async def test_failed_result_records_no_remote_prose( + span_exporter: InMemorySpanExporter, +) -> None: + client = ADCPClient(_config()) + secret = "remote-secret-shaped-error" + + async def failed(_params: dict[str, Any]) -> TaskResult[Any]: + return TaskResult( + status=TaskStatus.FAILED, + success=False, + error=secret, + ) + + with patch.object(client.adapter, "get_products", new=failed): + result = await client.get_products(_request()) + + assert not result.success + span = span_exporter.get_finished_spans()[0] + assert span.status.status_code is trace.StatusCode.ERROR + assert span.attributes is not None + assert span.attributes["error.type"] == "adcp.task.failed" + rendered = repr(span) + assert secret not in rendered + assert span.events == () + + +@pytest.mark.asyncio +async def test_nested_execute_task_does_not_duplicate_spans( + span_exporter: InMemorySpanExporter, +) -> None: + client = ADCPClient(_config()) + + async def completed(_params: dict[str, Any]) -> TaskResult[Any]: + return TaskResult( + status=TaskStatus.COMPLETED, + success=True, + data={"products": []}, + ) + + with patch.object(client.adapter, "get_products", new=completed): + result = await client.execute_task("get_products", _request()) + + assert result.success + spans = span_exporter.get_finished_spans() + assert [span.name for span in spans] == ["adcp.mcp.call_tool"] + assert spans[0].attributes is not None + assert spans[0].attributes["adcp.tool.name"] == "get_products" + + +@pytest.mark.asyncio +async def test_cold_strict_idempotency_preflight_gets_distinct_wire_span( + span_exporter: InMemorySpanExporter, +) -> None: + client = ADCPClient(_config(), strict_idempotency=True) + capabilities = GetAdcpCapabilitiesResponse.model_validate( + { + "adcp": { + "major_versions": [3], + "idempotency": {"supported": True, "replay_ttl_seconds": 86400}, + }, + "supported_protocols": ["media_buy"], + } + ) + + async def get_capabilities(_params: dict[str, Any]) -> TaskResult[Any]: + return TaskResult( + status=TaskStatus.COMPLETED, + success=True, + data=capabilities.model_dump(mode="json", exclude_none=True), + ) + + async def sync_catalogs(_params: dict[str, Any]) -> TaskResult[Any]: + assert client.adapter.idempotency_capability_check is not None + await client.adapter.idempotency_capability_check() + return TaskResult(status=TaskStatus.FAILED, success=False, error="expected test result") + + with ( + patch.object(client.adapter, "get_adcp_capabilities", new=get_capabilities), + patch.object(client.adapter, "sync_catalogs", new=sync_catalogs), + ): + result = await client.sync_catalogs(SyncCatalogsRequest.model_construct()) + + assert not result.success + capability_span, target_span = span_exporter.get_finished_spans() + assert capability_span.attributes is not None + assert capability_span.attributes["rpc.method"] == "get_adcp_capabilities" + assert target_span.attributes is not None + assert target_span.attributes["rpc.method"] == "sync_catalogs" + assert capability_span.parent is not None + assert capability_span.parent.span_id == target_span.context.span_id + + +@pytest.mark.asyncio +async def test_invalid_generic_name_and_agent_id_are_not_exported( + span_exporter: InMemorySpanExporter, +) -> None: + secret_task = "secret_" + "x" * 500 + secret_agent = "agent secret " + "y" * 500 + client = ADCPClient( + AgentConfig( + id=secret_agent, + agent_uri="https://seller.example/mcp", + protocol=Protocol.MCP, + ) + ) + + with pytest.raises(ValueError, match="Unknown canonical"): + await client.execute_task(secret_task, _request()) + + span = span_exporter.get_finished_spans()[0] + assert span.attributes is not None + assert span.attributes["rpc.method"] == "unknown" + assert span.attributes["adcp.tool"] == "unknown" + assert span.attributes["adcp.agent.id"] == "unknown" + assert secret_task not in repr(span) + assert secret_agent not in repr(span) + + +def test_workflow_span_uses_internal_schema_and_child_wire_spans( + span_exporter: InMemorySpanExporter, +) -> None: + token = object() + with obs.client_task_span( + token, + protocol="mcp", + task_name="wait_for_refinement_verified", + agent_id="seller-safe-id", + workflow=True, + ): + with obs.client_task_span( + token, + protocol="mcp", + task_name="get_task_status", + agent_id="seller-safe-id", + ): + pass + + child, workflow = span_exporter.get_finished_spans() + assert child.name == "adcp.mcp.call_tool" + assert child.attributes is not None + assert child.attributes["rpc.method"] == "get_task_status" + assert workflow.name == "adcp.client.workflow" + assert workflow.kind is trace.SpanKind.INTERNAL + assert workflow.attributes is not None + assert workflow.attributes["adcp.workflow.name"] == "wait_for_refinement_verified" + assert "adcp.tool" not in workflow.attributes + + +def test_trace_headers_are_w3c_only_and_replace_stale_case( + span_exporter: InMemorySpanExporter, +) -> None: + del span_exporter + tracer = obs.get_tracer() + assert tracer is not None + + with tracer.start_as_current_span("parent"): + headers = adcp.inject_trace_headers( + { + "Traceparent": "00-00000000000000000000000000000000-0000000000000000-00", + "TraceState": "secret_vendor=stale", + "baggage": "tenant=must-not-be-generated-by-adcp", + } + ) + + assert "Traceparent" not in headers + assert re.fullmatch(r"00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}", headers["traceparent"]) + assert not any(name.lower() == "tracestate" for name in headers) + assert "baggage" not in headers + + +@pytest.mark.asyncio +async def test_transport_hook_skips_discovery_and_injects_tool_requests( + span_exporter: InMemorySpanExporter, +) -> None: + del span_exporter + client = ADCPClient(_config()) + tracer = obs.get_tracer() + assert tracer is not None + + discovery = httpx.Request("GET", "https://seller.example/.well-known/agent-card.json") + with tracer.start_as_current_span("parent"): + trace_meta = obs.mcp_trace_meta() + await client._inject_outgoing_trace_context(discovery) + assert trace_meta is not None + + # The hook runs later in MCP's long-lived writer task, outside the caller's + # ContextVar snapshot. The bounded `_meta` bridge retains the exact parent. + tool = httpx.Request( + "POST", + "https://seller.example/mcp", + headers={"TraceState": "secret_vendor=stale"}, + json={ + "jsonrpc": "2.0", + "method": "tools/call", + "params": {"name": "get_products", "_meta": trace_meta}, + }, + ) + await client._inject_outgoing_trace_context(tool) + + assert "traceparent" not in discovery.headers + assert "traceparent" in tool.headers + assert tool.headers["traceparent"] == trace_meta[obs.MCP_TRACEPARENT_META_KEY] + assert "tracestate" not in tool.headers + assert "baggage" not in tool.headers + + +@pytest.mark.asyncio +async def test_a2a_operation_scope_enables_trace_injection( + span_exporter: InMemorySpanExporter, +) -> None: + del span_exporter + client = ADCPClient(_config(Protocol.A2A)) + tracer = obs.get_tracer() + assert tracer is not None + request = httpx.Request("POST", "https://seller.example/agent") + + token = current_operation.set("get_products") + try: + with tracer.start_as_current_span("parent"): + await client._inject_outgoing_trace_context(request) + finally: + current_operation.reset(token) + + assert "traceparent" in request.headers + + +@pytest.mark.asyncio +async def test_mcp_dispatch_captures_trace_for_writer_task( + span_exporter: InMemorySpanExporter, +) -> None: + del span_exporter + adapter = MCPAdapter(_config()) + captured_meta: dict[str, str] | None = None + + async def call_tool( + _name: str, + _params: dict[str, Any], + *, + meta: dict[str, str] | None = None, + ) -> Any: + nonlocal captured_meta + captured_meta = meta + result = MagicMock() + result.isError = False + result.content = [] + result.structuredContent = None + return result + + session = MagicMock() + session.call_tool = call_tool + adapter._get_session = AsyncMock(return_value=session) # type: ignore[method-assign] + tracer = obs.get_tracer() + assert tracer is not None + + with tracer.start_as_current_span("parent"): + await adapter._call_mcp_tool( + "get_products", _request().model_dump(mode="json", exclude_none=True) + ) + + assert captured_meta is not None + assert obs.MCP_TRACEPARENT_META_KEY in captured_meta + + +@pytest.mark.asyncio +async def test_protocol_clients_install_trace_hook_before_signing() -> None: + async def tracing(_request: Any) -> None: + return None + + async def signing(_request: Any) -> None: + return None + + mcp = MCPAdapter(_config()) + mcp.tracing_request_hook = tracing + mcp.signing_request_hook = signing + mcp_client = mcp._streamable_http_client_factory()() + try: + assert mcp_client.event_hooks["request"] == [tracing, signing] + finally: + await mcp_client.aclose() + + a2a = A2AAdapter(_config(Protocol.A2A)) + a2a.tracing_request_hook = tracing + a2a.signing_request_hook = signing + a2a_client = await a2a._get_httpx_client() + try: + assert a2a_client.event_hooks["request"] == [tracing, signing] + finally: + await a2a.close() + + +def test_no_op_behavior_without_optional_api(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(obs, "_bindings_checked", True) + monkeypatch.setattr(obs, "_bindings", None) + assert not obs.is_tracing_available() + assert obs.get_tracer() is None + assert obs.inject_trace_headers({"x-existing": "value"}) == {"x-existing": "value"} + + +def test_concurrent_first_load_never_observes_partial_bindings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + expected = obs._import_bindings() + assert expected is not None + entered = threading.Event() + release = threading.Event() + second_started = threading.Event() + second_returned = threading.Event() + import_count = 0 + + def blocked_import() -> obs._OpenTelemetryBindings | None: + nonlocal import_count + import_count += 1 + entered.set() + assert release.wait(2) + return expected + + def second_load() -> obs._OpenTelemetryBindings | None: + second_started.set() + result = obs._load_bindings() + second_returned.set() + return result + + monkeypatch.setattr(obs, "_bindings", None) + monkeypatch.setattr(obs, "_bindings_checked", False) + monkeypatch.setattr(obs, "_import_bindings", blocked_import) + + with ThreadPoolExecutor(max_workers=2) as executor: + first = executor.submit(obs._load_bindings) + assert entered.wait(1) + second = executor.submit(second_load) + assert second_started.wait(1) + try: + assert not second_returned.wait(0.1) + finally: + release.set() + assert first.result(timeout=1) is expected + assert second.result(timeout=1) is expected + assert import_count == 1 + + +@pytest.mark.parametrize( + "traceparent,tracestate", + [ + ("00-00000000000000000000000000000000-1111111111111111-01", "vendor=value"), + ("not-a-traceparent", "vendor=value"), + ("ff-11111111111111111111111111111111-1111111111111111-01", "vendor=value"), + ("00-11111111111111111111111111111111-1111111111111111-01", "bad\r\nheader"), + ], +) +def test_mcp_trace_bridge_rejects_invalid_or_unsafe_values( + traceparent: str, + tracestate: str, +) -> None: + headers = obs.mcp_trace_headers_from_payload( + { + "params": { + "_meta": { + obs.MCP_TRACEPARENT_META_KEY: traceparent, + obs.MCP_TRACESTATE_META_KEY: tracestate, + } + } + } + ) + if traceparent.startswith("00-111"): + assert headers == {"traceparent": traceparent} + else: + assert headers == {} + + +def test_public_observability_exports() -> None: + assert adcp.get_tracer is obs.get_tracer + assert adcp.inject_trace_headers is obs.inject_trace_headers + assert adcp.is_tracing_available is obs.is_tracing_available + assert set(obs.__all__) == { + "get_tracer", + "inject_trace_headers", + "is_tracing_available", + } diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py index 4b75aa681..dc42c23a5 100644 --- a/tests/test_security_hardening.py +++ b/tests/test_security_hardening.py @@ -307,7 +307,11 @@ def custom_factory(**kwargs): try: assert captured["trust_env"] is False assert captured["follow_redirects"] is False - assert http_client.event_hooks["request"] == [audit_hook, signing_hook] + assert http_client.event_hooks["request"] == [ + audit_hook, + client.adapter.tracing_request_hook, + signing_hook, + ] finally: await http_client.aclose() diff --git a/tests/type_checks/observability.py b/tests/type_checks/observability.py new file mode 100644 index 000000000..a27505847 --- /dev/null +++ b/tests/type_checks/observability.py @@ -0,0 +1,12 @@ +"""Static public-surface checks for optional OpenTelemetry helpers.""" + +from typing import Any + +from adcp import get_tracer, inject_trace_headers, is_tracing_available + +available: bool = is_tracing_available() +tracer: Any | None = get_tracer() +headers: dict[str, str] = inject_trace_headers({"x-request-id": "request-1"}) + +assert isinstance(available, bool) +assert headers["x-request-id"] == "request-1"