Module adcp.testing
Test helpers for AdCP client library.
Provides pre-configured test agents for examples and quick testing.
All test agents include a .simple accessor for ergonomic usage:
- Standard API (client methods): Full TaskResult with error handling
- Simple API (client.simple methods): Direct returns, raises on error
Example -----=
Standard API - explicit control
result = await test_agent.get_products(GetProductsRequest(brief='Coffee')) if result.success: print(result.data.products)
Simple API - ergonomic
products = await test_agent.simple.get_products(brief='Coffee') print(products.products)
Sub-modules
adcp.testing.decisioning-
Test helpers for the v6 DecisioningPlatform framework …
adcp.testing.harness-
SellerTestClient / SellerA2AClient — in-process AdCP harnesses for unit tests.
adcp.testing.models-
Explicit deferred-schema setup for tests that patch datetime classes.
adcp.testing.test_helpers-
Test agent helpers for easy examples and quick testing …
adcp.testing.upstream_recorder-
Reference upstream-traffic recorder for
query_upstream_trafficconformance …
Functions
def build_all_models() ‑> None-
Expand source code
def build_all_models() -> None: """Prebuild imported SDK models/adapters before freezing datetime. In ``tests/conftest.py``, after importing adopter models and before any ``freeze_time`` block:: import freezegun from adcp.testing import build_all_models freezegun.configure(extend_ignore_list=["adcp"]) build_all_models() Realizes advertised lazy exports on ``adcp``, ``adcp.types`` and every imported SDK facade (including facades imported during this walk). Then recursively builds imported SDK models, their adopter subclasses, and registered SDK TypeAdapters. Completed schemas and adapters are cached; repeated calls do not force rebuilds. This does not import arbitrary SDK submodules or discover application modules. Import additional application models/submodules first, and call again before freezing if imports change. Keep ``adcp`` on freezegun's ignore list so its retained datetime references are not patched. Call this helper while real datetime classes are active. """ _realize_lazy_exports() visited: set[type[BaseModel]] = set() def build(cls: type[BaseModel]) -> None: if cls in visited: return visited.add(cls) sdk_model = any(base.__module__.startswith("adcp.") for base in cls.__mro__) if sdk_model and not cls.__pydantic_complete__: cls.model_rebuild(force=False) for child in cls.__subclasses__(): build(child) build(BaseModel) build_registered_adapters()Prebuild imported SDK models/adapters before freezing datetime.
In
tests/conftest.py, after importing adopter models and before anyfreeze_timeblock::import freezegun from adcp.testing import build_all_models freezegun.configure(extend_ignore_list=["adcp"]) build_all_models()Realizes advertised lazy exports on
adcp,adcp.typesand every imported SDK facade (including facades imported during this walk). Then recursively builds imported SDK models, their adopter subclasses, and registered SDK TypeAdapters. Completed schemas and adapters are cached; repeated calls do not force rebuilds. This does not import arbitrary SDK submodules or discover application modules. Import additional application models/submodules first, and call again before freezing if imports change. Keepadcpon freezegun's ignore list so its retained datetime references are not patched. Call this helper while real datetime classes are active. def build_asgi_app(platform: DecisioningPlatform,
*,
transport: "Literal['mcp', 'a2a', 'both']" = 'mcp',
name: str | None = None,
advertise_all: bool = False,
auto_emit_completion_webhooks: bool = False,
allowed_hosts: Sequence[str] | None = None,
allowed_origins: Sequence[str] | None = None,
unauthenticated_routes: Sequence[Route | Mount] | None = None,
auth: BearerTokenAuth | None = None,
asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None,
context_factory: ContextFactory | None = None,
middleware: Sequence[SkillMiddleware] | None = None,
streaming_responses: bool = False,
stateless_http: bool = False,
session_idle_timeout: float | None = 1800.0,
max_active_sessions: int | None = None,
enable_dns_rebinding_protection: bool | None = None,
max_request_size: int | None = None,
validation: ValidationHookConfig | None = ValidationHookConfig(requests='strict', responses='strict', unknown_fields=None),
discovery_base_url: str | None = None,
pre_validation_hooks: PreValidationHooks | None = None,
response_enhancer: ResponseEnhancer | None = None,
mcp_result_text: MCPResultText | None = None,
on_startup: Sequence[LifespanHook] | None = None,
on_shutdown: Sequence[LifespanHook] | None = None,
**factory_kwargs: Any) ‑> Any-
Expand source code
def build_asgi_app( platform: DecisioningPlatform, *, transport: Literal["mcp", "a2a", "both"] = "mcp", name: str | None = None, advertise_all: bool = False, auto_emit_completion_webhooks: bool = False, allowed_hosts: Sequence[str] | None = None, allowed_origins: Sequence[str] | None = None, unauthenticated_routes: Sequence[Route | Mount] | None = None, auth: BearerTokenAuth | None = None, asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None, context_factory: ContextFactory | None = None, middleware: Sequence[SkillMiddleware] | None = None, streaming_responses: bool = False, stateless_http: bool = False, session_idle_timeout: float | None = 1800.0, max_active_sessions: int | None = None, enable_dns_rebinding_protection: bool | None = None, max_request_size: int | None = None, validation: ValidationHookConfig | None = DEFAULT_VALIDATION, discovery_base_url: str | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, mcp_result_text: MCPResultText | None = None, on_startup: Sequence[LifespanHook] | None = None, on_shutdown: Sequence[LifespanHook] | None = None, **factory_kwargs: Any, ) -> Any: """Build a Starlette ASGI app for in-process integration tests. Returns the same middleware stack that :func:`adcp.decisioning.serve` mounts, minus the network bind. Pass any kwargs you would pass to :func:`adcp.decisioning.serve` (except the uvicorn-binding ones: ``port``, ``host``) and the result is drop-in compatible with ``httpx.ASGITransport``, ``starlette.testclient.TestClient``, or any ASGI test harness. ``transport="mcp"`` maps to production's ``transport="streamable-http"`` spelling. ``"a2a"`` and ``"both"`` use the same app builders as their production counterparts. The wrapping order mirrors production (``_run_mcp_http`` in ``adcp.server.serve``): 1. ``auth`` innermost — body-peeks the JSON-RPC payload before the path normalizer reshapes it (discovery bypass). 2. Path normalizer — strips trailing slashes so ``/mcp/`` → ``/mcp`` without a 307 round-trip. 3. Discovery wrapper (only when ``discovery_base_url`` is provided) — serves ``/.well-known/adcp-agents.json``. 4. Size cap (``max_request_size``; ``None`` → 10 MB default). 5. Operational routes — ahead of SDK protocol authentication. 6. Host/Origin transport policy — wraps all HTTP routes. 7. ``asgi_middleware`` outermost — CORS, tenant resolution, custom auth, etc. :param platform: The :class:`DecisioningPlatform` instance under test. :param transport: In-process transport topology: ``"mcp"`` (the production ``"streamable-http"`` transport), ``"a2a"``, or ``"both"``. Defaults to ``"mcp"`` for backward compatibility. :param name: Server name on the AdCP capabilities envelope. Defaults to ``type(platform).__name__``. :param advertise_all: Forwarded to :func:`create_adcp_server_from_platform` and :func:`create_mcp_server`. Default ``False`` (override-detection filter on; matches :func:`serve`). :param auto_emit_completion_webhooks: Deprecated compatibility argument forwarded to :func:`create_adcp_server_from_platform`. Passing ``True`` warns and does not synthesize a webhook for inline terminal results. :param allowed_hosts: Host header values the MCP transport-security layer will accept. ``None`` → FastMCP's loopback-only default (``localhost``, ``127.0.0.1``, ``[::1]``). Pass the hostname embedded in your ``base_url`` when using a non-loopback test address (e.g. ``["test"]`` for ``base_url="http://test"``). :func:`build_test_client` sets this automatically. :param allowed_origins: CORS origin allowlist forwarded to :func:`create_mcp_server`. ``None`` → FastMCP default (no CORS). :param auth: Optional :class:`~adcp.server.auth.BearerTokenAuth` config applied to the MCP ASGI app. Drives :class:`~adcp.server.auth.BearerTokenAuthMiddleware`. ``None`` → no bearer-token validation (unauthenticated). :param asgi_middleware: Optional ASGI middleware entries applied outermost — same semantics as :func:`serve`'s ``asgi_middleware`` param. Use for CORS, request-id propagation, custom auth. :param context_factory: Optional factory that builds a :class:`~adcp.server.ToolContext` per tool call. Forwarded to :func:`create_mcp_server`. ``None`` → bare ``ToolContext()``. :param middleware: Optional sequence of :data:`~adcp.server.serve.SkillMiddleware` callables wrapping every tool dispatch. Forwarded to :func:`create_mcp_server`. :param streaming_responses: Forwarded to :func:`create_mcp_server`. Default ``False``. :param stateless_http: Forwarded to :func:`create_mcp_server` for ``transport="mcp"`` and to the production composition path for ``transport="both"``. Ignored by ``transport="a2a"``. :param session_idle_timeout: Idle reap deadline for stateful MCP sessions. Defaults to 1800 seconds. Forwarded for ``"mcp"`` and ``"both"``; ignored by ``"a2a"``. :param max_active_sessions: Optional cap for active stateful MCP sessions. Forwarded for ``"mcp"`` and ``"both"``; ignored by ``"a2a"``. :param enable_dns_rebinding_protection: Forwarded to :func:`create_mcp_server`. ``None`` → FastMCP default. :param max_request_size: Request body size cap in bytes. ``None`` → the framework default (10 MB). ``0`` → disabled. :param validation: Schema validation config forwarded to :func:`create_mcp_server`. Defaults to :data:`~adcp.server.serve.DEFAULT_VALIDATION` (strict on both requests and responses) — matches production. Pass ``validation=None`` to disable. :param discovery_base_url: When provided, mounts the ``/.well-known/adcp-agents.json`` discovery endpoint using this as the advertised base URL (e.g. ``"http://test"``). ``None`` → discovery endpoint not mounted. :param pre_validation_hooks: Optional dict mapping AdCP tool name to a ``(tool_name, raw_args) -> raw_args`` callable or ordered sequence. Forwarded to :func:`create_mcp_server`, identical semantics to :func:`adcp.decisioning.serve`'s ``pre_validation_hooks`` param. Use to install the same coercion hooks your production :func:`serve` call uses so in-process tests see the same validation surface as production. ``None`` → no hooks (default). :param response_enhancer: Optional server-wide :data:`~adcp.server.ResponseEnhancer` forwarded to :func:`create_mcp_server`, so in-process tests exercise the same enhancer wiring your production :func:`serve` call uses. ``None`` → no enhancer (default). :param on_startup: Async zero-argument hooks run after the MCP and A2A framework lifespans start. Requires ``transport="both"``. :param on_shutdown: Async zero-argument hooks run before the MCP and A2A framework lifespans stop. Requires ``transport="both"``. :param factory_kwargs: Forwarded to :func:`create_adcp_server_from_platform`. Accepted keys: ``executor``, ``registry``, ``webhook_sender``, ``webhook_supervisor``, ``buyer_agent_registry``, ``config_store``, ``property_list_fetcher``, ``state_reader``, ``resource_resolver``. :returns: A Starlette ASGI application. Usable with ``starlette.testclient.TestClient``, ``httpx.AsyncClient(app=app, ...)``, or any ASGI test harness. """ if transport not in ("mcp", "a2a", "both"): raise ValueError(f"Unsupported transport {transport!r}; expected 'mcp', 'a2a', or 'both'.") if (on_startup or on_shutdown) and transport != "both": raise ValueError( "on_startup / on_shutdown hooks require transport='both', " f"got transport={transport!r}." ) if transport == "a2a": ignored_session_settings = [] if stateless_http: ignored_session_settings.append("stateless_http") if session_idle_timeout != 1800.0: ignored_session_settings.append("session_idle_timeout") if max_active_sessions is not None: ignored_session_settings.append("max_active_sessions") if ignored_session_settings: warnings.warn( "build_asgi_app sets MCP-only session fields " f"{sorted(ignored_session_settings)} but transport='a2a'. " "These fields will be ignored.", UserWarning, stacklevel=2, ) from adcp.decisioning.serve import create_adcp_server_from_platform from adcp.server.serve import ( _apply_asgi_middleware, _build_a2a_app, _build_mcp_and_a2a_app, _wrap_mcp_with_auth, _wrap_with_path_normalize, _wrap_with_size_limit, create_mcp_server, ) handler, _executor, _registry = create_adcp_server_from_platform( platform, advertise_all=advertise_all, auto_emit_completion_webhooks=auto_emit_completion_webhooks, **factory_kwargs, ) server_name = name or type(platform).__name__ if transport == "both": # TODO(#1047): expose A2A-only message_parser/public_url and task / # push stores explicitly once the public testing signature owns them. app = _build_mcp_and_a2a_app( handler, name=server_name, port=0, host="127.0.0.1", instructions=None, test_controller=None, context_factory=context_factory, middleware=middleware, advertise_all=advertise_all, max_request_size=max_request_size, streaming_responses=streaming_responses, stateless_http=stateless_http, session_idle_timeout=session_idle_timeout, max_active_sessions=max_active_sessions, validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, mcp_result_text=mcp_result_text, base_url=discovery_base_url, allowed_hosts=allowed_hosts, allowed_origins=allowed_origins, enable_dns_rebinding_protection=enable_dns_rebinding_protection, unauthenticated_routes=unauthenticated_routes, auth=auth, on_startup=on_startup, on_shutdown=on_shutdown, include_discovery=discovery_base_url is not None, ) return _apply_asgi_middleware(app, asgi_middleware) if transport == "a2a": # TODO(#1047): expose A2A-only message_parser/public_url and task / # push stores explicitly once the public testing signature owns them. return _build_a2a_app( handler, name=server_name, port=0, test_controller=None, context_factory=context_factory, middleware=middleware, asgi_middleware=asgi_middleware, advertise_all=advertise_all, max_request_size=max_request_size, validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, base_url=discovery_base_url, allowed_hosts=allowed_hosts, allowed_origins=allowed_origins, enable_dns_rebinding_protection=enable_dns_rebinding_protection, unauthenticated_routes=unauthenticated_routes, auth=auth, include_discovery=discovery_base_url is not None, ) mcp = create_mcp_server( handler, name=server_name, advertise_all=advertise_all, allowed_hosts=allowed_hosts, allowed_origins=allowed_origins, context_factory=context_factory, middleware=middleware, streaming_responses=streaming_responses, stateless_http=stateless_http, session_idle_timeout=session_idle_timeout, max_active_sessions=max_active_sessions, enable_dns_rebinding_protection=enable_dns_rebinding_protection, validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, mcp_result_text=mcp_result_text, ) # Mirror the wrapping chain from _run_mcp_http (adcp.server.serve). # auth must be innermost so its JSON-RPC body-peek runs before the # path normalizer reshapes scope["path"]. app = mcp.streamable_http_app() app = _wrap_mcp_with_auth(app, auth) app = _wrap_with_path_normalize(app) if discovery_base_url is not None: from adcp.server.serve import _wrap_with_discovery app = _wrap_with_discovery( app, name=server_name, transports=["mcp"], base_url=discovery_base_url, ) from adcp.server.http_policy import HostOriginMiddleware from adcp.server.operational_routes import wrap_operational_routes app = wrap_operational_routes(app, unauthenticated_routes) app = _wrap_with_size_limit(app, max_request_size) app = HostOriginMiddleware(app, settings=mcp.settings.transport_security) app = _apply_asgi_middleware(app, asgi_middleware) return appBuild a Starlette ASGI app for in-process integration tests.
Returns the same middleware stack that :func:
serve()mounts, minus the network bind. Pass any kwargs you would pass to :func:serve()(except the uvicorn-binding ones:port,host) and the result is drop-in compatible withhttpx.ASGITransport,starlette.testclient.TestClient, or any ASGI test harness.transport="mcp"maps to production'stransport="streamable-http"spelling."a2a"and"both"use the same app builders as their production counterparts.The wrapping order mirrors production (
_run_mcp_httpinserve()):authinnermost — body-peeks the JSON-RPC payload before the path normalizer reshapes it (discovery bypass).- Path normalizer — strips trailing slashes so
/mcp/→/mcpwithout a 307 round-trip. - Discovery wrapper (only when
discovery_base_urlis provided) — serves/.well-known/adcp-agents.json. - Size cap (
max_request_size;None→ 10 MB default). - Operational routes — ahead of SDK protocol authentication.
- Host/Origin transport policy — wraps all HTTP routes.
asgi_middlewareoutermost — CORS, tenant resolution, custom auth, etc.
:param platform: The :class:
DecisioningPlatforminstance under test. :param transport: In-process transport topology:"mcp"(the production"streamable-http"transport),"a2a", or"both". Defaults to"mcp"for backward compatibility. :param name: Server name on the AdCP capabilities envelope. Defaults totype(platform).__name__. :param advertise_all: Forwarded to :func:create_adcp_server_from_platformand :func:create_mcp_server. DefaultFalse(override-detection filter on; matches :func:serve). :param auto_emit_completion_webhooks: Deprecated compatibility argument forwarded to :func:create_adcp_server_from_platform. PassingTruewarns and does not synthesize a webhook for inline terminal results. :param allowed_hosts: Host header values the MCP transport-security layer will accept.None→ FastMCP's loopback-only default (localhost,127.0.0.1,[::1]). Pass the hostname embedded in yourbase_urlwhen using a non-loopback test address (e.g.["test"]forbase_url="http://test"). :func:build_test_client()sets this automatically. :param allowed_origins: CORS origin allowlist forwarded to :func:create_mcp_server.None→ FastMCP default (no CORS). :param auth: Optional :class:~adcp.server.auth.BearerTokenAuthconfig applied to the MCP ASGI app. Drives :class:~adcp.server.auth.BearerTokenAuthMiddleware.None→ no bearer-token validation (unauthenticated). :param asgi_middleware: Optional ASGI middleware entries applied outermost — same semantics as :func:serve'sasgi_middlewareparam. Use for CORS, request-id propagation, custom auth. :param context_factory: Optional factory that builds a :class:~adcp.server.ToolContextper tool call. Forwarded to :func:create_mcp_server.None→ bareToolContext(). :param middleware: Optional sequence of :data:~adcp.server.serve.SkillMiddlewarecallables wrapping every tool dispatch. Forwarded to :func:create_mcp_server. :param streaming_responses: Forwarded to :func:create_mcp_server. DefaultFalse. :param stateless_http: Forwarded to :func:create_mcp_serverfortransport="mcp"and to the production composition path fortransport="both". Ignored bytransport="a2a". :param session_idle_timeout: Idle reap deadline for stateful MCP sessions. Defaults to 1800 seconds. Forwarded for"mcp"and"both"; ignored by"a2a". :param max_active_sessions: Optional cap for active stateful MCP sessions. Forwarded for"mcp"and"both"; ignored by"a2a". :param enable_dns_rebinding_protection: Forwarded to :func:create_mcp_server.None→ FastMCP default. :param max_request_size: Request body size cap in bytes.None→ the framework default (10 MB).0→ disabled. :param validation: Schema validation config forwarded to :func:create_mcp_server. Defaults to :data:~adcp.server.serve.DEFAULT_VALIDATION(strict on both requests and responses) — matches production. Passvalidation=Noneto disable. :param discovery_base_url: When provided, mounts the/.well-known/adcp-agents.jsondiscovery endpoint using this as the advertised base URL (e.g."http://test").None→ discovery endpoint not mounted. :param pre_validation_hooks: Optional dict mapping AdCP tool name to a(tool_name, raw_args) -> raw_argscallable or ordered sequence. Forwarded to :func:create_mcp_server, identical semantics to :func:serve()'spre_validation_hooksparam. Use to install the same coercion hooks your production :func:servecall uses so in-process tests see the same validation surface as production.None→ no hooks (default). :param response_enhancer: Optional server-wide :data:~adcp.server.ResponseEnhancerforwarded to :func:create_mcp_server, so in-process tests exercise the same enhancer wiring your production :func:servecall uses.None→ no enhancer (default). :param on_startup: Async zero-argument hooks run after the MCP and A2A framework lifespans start. Requirestransport="both". :param on_shutdown: Async zero-argument hooks run before the MCP and A2A framework lifespans stop. Requirestransport="both". :param factory_kwargs: Forwarded to :func:create_adcp_server_from_platform. Accepted keys:executor,registry,webhook_sender,webhook_supervisor,buyer_agent_registry,config_store,property_list_fetcher,state_reader,resource_resolver.:returns: A Starlette ASGI application. Usable with
starlette.testclient.TestClient,httpx.AsyncClient(app=app, ...), or any ASGI test harness. async def build_test_client(platform: DecisioningPlatform,
*,
base_url: str = 'http://test',
transport: "Literal['mcp', 'a2a', 'both']" = 'mcp',
name: str | None = None,
advertise_all: bool = False,
auto_emit_completion_webhooks: bool = False,
follow_redirects: bool = True,
headers: Mapping[str, str] | None = None,
unauthenticated_routes: Sequence[Route | Mount] | None = None,
auth: BearerTokenAuth | None = None,
allowed_origins: Sequence[str] | None = None,
asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None,
context_factory: ContextFactory | None = None,
middleware: Sequence[SkillMiddleware] | None = None,
streaming_responses: bool = False,
stateless_http: bool = False,
session_idle_timeout: float | None = 1800.0,
max_active_sessions: int | None = None,
enable_dns_rebinding_protection: bool | None = None,
max_request_size: int | None = None,
validation: ValidationHookConfig | None = ValidationHookConfig(requests='strict', responses='strict', unknown_fields=None),
discovery_base_url: str | None = None,
pre_validation_hooks: PreValidationHooks | None = None,
response_enhancer: ResponseEnhancer | None = None,
mcp_result_text: MCPResultText | None = None,
on_startup: Sequence[LifespanHook] | None = None,
on_shutdown: Sequence[LifespanHook] | None = None,
**factory_kwargs: Any) ‑> AsyncIterator[httpx.AsyncClient]-
Expand source code
@asynccontextmanager async def build_test_client( platform: DecisioningPlatform, *, base_url: str = "http://test", transport: Literal["mcp", "a2a", "both"] = "mcp", name: str | None = None, advertise_all: bool = False, auto_emit_completion_webhooks: bool = False, follow_redirects: bool = True, headers: Mapping[str, str] | None = None, unauthenticated_routes: Sequence[Route | Mount] | None = None, auth: BearerTokenAuth | None = None, allowed_origins: Sequence[str] | None = None, asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None, context_factory: ContextFactory | None = None, middleware: Sequence[SkillMiddleware] | None = None, streaming_responses: bool = False, stateless_http: bool = False, session_idle_timeout: float | None = 1800.0, max_active_sessions: int | None = None, enable_dns_rebinding_protection: bool | None = None, max_request_size: int | None = None, validation: ValidationHookConfig | None = DEFAULT_VALIDATION, discovery_base_url: str | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, mcp_result_text: MCPResultText | None = None, on_startup: Sequence[LifespanHook] | None = None, on_shutdown: Sequence[LifespanHook] | None = None, **factory_kwargs: Any, ) -> AsyncIterator[httpx.AsyncClient]: """Async context manager yielding an ``httpx.AsyncClient`` wired against the platform's ASGI app via ``httpx.ASGITransport`` + ``LifespanManager``. Collapses the four-line boilerplate that every in-process integration test previously needed — ``build_asgi_app`` + ``LifespanManager`` + ``httpx.AsyncClient`` — into a single ``async with`` block:: async with build_test_client(platform) as client: resp = await client.post("/mcp/", json=...) The context manager starts the ASGI lifespan on entry and shuts down both the client and the lifespan manager on exit. ``build_test_client(...)`` itself is an ``AbstractAsyncContextManager[httpx.AsyncClient]``; the yielded object is a plain ``httpx.AsyncClient``. ``allowed_hosts`` is derived automatically from ``base_url`` — the hostname is extracted and added to FastMCP's transport-security allowlist. Pass ``allowed_hosts`` to :func:`build_asgi_app` directly when you need custom control. Requires ``asgi-lifespan`` (included in ``adcp[dev]``). Raises :class:`ImportError` with an actionable message if it is not installed. :param platform: The :class:`DecisioningPlatform` instance under test. :param base_url: Base URL for all requests. Default ``"http://test"``. The hostname is extracted and added to the transport-security ``allowed_hosts`` list automatically — no manual wiring needed. :param transport: Forwarded to :func:`build_asgi_app`. :param name: Server name forwarded to :func:`build_asgi_app`. :param advertise_all: Forwarded to :func:`build_asgi_app`. :param auto_emit_completion_webhooks: Forwarded to :func:`build_asgi_app`. :param follow_redirects: Forwarded to ``httpx.AsyncClient``. Default ``True`` — FastMCP's streamable-HTTP endpoint can issue a 307 redirect (``/mcp`` → ``/mcp/``) and callers shouldn't have to handle it manually. :param headers: Default headers attached to every request. Useful for auth tests: ``headers={"x-adcp-auth": "tok_..."}``. ``None`` → no default headers. :param auth: Forwarded to :func:`build_asgi_app`. ``None`` → no bearer-token validation. :param allowed_origins: CORS origin allowlist forwarded to :func:`build_asgi_app`. ``None`` → FastMCP default (no CORS). :param asgi_middleware: Forwarded to :func:`build_asgi_app`. :param context_factory: Forwarded to :func:`build_asgi_app`. :param middleware: Forwarded to :func:`build_asgi_app`. :param streaming_responses: Forwarded to :func:`build_asgi_app`. :param stateless_http: Forwarded to :func:`build_asgi_app`. :param session_idle_timeout: Forwarded to :func:`build_asgi_app`. :param max_active_sessions: Forwarded to :func:`build_asgi_app`. :param enable_dns_rebinding_protection: Forwarded to :func:`build_asgi_app`. :param max_request_size: Forwarded to :func:`build_asgi_app`. :param validation: Forwarded to :func:`build_asgi_app`. Defaults to :data:`~adcp.server.serve.DEFAULT_VALIDATION` (strict). :param discovery_base_url: Forwarded to :func:`build_asgi_app`. When ``None`` (default), the discovery endpoint is not mounted. Pass ``base_url`` here if your tests exercise ``/.well-known/adcp-agents.json``. :param pre_validation_hooks: Forwarded to :func:`build_asgi_app`. Install the same hooks your production :func:`serve` call uses so in-process tests see the same validation surface. :param response_enhancer: Forwarded to :func:`build_asgi_app`. Wire the same enhancer your production :func:`serve` call uses so in-process tests exercise the enhancer path. :param on_startup: Forwarded to :func:`build_asgi_app`. Requires ``transport="both"``. :param on_shutdown: Forwarded to :func:`build_asgi_app`. Requires ``transport="both"``. :param factory_kwargs: Forwarded to :func:`create_adcp_server_from_platform` via :func:`build_asgi_app` (executor, registry, webhook_sender, etc.). """ try: from asgi_lifespan import LifespanManager except ImportError as exc: raise ImportError( "asgi-lifespan is required for build_test_client. " "Install it with: pip install 'adcp[dev]'" ) from exc import httpx as _httpx hostname = urlparse(base_url).hostname or "localhost" # ``create_adcp_server_from_platform`` is a sync function whose # default ``validate_at_init=True`` path drives the async # capabilities handler through ``asyncio.run`` — incompatible with # the running event loop we're in here (#700). Force # ``validate_at_init=False`` regardless of what the adopter passed # — a test client is by definition inside a running loop, so a # caller asking for ``True`` would hit the exact bug this PR # fixes. Loud error is better than mysterious ``RuntimeError``. if factory_kwargs.get("validate_at_init") is True: raise ValueError( "build_test_client cannot validate capabilities at init — " "the test client runs inside the test's event loop, and " "the sync validator uses asyncio.run() which is " "incompatible with a running loop. Drop validate_at_init " "from factory_kwargs and `await " "validate_capabilities_response_shape_async(handler)` " "yourself if you need the boot-time check. See #700." ) factory_kwargs["validate_at_init"] = False # The other boot validators (``validate_platform``, webhook # signing, idempotency wiring) are sync-pure and always run # regardless. Only the capabilities-shape validator is gated. app = build_asgi_app( platform, transport=transport, name=name, advertise_all=advertise_all, auto_emit_completion_webhooks=auto_emit_completion_webhooks, allowed_hosts=[hostname], allowed_origins=allowed_origins, unauthenticated_routes=unauthenticated_routes, auth=auth, asgi_middleware=asgi_middleware, context_factory=context_factory, middleware=middleware, streaming_responses=streaming_responses, stateless_http=stateless_http, session_idle_timeout=session_idle_timeout, max_active_sessions=max_active_sessions, enable_dns_rebinding_protection=enable_dns_rebinding_protection, max_request_size=max_request_size, validation=validation, discovery_base_url=discovery_base_url, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, mcp_result_text=mcp_result_text, on_startup=on_startup, on_shutdown=on_shutdown, **factory_kwargs, ) async with LifespanManager(app): async with _httpx.AsyncClient( transport=_httpx.ASGITransport(app=app), timeout=5.0, base_url=base_url, headers=headers, follow_redirects=follow_redirects, ) as client: yield clientAsync context manager yielding an
httpx.AsyncClientwired against the platform's ASGI app viahttpx.ASGITransport+LifespanManager.Collapses the four-line boilerplate that every in-process integration test previously needed —
build_asgi_app()+LifespanManager+httpx.AsyncClient— into a singleasync withblock::async with build_test_client(platform) as client: resp = await client.post("/mcp/", json=...)The context manager starts the ASGI lifespan on entry and shuts down both the client and the lifespan manager on exit.
build_test_client()(…)itself is anAbstractAsyncContextManager[httpx.AsyncClient]; the yielded object is a plainhttpx.AsyncClient.allowed_hostsis derived automatically frombase_url— the hostname is extracted and added to FastMCP's transport-security allowlist. Passallowed_hoststo :func:build_asgi_app()directly when you need custom control.Requires
asgi-lifespan(included inadcp[dev]). Raises :class:ImportErrorwith an actionable message if it is not installed.:param platform: The :class:
DecisioningPlatforminstance under test. :param base_url: Base URL for all requests. Default"http://test". The hostname is extracted and added to the transport-securityallowed_hostslist automatically — no manual wiring needed. :param transport: Forwarded to :func:build_asgi_app(). :param name: Server name forwarded to :func:build_asgi_app(). :param advertise_all: Forwarded to :func:build_asgi_app(). :param auto_emit_completion_webhooks: Forwarded to :func:build_asgi_app(). :param follow_redirects: Forwarded tohttpx.AsyncClient. DefaultTrue— FastMCP's streamable-HTTP endpoint can issue a 307 redirect (/mcp→/mcp/) and callers shouldn't have to handle it manually. :param headers: Default headers attached to every request. Useful for auth tests:headers={"x-adcp-auth": "tok_..."}.None→ no default headers. :param auth: Forwarded to :func:build_asgi_app().None→ no bearer-token validation. :param allowed_origins: CORS origin allowlist forwarded to :func:build_asgi_app().None→ FastMCP default (no CORS). :param asgi_middleware: Forwarded to :func:build_asgi_app(). :param context_factory: Forwarded to :func:build_asgi_app(). :param middleware: Forwarded to :func:build_asgi_app(). :param streaming_responses: Forwarded to :func:build_asgi_app(). :param stateless_http: Forwarded to :func:build_asgi_app(). :param session_idle_timeout: Forwarded to :func:build_asgi_app(). :param max_active_sessions: Forwarded to :func:build_asgi_app(). :param enable_dns_rebinding_protection: Forwarded to :func:build_asgi_app(). :param max_request_size: Forwarded to :func:build_asgi_app(). :param validation: Forwarded to :func:build_asgi_app(). Defaults to :data:~adcp.server.serve.DEFAULT_VALIDATION(strict). :param discovery_base_url: Forwarded to :func:build_asgi_app(). WhenNone(default), the discovery endpoint is not mounted. Passbase_urlhere if your tests exercise/.well-known/adcp-agents.json. :param pre_validation_hooks: Forwarded to :func:build_asgi_app(). Install the same hooks your production :func:servecall uses so in-process tests see the same validation surface. :param response_enhancer: Forwarded to :func:build_asgi_app(). Wire the same enhancer your production :func:servecall uses so in-process tests exercise the enhancer path. :param on_startup: Forwarded to :func:build_asgi_app(). Requirestransport="both". :param on_shutdown: Forwarded to :func:build_asgi_app(). Requirestransport="both". :param factory_kwargs: Forwarded to :func:create_adcp_server_from_platformvia :func:build_asgi_app()(executor, registry, webhook_sender, etc.). def create_test_agent(**overrides: Any) ‑> AgentConfig-
Expand source code
def create_test_agent(**overrides: Any) -> AgentConfig: """Create a custom test agent configuration. Useful when you need to modify the default test agent setup. Args: **overrides: Keyword arguments to override default config values Returns: Complete agent configuration Example: ```python from adcp.testing import create_test_agent from adcp.client import ADCPClient # Use default test agent with custom ID config = create_test_agent(id="my-test-agent") client = ADCPClient(config) ``` Example: ```python # Use A2A protocol instead of MCP from adcp.types.core import Protocol config = create_test_agent( protocol=Protocol.A2A, agent_uri="https://test-agent.adcontextprotocol.org" ) ``` """ base_config = TEST_AGENT_MCP_CONFIG.model_dump() base_config.update(overrides) return AgentConfig(**base_config)Create a custom test agent configuration.
Useful when you need to modify the default test agent setup.
- Args
- -----=
**overrides- Keyword arguments to override default config values
Returns -----= Complete agent configuration
Example -----=
from adcp.testing import create_test_agent from adcp.client import ADCPClient # Use default test agent with custom ID config = create_test_agent(id="my-test-agent") client = ADCPClient(config)Example -----=
# Use A2A protocol instead of MCP from adcp.types.core import Protocol config = create_test_agent( protocol=Protocol.A2A, agent_uri="https://test-agent.adcontextprotocol.org" ) def make_request_context(*,
account: Account[Any] | str | None = None,
auth_info: AuthInfo | None = None,
auth_principal: str | None = None,
buyer_agent: BuyerAgent | None = None,
now: datetime | None = None,
state: StateReader | None = None,
resolve: ResourceResolver | None = None,
request_id: str | None = None,
tenant_id: str | None = None,
caller_identity: str | None = None,
metadata: dict[str, Any] | None = None) ‑> RequestContext[Any]-
Expand source code
def make_request_context( *, account: Account[Any] | str | None = None, auth_info: AuthInfo | None = None, auth_principal: str | None = None, buyer_agent: BuyerAgent | None = None, now: datetime | None = None, state: StateReader | None = None, resolve: ResourceResolver | None = None, request_id: str | None = None, tenant_id: str | None = None, caller_identity: str | None = None, metadata: dict[str, Any] | None = None, ) -> RequestContext[Any]: """Build a :class:`RequestContext` for unit tests. All parameters are optional. The defaults are stable test contract: an empty ``Account(id="test-account")``, no auth, framework-default ``state`` / ``resolve`` (the v6.0 stub readers), ``now`` set to wall clock, and an empty metadata dict. Pass ``account=`` as either an :class:`Account` instance (full control) or a string (shorthand for ``Account(id=<string>)``) — the common test case. :param account: Resolved account. ``None`` → ``Account(id="test-account")``. ``str`` → ``Account(id=<string>)``. :param auth_info: Verified principal info. ``None`` for unauthenticated / ``'derived'`` test fixtures. :param auth_principal: Convenience field for tests that read ``ctx.auth_principal`` without constructing an ``AuthInfo``. :param buyer_agent: Resolved commercial buyer agent. ``None`` for tests not exercising the registry path. :param now: Request timestamp. ``None`` → wall clock at construction. :param state: Workflow-state reader. ``None`` → framework default (v6.0 stub returning empty values). :param resolve: Async resource resolver. ``None`` → framework default (v6.0 stub raising ``NotImplementedError``). :param request_id: Inherited from :class:`adcp.server.ToolContext`. :param tenant_id: Inherited from :class:`adcp.server.ToolContext`. :param caller_identity: Inherited from :class:`adcp.server.ToolContext`. The framework's idempotency middleware reads this; tests that exercise idempotency paths should set it explicitly. :param metadata: Inherited from :class:`adcp.server.ToolContext`. ``None`` → empty dict. :returns: A populated :class:`RequestContext[Any]`. """ resolved_account: Account[Any] if account is None: resolved_account = Account(id="test-account") elif isinstance(account, str): resolved_account = Account(id=account) else: resolved_account = account kwargs: dict[str, Any] = {"account": resolved_account} if auth_info is not None: kwargs["auth_info"] = auth_info if auth_principal is not None: kwargs["auth_principal"] = auth_principal if buyer_agent is not None: kwargs["buyer_agent"] = buyer_agent if now is not None: kwargs["now"] = now if state is not None: kwargs["state"] = state if resolve is not None: kwargs["resolve"] = resolve if request_id is not None: kwargs["request_id"] = request_id if tenant_id is not None: kwargs["tenant_id"] = tenant_id if caller_identity is not None: kwargs["caller_identity"] = caller_identity if metadata is not None: kwargs["metadata"] = metadata return RequestContext(**kwargs)Build a :class:
RequestContextfor unit tests.All parameters are optional. The defaults are stable test contract: an empty
Account(id="test-account"), no auth, framework-defaultstate/resolve(the v6.0 stub readers),nowset to wall clock, and an empty metadata dict.Pass
account=as either an :class:Accountinstance (full control) or a string (shorthand forAccount(id=<string>)) — the common test case.:param account: Resolved account.
None→Account(id="test-account").str→Account(id=<string>). :param auth_info: Verified principal info.Nonefor unauthenticated /'derived'test fixtures. :param auth_principal: Convenience field for tests that readctx.auth_principalwithout constructing anAuthInfo. :param buyer_agent: Resolved commercial buyer agent.Nonefor tests not exercising the registry path. :param now: Request timestamp.None→ wall clock at construction. :param state: Workflow-state reader.None→ framework default (v6.0 stub returning empty values). :param resolve: Async resource resolver.None→ framework default (v6.0 stub raisingNotImplementedError). :param request_id: Inherited from :class:ToolContext. :param tenant_id: Inherited from :class:ToolContext. :param caller_identity: Inherited from :class:ToolContext. The framework's idempotency middleware reads this; tests that exercise idempotency paths should set it explicitly. :param metadata: Inherited from :class:ToolContext.None→ empty dict.:returns: A populated :class:
RequestContext[Any].
Classes
class AdcpErrorPayload (code: str,
message: str,
recovery: str | None = None,
field: str | None = None,
suggestion: str | None = None,
retry_after: int | None = None,
details: dict[str, Any] | None = None)-
Expand source code
@dataclass(frozen=True) class AdcpErrorPayload: """Typed container for a wire ``adcp_error`` dict. Maps to the AdCP transport-errors spec §MCP Binding fields. """ code: str message: str recovery: str | None = None field: str | None = None suggestion: str | None = None retry_after: int | None = None details: dict[str, Any] | None = NoneTyped container for a wire
adcp_errordict.Maps to the AdCP transport-errors spec §MCP Binding fields.
Instance variables
var code : strvar details : dict[str, typing.Any] | Nonevar field : str | Nonevar message : strvar recovery : str | Nonevar retry_after : int | Nonevar suggestion : str | None
class RecordedCall (method: HttpMethod,
url: str,
host: str,
path: str,
content_type: str,
payload: Any,
payload_length: int,
timestamp: datetime,
status_code: int | None = None,
purpose: Purpose | None = None,
principal: str = '')-
Expand source code
@dataclass(frozen=True) class RecordedCall: """One outbound HTTP call captured by :class:`UpstreamRecorder`. Field names mirror the ``recorded_calls[]`` item shape of the ``UpstreamTrafficSuccess`` response so :meth:`to_recorded_call_dict` is a near-identity projection. This reference recorder emits ``attestation_mode="raw"`` — the load-bearing assertion target for ``payload_must_contain`` — and leaves digest mode to adopters whose privacy policy requires it. """ method: HttpMethod url: str host: str path: str content_type: str payload: Any payload_length: int timestamp: datetime status_code: int | None = None purpose: Purpose | None = None principal: str = "" @property def endpoint(self) -> str: """Composed ``<METHOD> <URL>`` string for ``endpoint_pattern`` matching.""" return f"{self.method} {self.url}" def to_recorded_call_dict(self) -> dict[str, Any]: """Project to a ``recorded_calls[]`` item (raw attestation).""" item: dict[str, Any] = { "method": self.method, "endpoint": self.endpoint, "url": self.url, "host": self.host, "path": self.path, "content_type": self.content_type, "attestation_mode": "raw", "payload": self.payload, "payload_length": self.payload_length, "timestamp": self.timestamp.isoformat(), } if self.status_code is not None: item["status_code"] = self.status_code if self.purpose is not None: item["purpose"] = self.purpose return itemOne outbound HTTP call captured by :class:
UpstreamRecorder.Field names mirror the
recorded_calls[]item shape of theUpstreamTrafficSuccessresponse so :meth:to_recorded_call_dictis a near-identity projection. This reference recorder emitsattestation_mode="raw"— the load-bearing assertion target forpayload_must_contain— and leaves digest mode to adopters whose privacy policy requires it.Instance variables
var content_type : strprop endpoint : str-
Expand source code
@property def endpoint(self) -> str: """Composed ``<METHOD> <URL>`` string for ``endpoint_pattern`` matching.""" return f"{self.method} {self.url}"Composed
<METHOD> <URL>string forendpoint_patternmatching. var host : strvar method : Literal['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS']var path : strvar payload : Anyvar payload_length : intvar principal : strvar purpose : Literal['platform_primary', 'measurement', 'attribution', 'creative_serving', 'identity', 'other'] | Nonevar status_code : int | Nonevar timestamp : datetime.datetimevar url : str
Methods
def to_recorded_call_dict(self) ‑> dict[str, typing.Any]-
Expand source code
def to_recorded_call_dict(self) -> dict[str, Any]: """Project to a ``recorded_calls[]`` item (raw attestation).""" item: dict[str, Any] = { "method": self.method, "endpoint": self.endpoint, "url": self.url, "host": self.host, "path": self.path, "content_type": self.content_type, "attestation_mode": "raw", "payload": self.payload, "payload_length": self.payload_length, "timestamp": self.timestamp.isoformat(), } if self.status_code is not None: item["status_code"] = self.status_code if self.purpose is not None: item["purpose"] = self.purpose return itemProject to a
recorded_calls[]item (raw attestation).
class SellerA2AClient (platform: DecisioningPlatform,
*,
validation: ValidationHookConfig | None = None)-
Expand source code
class SellerA2AClient: """In-process A2A test client for AdCP seller implementations. The A2A sibling to :class:`SellerTestClient`. Same call shape (``await client.invoke(skill, payload)``), same return type (:class:`ToolInvokeResult`), but routes through the A2A executor + event-queue dispatch path instead of MCP's tool call. Usage:: @pytest.fixture def seller_a2a(): return SellerA2AClient(MySeller()) async def test_buy_not_found_a2a(seller_a2a): result = await seller_a2a.invoke( "update_media_buy", {"media_buy_id": "missing", ...} ) assert not result.ok assert result.adcp_error.code == "MEDIA_BUY_NOT_FOUND" The harness constructs an :class:`~adcp.server.a2a_server.ADCPAgentExecutor`, builds a minimal A2A :class:`~a2a.server.agent_execution.context.RequestContext` carrying a ``DataPart`` with ``{"skill": ..., "parameters": ...}``, runs the executor against a fresh event queue, and drains the queue until a terminal Task arrives. Success payloads come from the Task's first DataPart artifact; structured errors land in that same DataPart keyed under ``adcp_error`` per transport-errors.mdx §A2A Binding. Limitations (each tracked as a follow-up on #678): * **No push-notification capture.** Adopters who need to assert on outbound signed webhook delivery need a sink primitive this client doesn't yet provide. * **No intermediate state observation.** :meth:`invoke` drains to the terminal Task and returns. Tasks that pass through ``working`` / ``input_required`` are observed only at their final state. * **No task cancellation harness.** Once :meth:`invoke` is awaited there is no handle to cancel a long-running task. """ def __init__( self, platform: DecisioningPlatform, *, validation: ValidationHookConfig | None = None, ) -> None: """ Args: platform: The :class:`~adcp.decisioning.DecisioningPlatform` instance under test. validation: Schema validation config. ``None`` (default) disables validation so tests focus on handler behavior, not schema conformance. Pass :data:`~adcp.validation.client_hooks.SERVER_DEFAULT_VALIDATION` to match production behavior. """ self._platform = platform self._validation = validation self._executor: Any | None = None self._executor_lock = asyncio.Lock() def _build_executor_sync(self) -> Any: from adcp.decisioning.serve import create_adcp_server_from_platform from adcp.server.a2a_server import ADCPAgentExecutor handler, _executor, _registry = create_adcp_server_from_platform( self._platform, auto_emit_completion_webhooks=False, ) return ADCPAgentExecutor(handler, validation=self._validation) async def _ensure_executor(self) -> Any: async with self._executor_lock: if self._executor is None: # create_adcp_server_from_platform calls asyncio.run() internally # (via validate_capabilities_response_shape) — must run in a thread # to avoid "cannot be called from a running event loop". self._executor = await asyncio.to_thread(self._build_executor_sync) return self._executor async def invoke( self, skill: str, payload: dict[str, Any] | None = None, *, timeout_seconds: float = 5.0, max_events: int = 32, ) -> ToolInvokeResult: """Invoke a skill and return the terminal-state result. Args: skill: AdCP skill name (e.g. ``"update_media_buy"``). payload: Arguments forwarded to the skill. ``None`` → empty dict. timeout_seconds: Per-event dequeue timeout. The harness drains the event queue waiting for a terminal Task; this caps how long any single dequeue waits. Default 5s. max_events: Maximum number of A2A events to drain while waiting for a terminal Task. Default 32 preserves the previous hard cap. Returns: :class:`ToolInvokeResult` — ``ok`` reflects whether the terminal Task state was ``COMPLETED``; ``adcp_error`` carries the structured error from a failed Task's DataPart per the A2A binding. """ # TODO(#699): Migrate off EventQueueLegacy when a2a-sdk ships a # type-clean successor. a2aproject/a2a-python#944 split the class # into an abstract EventQueue base + concrete EventQueueLegacy; the # base's __new__ redirects EventQueue() to EventQueueLegacy() at # runtime, but mypy correctly flags abstract instantiation and # a2a-sdk's own internals still use EventQueueLegacy. Tracked # upstream at a2aproject/a2a-python#1064. from a2a import types as pb from a2a.auth.user import UnauthenticatedUser from a2a.server.agent_execution.context import ( RequestContext as A2ARequestContext, ) from a2a.server.context import ServerCallContext from a2a.server.events.event_queue import EventQueueLegacy from google.protobuf.json_format import MessageToDict, ParseDict from google.protobuf.struct_pb2 import Value executor = await self._ensure_executor() kwargs = payload or {} # Build the DataPart-shaped skill invocation that ADCPAgentExecutor's # default parser accepts: {"skill": "...", "parameters": {...}}. value = Value() ParseDict({"skill": skill, "parameters": kwargs}, value) msg = pb.Message( message_id=str(uuid4()), role="ROLE_USER", parts=[pb.Part(data=value)], ) call_ctx = ServerCallContext(user=UnauthenticatedUser()) request_ctx = A2ARequestContext( call_context=call_ctx, request=pb.SendMessageRequest(message=msg), ) queue = EventQueueLegacy() await executor.execute(request_ctx, queue) # Drain the event queue until a terminal Task arrives. Bounded so a # buggy handler that never publishes a terminal event can't hang the # test runner — each dequeue carries `timeout_seconds`, and the loop # is bounded to `max_events` intermediate state events. terminal_task: Any = None for _ in range(max_events): try: event = await asyncio.wait_for(queue.dequeue_event(), timeout=timeout_seconds) except asyncio.TimeoutError: break if isinstance(event, pb.Task) and event.status.state in ( pb.TaskState.TASK_STATE_COMPLETED, pb.TaskState.TASK_STATE_FAILED, pb.TaskState.TASK_STATE_CANCELED, ): terminal_task = event break if terminal_task is None: raise RuntimeError( f"A2A executor for skill={skill!r} produced no terminal Task " f"within {timeout_seconds}s x {max_events} events — " "check executor middleware for hangs" ) # Project the terminal Task's first DataPart artifact to a dict. structured: dict[str, Any] = {} if terminal_task.artifacts: for part in terminal_task.artifacts[0].parts: if part.WhichOneof("content") == "data": projected = MessageToDict(part.data) if isinstance(projected, dict): structured = projected break raw_error = structured.get("adcp_error") adcp_error: AdcpErrorPayload | None = None if raw_error is not None: code = raw_error.get("code") message = raw_error.get("message") if not code: raise RuntimeError( "adcp_error envelope is missing required 'code' field; " "server is non-conformant to AdCP transport-errors spec" ) if not message: raise RuntimeError( "adcp_error envelope is missing required 'message' field; " "server is non-conformant to AdCP transport-errors spec" ) adcp_error = AdcpErrorPayload( code=code, message=message, recovery=raw_error.get("recovery"), field=raw_error.get("field"), suggestion=raw_error.get("suggestion"), retry_after=raw_error.get("retry_after"), details=raw_error.get("details"), ) elif terminal_task.status.state == pb.TaskState.TASK_STATE_FAILED: # A2A unstructured-error path: failed Task without an `adcp_error` # DataPart (unknown skill, no parseable skill, transport-layer # rejection). Synthesize an envelope so callers can assert on # ``result.adcp_error`` uniformly across structured/unstructured # failures. status_msg = "" if terminal_task.status.HasField("message"): for part in terminal_task.status.message.parts: if part.WhichOneof("content") == "text": status_msg = part.text break adcp_error = AdcpErrorPayload( code="INTERNAL_ERROR", message=status_msg or "A2A task failed without structured adcp_error envelope", ) data: dict[str, Any] | None = None if raw_error is None and structured: data = dict(structured) return ToolInvokeResult(data=data, adcp_error=adcp_error, structured_content=structured)In-process A2A test client for AdCP seller implementations.
The A2A sibling to :class:
SellerTestClient. Same call shape (await client.invoke(skill, payload)), same return type (:class:ToolInvokeResult), but routes through the A2A executor + event-queue dispatch path instead of MCP's tool call.Usage::
@pytest.fixture def seller_a2a(): return SellerA2AClient(MySeller()) async def test_buy_not_found_a2a(seller_a2a): result = await seller_a2a.invoke( "update_media_buy", {"media_buy_id": "missing", ...} ) assert not result.ok assert result.adcp_error.code == "MEDIA_BUY_NOT_FOUND"The harness constructs an :class:
~adcp.server.a2a_server.ADCPAgentExecutor, builds a minimal A2A :class:~a2a.server.agent_execution.context.RequestContextcarrying aDataPartwith{"skill": ..., "parameters": ...}, runs the executor against a fresh event queue, and drains the queue until a terminal Task arrives. Success payloads come from the Task's first DataPart artifact; structured errors land in that same DataPart keyed underadcp_errorper transport-errors.mdx §A2A Binding.Limitations (each tracked as a follow-up on #678):
- No push-notification capture. Adopters who need to assert on outbound signed webhook delivery need a sink primitive this client doesn't yet provide.
- No intermediate state observation. :meth:
invokedrains to the terminal Task and returns. Tasks that pass throughworking/input_requiredare observed only at their final state. - No task cancellation harness. Once :meth:
invokeis awaited there is no handle to cancel a long-running task.
- Args
- -----=
platform- The :class:
~adcp.decisioning.DecisioningPlatforminstance under test. validation- Schema validation config.
None(default) disables validation so tests focus on handler behavior, not schema conformance. Pass :data:~adcp.validation.client_hooks.SERVER_DEFAULT_VALIDATIONto match production behavior.
Methods
async def invoke(self,
skill: str,
payload: dict[str, Any] | None = None,
*,
timeout_seconds: float = 5.0,
max_events: int = 32) ‑> ToolInvokeResult-
Expand source code
async def invoke( self, skill: str, payload: dict[str, Any] | None = None, *, timeout_seconds: float = 5.0, max_events: int = 32, ) -> ToolInvokeResult: """Invoke a skill and return the terminal-state result. Args: skill: AdCP skill name (e.g. ``"update_media_buy"``). payload: Arguments forwarded to the skill. ``None`` → empty dict. timeout_seconds: Per-event dequeue timeout. The harness drains the event queue waiting for a terminal Task; this caps how long any single dequeue waits. Default 5s. max_events: Maximum number of A2A events to drain while waiting for a terminal Task. Default 32 preserves the previous hard cap. Returns: :class:`ToolInvokeResult` — ``ok`` reflects whether the terminal Task state was ``COMPLETED``; ``adcp_error`` carries the structured error from a failed Task's DataPart per the A2A binding. """ # TODO(#699): Migrate off EventQueueLegacy when a2a-sdk ships a # type-clean successor. a2aproject/a2a-python#944 split the class # into an abstract EventQueue base + concrete EventQueueLegacy; the # base's __new__ redirects EventQueue() to EventQueueLegacy() at # runtime, but mypy correctly flags abstract instantiation and # a2a-sdk's own internals still use EventQueueLegacy. Tracked # upstream at a2aproject/a2a-python#1064. from a2a import types as pb from a2a.auth.user import UnauthenticatedUser from a2a.server.agent_execution.context import ( RequestContext as A2ARequestContext, ) from a2a.server.context import ServerCallContext from a2a.server.events.event_queue import EventQueueLegacy from google.protobuf.json_format import MessageToDict, ParseDict from google.protobuf.struct_pb2 import Value executor = await self._ensure_executor() kwargs = payload or {} # Build the DataPart-shaped skill invocation that ADCPAgentExecutor's # default parser accepts: {"skill": "...", "parameters": {...}}. value = Value() ParseDict({"skill": skill, "parameters": kwargs}, value) msg = pb.Message( message_id=str(uuid4()), role="ROLE_USER", parts=[pb.Part(data=value)], ) call_ctx = ServerCallContext(user=UnauthenticatedUser()) request_ctx = A2ARequestContext( call_context=call_ctx, request=pb.SendMessageRequest(message=msg), ) queue = EventQueueLegacy() await executor.execute(request_ctx, queue) # Drain the event queue until a terminal Task arrives. Bounded so a # buggy handler that never publishes a terminal event can't hang the # test runner — each dequeue carries `timeout_seconds`, and the loop # is bounded to `max_events` intermediate state events. terminal_task: Any = None for _ in range(max_events): try: event = await asyncio.wait_for(queue.dequeue_event(), timeout=timeout_seconds) except asyncio.TimeoutError: break if isinstance(event, pb.Task) and event.status.state in ( pb.TaskState.TASK_STATE_COMPLETED, pb.TaskState.TASK_STATE_FAILED, pb.TaskState.TASK_STATE_CANCELED, ): terminal_task = event break if terminal_task is None: raise RuntimeError( f"A2A executor for skill={skill!r} produced no terminal Task " f"within {timeout_seconds}s x {max_events} events — " "check executor middleware for hangs" ) # Project the terminal Task's first DataPart artifact to a dict. structured: dict[str, Any] = {} if terminal_task.artifacts: for part in terminal_task.artifacts[0].parts: if part.WhichOneof("content") == "data": projected = MessageToDict(part.data) if isinstance(projected, dict): structured = projected break raw_error = structured.get("adcp_error") adcp_error: AdcpErrorPayload | None = None if raw_error is not None: code = raw_error.get("code") message = raw_error.get("message") if not code: raise RuntimeError( "adcp_error envelope is missing required 'code' field; " "server is non-conformant to AdCP transport-errors spec" ) if not message: raise RuntimeError( "adcp_error envelope is missing required 'message' field; " "server is non-conformant to AdCP transport-errors spec" ) adcp_error = AdcpErrorPayload( code=code, message=message, recovery=raw_error.get("recovery"), field=raw_error.get("field"), suggestion=raw_error.get("suggestion"), retry_after=raw_error.get("retry_after"), details=raw_error.get("details"), ) elif terminal_task.status.state == pb.TaskState.TASK_STATE_FAILED: # A2A unstructured-error path: failed Task without an `adcp_error` # DataPart (unknown skill, no parseable skill, transport-layer # rejection). Synthesize an envelope so callers can assert on # ``result.adcp_error`` uniformly across structured/unstructured # failures. status_msg = "" if terminal_task.status.HasField("message"): for part in terminal_task.status.message.parts: if part.WhichOneof("content") == "text": status_msg = part.text break adcp_error = AdcpErrorPayload( code="INTERNAL_ERROR", message=status_msg or "A2A task failed without structured adcp_error envelope", ) data: dict[str, Any] | None = None if raw_error is None and structured: data = dict(structured) return ToolInvokeResult(data=data, adcp_error=adcp_error, structured_content=structured)Invoke a skill and return the terminal-state result.
- Args
- -----=
skill- AdCP skill name (e.g.
"update_media_buy"). payload- Arguments forwarded to the skill.
None→ empty dict. timeout_seconds- Per-event dequeue timeout. The harness drains the event queue waiting for a terminal Task; this caps how long any single dequeue waits. Default 5s.
max_events- Maximum number of A2A events to drain while waiting for a terminal Task. Default 32 preserves the previous hard cap.
Returns -----= :class:
ToolInvokeResult—okreflects whether the terminal Task state wasCOMPLETED;adcp_errorcarries the structured error from a failed Task's DataPart per the A2A binding.
class SellerTestClient (platform: DecisioningPlatform,
*,
name: str | None = None,
validation: ValidationHookConfig | None = None)-
Expand source code
class SellerTestClient: """In-process MCP test client for AdCP seller implementations. Wraps a :class:`~adcp.decisioning.DecisioningPlatform` with :meth:`invoke` — a one-call helper that handles tool dispatch and ``adcp_error`` extraction from ``structuredContent``, eliminating the boilerplate every adopter previously rewrote in each test file. Usage:: @pytest.fixture def seller(): return SellerTestClient(MySeller()) async def test_buy_not_found(seller): result = await seller.invoke( "update_media_buy", {"media_buy_id": "missing", ...} ) assert not result.ok assert result.adcp_error.code == "MEDIA_BUY_NOT_FOUND" async def test_get_products_success(seller): result = await seller.invoke("get_products", {}) assert result.ok assert "products" in result.data The harness calls :meth:`mcp.call_tool` in-process — it exercises the full handler dispatch and error-translation stack without HTTP or SSE framing. For HTTP-level tests (auth middleware, CORS, size limits), use :func:`adcp.testing.build_test_client` directly. For A2A transport, use :class:`SellerA2AClient` — same call shape (:meth:`SellerA2AClient.invoke`), different in-process dispatch path (executor + event queue rather than MCP tool call). This client covers per-handler unit and integration tests. Full compliance scenario grading runs through the ``@adcp/sdk`` storyboard runner (``adcp storyboard run``), whose canonical scenarios and grader are the single source of truth. See ``docs/testing-your-adcp-server.md`` for both paths. """ def __init__( self, platform: DecisioningPlatform, *, name: str | None = None, validation: ValidationHookConfig | None = None, ) -> None: """ Args: platform: The :class:`~adcp.decisioning.DecisioningPlatform` instance under test. name: Server name forwarded to :func:`~adcp.server.serve.create_mcp_server`. Defaults to ``type(platform).__name__``. validation: Schema validation config. ``None`` (default) disables validation so tests focus on handler behavior, not schema conformance. Pass :data:`~adcp.validation.client_hooks.SERVER_DEFAULT_VALIDATION` to match production behavior. """ self._platform = platform self._name = name self._validation = validation self._mcp: Any | None = None self._mcp_lock = asyncio.Lock() def _build_mcp_sync(self) -> Any: from adcp.decisioning.serve import create_adcp_server_from_platform from adcp.server.serve import create_mcp_server handler, _executor, _registry = create_adcp_server_from_platform( self._platform, auto_emit_completion_webhooks=False, ) return create_mcp_server( handler, name=self._name or type(self._platform).__name__, validation=self._validation, ) async def _ensure_mcp(self) -> Any: async with self._mcp_lock: if self._mcp is None: # create_adcp_server_from_platform calls asyncio.run() internally # (via validate_capabilities_response_shape) — must run in a thread # to avoid "cannot be called from a running event loop". self._mcp = await asyncio.to_thread(self._build_mcp_sync) return self._mcp async def invoke( self, tool: str, payload: dict[str, Any] | None = None, ) -> ToolInvokeResult: """Invoke a tool and return the result with adcp_error extraction. Args: tool: AdCP tool name (e.g. ``"update_media_buy"``). payload: Arguments forwarded to the tool. ``None`` → empty dict. Returns: :class:`ToolInvokeResult` — check :attr:`~ToolInvokeResult.ok` for success/failure and :attr:`~ToolInvokeResult.adcp_error` for error details. """ from mcp.types import CallToolResult mcp = await self._ensure_mcp() kwargs = payload or {} raw_result = await mcp.call_tool(tool, kwargs) # Success shape comes from this SDK's `_AdcpFuncMetadata.convert_result` # (`src/adcp/server/serve.py`), which yields a 2-tuple of # `(list[ContentBlock], dict | None)` — that's a coupling to our internal # FuncMetadata override, not the public FastMCP.call_tool contract. # Error shape (AdcpError raised by handler) lands as `CallToolResult` # with `isError=True` and `structuredContent={"adcp_error": {...}}`. # The plain-dict branch is defensive against FastMCP flattening the # success return in a future version. if isinstance(raw_result, CallToolResult): structured: dict[str, Any] = raw_result.structured_content or {} elif isinstance(raw_result, (tuple, list)) and len(raw_result) == 2: success_dict: Any = raw_result[1] structured = dict(success_dict) if success_dict is not None else {} elif isinstance(raw_result, dict): structured = dict(raw_result) else: raise RuntimeError( f"FastMCP.call_tool returned unexpected shape {type(raw_result)!r}; " "harness needs updating for this MCP version" ) raw_error = structured.get("adcp_error") adcp_error: AdcpErrorPayload | None = None if raw_error is not None: code = raw_error.get("code") message = raw_error.get("message") if not code: raise RuntimeError( "adcp_error envelope is missing required 'code' field; " "server is non-conformant to AdCP transport-errors spec" ) if not message: raise RuntimeError( "adcp_error envelope is missing required 'message' field; " "server is non-conformant to AdCP transport-errors spec" ) adcp_error = AdcpErrorPayload( code=code, message=message, recovery=raw_error.get("recovery"), field=raw_error.get("field"), suggestion=raw_error.get("suggestion"), retry_after=raw_error.get("retry_after"), details=raw_error.get("details"), ) data: dict[str, Any] | None = None if raw_error is None and structured: data = dict(structured) return ToolInvokeResult(data=data, adcp_error=adcp_error, structured_content=structured)In-process MCP test client for AdCP seller implementations.
Wraps a :class:
~adcp.decisioning.DecisioningPlatformwith :meth:invoke— a one-call helper that handles tool dispatch andadcp_errorextraction fromstructuredContent, eliminating the boilerplate every adopter previously rewrote in each test file.Usage::
@pytest.fixture def seller(): return SellerTestClient(MySeller()) async def test_buy_not_found(seller): result = await seller.invoke( "update_media_buy", {"media_buy_id": "missing", ...} ) assert not result.ok assert result.adcp_error.code == "MEDIA_BUY_NOT_FOUND" async def test_get_products_success(seller): result = await seller.invoke("get_products", {}) assert result.ok assert "products" in result.dataThe harness calls :meth:
mcp.call_toolin-process — it exercises the full handler dispatch and error-translation stack without HTTP or SSE framing. For HTTP-level tests (auth middleware, CORS, size limits), use :func:build_test_client()directly.For A2A transport, use :class:
SellerA2AClient— same call shape (:meth:SellerA2AClient.invoke()), different in-process dispatch path (executor + event queue rather than MCP tool call).This client covers per-handler unit and integration tests. Full compliance scenario grading runs through the
@adcp/sdkstoryboard runner (adcp storyboard run), whose canonical scenarios and grader are the single source of truth. Seedocs/testing-your-adcp-server.mdfor both paths.- Args
- -----=
platform- The :class:
~adcp.decisioning.DecisioningPlatforminstance under test. name- Server name forwarded to :func:
~adcp.server.serve.create_mcp_server. Defaults totype(platform).__name__. validation- Schema validation config.
None(default) disables validation so tests focus on handler behavior, not schema conformance. Pass :data:~adcp.validation.client_hooks.SERVER_DEFAULT_VALIDATIONto match production behavior.
Methods
async def invoke(self, tool: str, payload: dict[str, Any] | None = None) ‑> ToolInvokeResult-
Expand source code
async def invoke( self, tool: str, payload: dict[str, Any] | None = None, ) -> ToolInvokeResult: """Invoke a tool and return the result with adcp_error extraction. Args: tool: AdCP tool name (e.g. ``"update_media_buy"``). payload: Arguments forwarded to the tool. ``None`` → empty dict. Returns: :class:`ToolInvokeResult` — check :attr:`~ToolInvokeResult.ok` for success/failure and :attr:`~ToolInvokeResult.adcp_error` for error details. """ from mcp.types import CallToolResult mcp = await self._ensure_mcp() kwargs = payload or {} raw_result = await mcp.call_tool(tool, kwargs) # Success shape comes from this SDK's `_AdcpFuncMetadata.convert_result` # (`src/adcp/server/serve.py`), which yields a 2-tuple of # `(list[ContentBlock], dict | None)` — that's a coupling to our internal # FuncMetadata override, not the public FastMCP.call_tool contract. # Error shape (AdcpError raised by handler) lands as `CallToolResult` # with `isError=True` and `structuredContent={"adcp_error": {...}}`. # The plain-dict branch is defensive against FastMCP flattening the # success return in a future version. if isinstance(raw_result, CallToolResult): structured: dict[str, Any] = raw_result.structured_content or {} elif isinstance(raw_result, (tuple, list)) and len(raw_result) == 2: success_dict: Any = raw_result[1] structured = dict(success_dict) if success_dict is not None else {} elif isinstance(raw_result, dict): structured = dict(raw_result) else: raise RuntimeError( f"FastMCP.call_tool returned unexpected shape {type(raw_result)!r}; " "harness needs updating for this MCP version" ) raw_error = structured.get("adcp_error") adcp_error: AdcpErrorPayload | None = None if raw_error is not None: code = raw_error.get("code") message = raw_error.get("message") if not code: raise RuntimeError( "adcp_error envelope is missing required 'code' field; " "server is non-conformant to AdCP transport-errors spec" ) if not message: raise RuntimeError( "adcp_error envelope is missing required 'message' field; " "server is non-conformant to AdCP transport-errors spec" ) adcp_error = AdcpErrorPayload( code=code, message=message, recovery=raw_error.get("recovery"), field=raw_error.get("field"), suggestion=raw_error.get("suggestion"), retry_after=raw_error.get("retry_after"), details=raw_error.get("details"), ) data: dict[str, Any] | None = None if raw_error is None and structured: data = dict(structured) return ToolInvokeResult(data=data, adcp_error=adcp_error, structured_content=structured)Invoke a tool and return the result with adcp_error extraction.
- Args
- -----=
tool- AdCP tool name (e.g.
"update_media_buy"). payload- Arguments forwarded to the tool.
None→ empty dict.
Returns -----= :class:
ToolInvokeResult— check :attr:~ToolInvokeResult.okfor success/failure and :attr:~ToolInvokeResult.adcp_errorfor error details.
class ToolInvokeResult (data: dict[str, Any] | None,
adcp_error: AdcpErrorPayload | None,
structured_content: dict[str, Any])-
Expand source code
@dataclass(frozen=True) class ToolInvokeResult: """Result of a :meth:`SellerTestClient.invoke` call.""" data: dict[str, Any] | None adcp_error: AdcpErrorPayload | None structured_content: dict[str, Any] @property def ok(self) -> bool: """True when the tool returned a success envelope (no adcp_error).""" return self.adcp_error is NoneResult of a :meth:
SellerTestClient.invoke()call.Instance variables
var adcp_error : AdcpErrorPayload | Nonevar data : dict[str, typing.Any] | Noneprop ok : bool-
Expand source code
@property def ok(self) -> bool: """True when the tool returned a success envelope (no adcp_error).""" return self.adcp_error is NoneTrue when the tool returned a success envelope (no adcp_error).
var structured_content : dict[str, typing.Any]
class UpstreamRecorder (*,
max_payload_bytes: int = 65536,
purpose: Callable[[RecordedCall], Purpose | None] | None = None)-
Expand source code
class UpstreamRecorder: """Records principal-scoped outbound HTTP traffic for conformance tests. Bind a principal with :meth:`principal_scope` (sync) or :meth:`run_with_principal` (async), make httpx calls through a client wired with :meth:`httpx_hooks`, then :meth:`query` the buffer. Args: max_payload_bytes: Per-call payload cap. Payloads whose UTF-8 length exceeds this are truncated with a trailing marker. Defaults to 64 KiB per the spec recommendation. purpose: Optional classifier called with each :class:`RecordedCall` (pre-purpose) to tag the call's semantic role. Calls without a purpose are treated as ``other`` by ``purpose_filter`` matching. """ def __init__( self, *, max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES, purpose: Callable[[RecordedCall], Purpose | None] | None = None, ) -> None: self._max_payload_bytes = max_payload_bytes self._purpose = purpose # Principal-keyed buffer; multi-tenant sandboxes MUST key on the # invocation's auth principal, not just process-global. self._buffer: dict[str, list[RecordedCall]] = {} # -- principal scoping ------------------------------------------------ @contextmanager def principal_scope(self, principal: str) -> Iterator[None]: """Bind ``principal`` for the duration of the ``with`` block. Calls recorded inside the block are attributed to ``principal``. Raises :class:`UpstreamRecorderScopeError` if ``principal`` is empty or not a string. """ token = _principal_var.set(_require_principal(principal)) try: yield finally: _principal_var.reset(token) @asynccontextmanager async def principal_scope_async(self, principal: str) -> AsyncIterator[None]: """Async sibling of :meth:`principal_scope`. The ContextVar binding survives ``await`` boundaries within the block. Note that ``asyncio.create_task`` copies the current context at spawn time, so tasks spawned inside the block inherit the principal; tasks spawned outside it record under no scope and are dropped (fail-closed). """ token = _principal_var.set(_require_principal(principal)) try: yield finally: _principal_var.reset(token) async def run_with_principal(self, principal: str, coro: Awaitable[Any]) -> Any: """Await ``coro`` with ``principal`` bound for its duration.""" async with self.principal_scope_async(principal): return await coro # -- recording -------------------------------------------------------- def record( self, *, method: str, url: str, content_type: str, payload: Any = None, headers: Mapping[str, str] | None = None, status_code: int | None = None, timestamp: datetime | None = None, principal: str | None = None, ) -> RecordedCall | None: """Record one outbound call. Manual escape hatch for non-httpx clients. Returns the stored :class:`RecordedCall`, or ``None`` when no principal is bound and none is supplied — an unattributable call is dropped rather than recorded globally. Redaction is applied here, at record time, across the surfaces that reach the wire: secret-keyed query-param values in ``url`` (which also feeds ``endpoint``) and secret-keyed values in ``payload`` (JSON and form-urlencoded bodies) are replaced with ``[redacted]`` before storage. The ``headers`` argument carries no header to the wire (the recorded-call shape has no header field) and is unused. """ bound = principal if principal is not None else _principal_var.get() if not bound: return None bound = _require_principal(bound) redacted_url = _redact_url(url) split = urlsplit(redacted_url) redacted_payload, payload_length = self._normalize_payload(payload, content_type) call = RecordedCall( method=method.upper(), # type: ignore[arg-type] url=redacted_url, host=split.hostname or "", path=split.path, content_type=content_type, payload=redacted_payload, payload_length=payload_length, timestamp=timestamp or datetime.now(timezone.utc), status_code=status_code, principal=bound, ) if self._purpose is not None: purpose = self._purpose(call) if purpose is not None: call = replace(call, purpose=purpose) self._buffer.setdefault(bound, []).append(call) return call def _normalize_payload(self, payload: Any, content_type: str) -> tuple[Any, int]: """Redact and size a payload; return ``(redacted_payload, byte_length)``. JSON-shaped bodies are decoded, recursively redacted, and re-encoded. Form-urlencoded bodies are parsed, key-redacted, and re-encoded. Other bodies are coerced to a string and emitted as-is (no structure to walk). ``payload_length`` is the UTF-8 byte length of the *emitted* value — after redaction and any truncation — matching the spec's raw-mode definition (``payload_length`` MUST equal the emitted bytes). """ if payload is None: return None, 0 ct = content_type.lower() is_json = "json" in ct is_form = "application/x-www-form-urlencoded" in ct decoded: Any = payload if isinstance(payload, (bytes, bytearray)): text = payload.decode("utf-8", errors="replace") if is_json: try: decoded = json.loads(text) except (json.JSONDecodeError, ValueError): decoded = text else: decoded = text elif isinstance(payload, str) and is_json: try: decoded = json.loads(payload) except (json.JSONDecodeError, ValueError): decoded = payload if isinstance(decoded, (dict, list)): redacted = _redact(decoded) measured = json.dumps(redacted, separators=(",", ":")).encode("utf-8") if len(measured) > self._max_payload_bytes: # Object payloads can't carry an inline truncation marker # without breaking JSON; truncate the serialized form to a # string and report the emitted (truncated) byte length. return self._truncate_string(measured.decode("utf-8", errors="ignore")) return redacted, len(measured) text = decoded if isinstance(decoded, str) else str(decoded) if is_form: text = _redact_form_body(text) encoded = text.encode("utf-8") if len(encoded) > self._max_payload_bytes: return self._truncate_string(text) return text, len(encoded) def _truncate_string(self, text: str) -> tuple[str, int]: """Truncate ``text`` to fit within ``max_payload_bytes`` including marker. Reserves room for ``_TRUNCATION_MARKER`` so the emitted string stays within the schema's ``payload`` ``maxLength``, and reports the UTF-8 byte length of the emitted (truncated) value per the raw-mode contract. """ marker_bytes = len(_TRUNCATION_MARKER.encode("utf-8")) budget = max(self._max_payload_bytes - marker_bytes, 0) head = text.encode("utf-8")[:budget].decode("utf-8", errors="ignore") emitted = head + _TRUNCATION_MARKER return emitted, len(emitted.encode("utf-8")) # -- httpx integration ------------------------------------------------ def httpx_hooks(self) -> dict[str, list[Callable[..., Awaitable[None]]]]: """Return ``event_hooks`` for an :class:`httpx.AsyncClient`. Pass to ``httpx.AsyncClient(event_hooks=recorder.httpx_hooks())``. The response hook records the call once the status code is known, so each call is captured exactly once with its ``status_code``. """ return {"response": [self._on_response]} async def _on_response(self, response: httpx.Response) -> None: import httpx request = response.request content_type = request.headers.get("content-type", "") # httpx exposes the request body bytes via `request.content`; a # streaming request body (generator / async-iterator) has not been # read and raises RequestNotRead — there is nothing to record. try: payload: Any = bytes(request.content) if request.content else None except httpx.RequestNotRead: payload = None self.record( method=request.method, url=str(request.url), content_type=content_type, payload=payload, headers=request.headers, status_code=response.status_code, ) # -- querying --------------------------------------------------------- def query( self, *, principal: str, since_timestamp: datetime | None = None, endpoint_pattern: str | None = None, limit: int = 100, ) -> UpstreamTrafficResult: """Return calls recorded for ``principal``, scoped and ordered. Caller scoping is enforced: only ``principal``'s calls are considered — cross-caller traffic is never returned. Results are ordered by ``timestamp`` ascending and capped at ``limit``; ``total_count`` / ``truncated`` reflect the full match set before the cap. Args: principal: Authenticated caller identity. Required and validated — derive it from the ``comply_test_controller`` invocation's auth context, never from caller-supplied request params. since_timestamp: Only return calls at or after this time. ``None`` returns the whole buffer (session start). endpoint_pattern: Optional ``<METHOD> <URL>`` glob. ``*`` matches any run of characters; anchored full-string match. limit: Maximum calls to return (default 100). """ principal = _require_principal(principal) if limit < 1: raise ValueError(f"limit must be >= 1, got {limit}") calls = self._buffer.get(principal, []) matched = [ call for call in calls if (since_timestamp is None or call.timestamp >= since_timestamp) ] if endpoint_pattern is not None: regex = _compile_endpoint_pattern(endpoint_pattern) matched = [call for call in matched if regex.match(call.endpoint)] matched.sort(key=lambda c: c.timestamp) total = len(matched) page = tuple(matched[:limit]) effective_since = since_timestamp or ( page[0].timestamp if page else datetime.now(timezone.utc) ) return UpstreamTrafficResult( recorded_calls=page, total_count=total, since_timestamp=effective_since, truncated=total > len(page), ) # -- introspection ---------------------------------------------------- def debug(self) -> dict[str, int]: """Return per-principal call counts for test introspection.""" return {principal: len(calls) for principal, calls in self._buffer.items()} def clear(self) -> None: """Drop all recorded calls.""" self._buffer.clear()Records principal-scoped outbound HTTP traffic for conformance tests.
Bind a principal with :meth:
principal_scope(sync) or :meth:run_with_principal(async), make httpx calls through a client wired with :meth:httpx_hooks, then :meth:querythe buffer.- Args
- -----=
max_payload_bytes- Per-call payload cap. Payloads whose UTF-8 length exceeds this are truncated with a trailing marker. Defaults to 64 KiB per the spec recommendation.
purpose- Optional classifier called with each :class:
RecordedCall(pre-purpose) to tag the call's semantic role. Calls without a purpose are treated asotherbypurpose_filtermatching.
Methods
def clear(self) ‑> None-
Expand source code
def clear(self) -> None: """Drop all recorded calls.""" self._buffer.clear()Drop all recorded calls.
def debug(self) ‑> dict[str, int]-
Expand source code
def debug(self) -> dict[str, int]: """Return per-principal call counts for test introspection.""" return {principal: len(calls) for principal, calls in self._buffer.items()}Return per-principal call counts for test introspection.
def httpx_hooks(self) ‑> dict[str, list[Callable[..., Awaitable[None]]]]-
Expand source code
def httpx_hooks(self) -> dict[str, list[Callable[..., Awaitable[None]]]]: """Return ``event_hooks`` for an :class:`httpx.AsyncClient`. Pass to ``httpx.AsyncClient(event_hooks=recorder.httpx_hooks())``. The response hook records the call once the status code is known, so each call is captured exactly once with its ``status_code``. """ return {"response": [self._on_response]}Return
event_hooksfor an :class:httpx.AsyncClient.Pass to
httpx.AsyncClient(event_hooks=recorder.httpx_hooks()). The response hook records the call once the status code is known, so each call is captured exactly once with itsstatus_code. def principal_scope(self, principal: str) ‑> Iterator[None]-
Expand source code
@contextmanager def principal_scope(self, principal: str) -> Iterator[None]: """Bind ``principal`` for the duration of the ``with`` block. Calls recorded inside the block are attributed to ``principal``. Raises :class:`UpstreamRecorderScopeError` if ``principal`` is empty or not a string. """ token = _principal_var.set(_require_principal(principal)) try: yield finally: _principal_var.reset(token)Bind
principalfor the duration of thewithblock.Calls recorded inside the block are attributed to
principal. Raises :class:UpstreamRecorderScopeErrorifprincipalis empty or not a string. async def principal_scope_async(self, principal: str) ‑> AsyncIterator[None]-
Expand source code
@asynccontextmanager async def principal_scope_async(self, principal: str) -> AsyncIterator[None]: """Async sibling of :meth:`principal_scope`. The ContextVar binding survives ``await`` boundaries within the block. Note that ``asyncio.create_task`` copies the current context at spawn time, so tasks spawned inside the block inherit the principal; tasks spawned outside it record under no scope and are dropped (fail-closed). """ token = _principal_var.set(_require_principal(principal)) try: yield finally: _principal_var.reset(token)Async sibling of :meth:
principal_scope.The ContextVar binding survives
awaitboundaries within the block. Note thatasyncio.create_taskcopies the current context at spawn time, so tasks spawned inside the block inherit the principal; tasks spawned outside it record under no scope and are dropped (fail-closed). def query(self,
*,
principal: str,
since_timestamp: datetime | None = None,
endpoint_pattern: str | None = None,
limit: int = 100) ‑> UpstreamTrafficResult-
Expand source code
def query( self, *, principal: str, since_timestamp: datetime | None = None, endpoint_pattern: str | None = None, limit: int = 100, ) -> UpstreamTrafficResult: """Return calls recorded for ``principal``, scoped and ordered. Caller scoping is enforced: only ``principal``'s calls are considered — cross-caller traffic is never returned. Results are ordered by ``timestamp`` ascending and capped at ``limit``; ``total_count`` / ``truncated`` reflect the full match set before the cap. Args: principal: Authenticated caller identity. Required and validated — derive it from the ``comply_test_controller`` invocation's auth context, never from caller-supplied request params. since_timestamp: Only return calls at or after this time. ``None`` returns the whole buffer (session start). endpoint_pattern: Optional ``<METHOD> <URL>`` glob. ``*`` matches any run of characters; anchored full-string match. limit: Maximum calls to return (default 100). """ principal = _require_principal(principal) if limit < 1: raise ValueError(f"limit must be >= 1, got {limit}") calls = self._buffer.get(principal, []) matched = [ call for call in calls if (since_timestamp is None or call.timestamp >= since_timestamp) ] if endpoint_pattern is not None: regex = _compile_endpoint_pattern(endpoint_pattern) matched = [call for call in matched if regex.match(call.endpoint)] matched.sort(key=lambda c: c.timestamp) total = len(matched) page = tuple(matched[:limit]) effective_since = since_timestamp or ( page[0].timestamp if page else datetime.now(timezone.utc) ) return UpstreamTrafficResult( recorded_calls=page, total_count=total, since_timestamp=effective_since, truncated=total > len(page), )Return calls recorded for
principal, scoped and ordered.Caller scoping is enforced: only
principal's calls are considered — cross-caller traffic is never returned. Results are ordered bytimestampascending and capped atlimit;total_count/truncatedreflect the full match set before the cap.- Args
- -----=
principal- Authenticated caller identity. Required and validated
— derive it from the
comply_test_controllerinvocation's auth context, never from caller-supplied request params. since_timestamp- Only return calls at or after this time.
Nonereturns the whole buffer (session start). endpoint_pattern- Optional
<METHOD> <URL>glob.*matches any run of characters; anchored full-string match. limit- Maximum calls to return (default 100).
def record(self,
*,
method: str,
url: str,
content_type: str,
payload: Any = None,
headers: Mapping[str, str] | None = None,
status_code: int | None = None,
timestamp: datetime | None = None,
principal: str | None = None) ‑> RecordedCall | None-
Expand source code
def record( self, *, method: str, url: str, content_type: str, payload: Any = None, headers: Mapping[str, str] | None = None, status_code: int | None = None, timestamp: datetime | None = None, principal: str | None = None, ) -> RecordedCall | None: """Record one outbound call. Manual escape hatch for non-httpx clients. Returns the stored :class:`RecordedCall`, or ``None`` when no principal is bound and none is supplied — an unattributable call is dropped rather than recorded globally. Redaction is applied here, at record time, across the surfaces that reach the wire: secret-keyed query-param values in ``url`` (which also feeds ``endpoint``) and secret-keyed values in ``payload`` (JSON and form-urlencoded bodies) are replaced with ``[redacted]`` before storage. The ``headers`` argument carries no header to the wire (the recorded-call shape has no header field) and is unused. """ bound = principal if principal is not None else _principal_var.get() if not bound: return None bound = _require_principal(bound) redacted_url = _redact_url(url) split = urlsplit(redacted_url) redacted_payload, payload_length = self._normalize_payload(payload, content_type) call = RecordedCall( method=method.upper(), # type: ignore[arg-type] url=redacted_url, host=split.hostname or "", path=split.path, content_type=content_type, payload=redacted_payload, payload_length=payload_length, timestamp=timestamp or datetime.now(timezone.utc), status_code=status_code, principal=bound, ) if self._purpose is not None: purpose = self._purpose(call) if purpose is not None: call = replace(call, purpose=purpose) self._buffer.setdefault(bound, []).append(call) return callRecord one outbound call. Manual escape hatch for non-httpx clients.
Returns the stored :class:
RecordedCall, orNonewhen no principal is bound and none is supplied — an unattributable call is dropped rather than recorded globally.Redaction is applied here, at record time, across the surfaces that reach the wire: secret-keyed query-param values in
url(which also feedsendpoint) and secret-keyed values inpayload(JSON and form-urlencoded bodies) are replaced with[redacted]before storage. Theheadersargument carries no header to the wire (the recorded-call shape has no header field) and is unused. async def run_with_principal(self, principal: str, coro: Awaitable[Any]) ‑> Any-
Expand source code
async def run_with_principal(self, principal: str, coro: Awaitable[Any]) -> Any: """Await ``coro`` with ``principal`` bound for its duration.""" async with self.principal_scope_async(principal): return await coroAwait
corowithprincipalbound for its duration.
class UpstreamRecorderScopeError (*args, **kwargs)-
Expand source code
class UpstreamRecorderScopeError(RuntimeError): """Raised when a principal scope is opened or queried with a bad principal. A principal must be a non-empty string. Binding or querying with an empty / non-string principal is a wiring bug, not a no-match — failing closed prevents traffic from silently leaking into an unqueryable or globally readable bucket. """Raised when a principal scope is opened or queried with a bad principal.
A principal must be a non-empty string. Binding or querying with an empty / non-string principal is a wiring bug, not a no-match — failing closed prevents traffic from silently leaking into an unqueryable or globally readable bucket.
Ancestors
- builtins.RuntimeError
- builtins.Exception
- builtins.BaseException
class UpstreamTrafficResult (recorded_calls: tuple[RecordedCall, ...],
total_count: int,
since_timestamp: datetime,
truncated: bool)-
Expand source code
@dataclass(frozen=True) class UpstreamTrafficResult: """Outcome of :meth:`UpstreamRecorder.query`, shaped for the wire response.""" recorded_calls: tuple[RecordedCall, ...] total_count: int since_timestamp: datetime truncated: bool def to_response_dict(self) -> dict[str, Any]: """Project to an ``UpstreamTrafficSuccess`` payload. The ``success`` / ``recorded_calls`` / ``total_count`` / ``since_timestamp`` fields are required by the response schema; a ``comply_test_controller`` handler returns this directly. """ return { "success": True, "recorded_calls": [call.to_recorded_call_dict() for call in self.recorded_calls], "total_count": self.total_count, "truncated": self.truncated, "since_timestamp": self.since_timestamp.isoformat(), }Outcome of :meth:
UpstreamRecorder.query(), shaped for the wire response.Instance variables
var recorded_calls : tuple[RecordedCall, ...]var since_timestamp : datetime.datetimevar total_count : intvar truncated : bool
Methods
def to_response_dict(self) ‑> dict[str, typing.Any]-
Expand source code
def to_response_dict(self) -> dict[str, Any]: """Project to an ``UpstreamTrafficSuccess`` payload. The ``success`` / ``recorded_calls`` / ``total_count`` / ``since_timestamp`` fields are required by the response schema; a ``comply_test_controller`` handler returns this directly. """ return { "success": True, "recorded_calls": [call.to_recorded_call_dict() for call in self.recorded_calls], "total_count": self.total_count, "truncated": self.truncated, "since_timestamp": self.since_timestamp.isoformat(), }Project to an
UpstreamTrafficSuccesspayload.The
success/recorded_calls/total_count/since_timestampfields are required by the response schema; acomply_test_controllerhandler returns this directly.