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.test_helpers: Test agent helpers for easy examples and quick testing …
adcp.testing.upstream_recorder: Reference upstream-traffic recorder for query_upstream_traffic conformance …

Functions

def build_asgi_app(platform: DecisioningPlatform, *, 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, auth: BearerTokenAuth | None = None, asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None, context_factory: ContextFactory | None = None, middleware: Sequence[SkillMiddleware] | None = None, streaming_responses: bool = False, 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, **factory_kwargs: Any) ‑> Any

Expand source code

def build_asgi_app(
    platform: DecisioningPlatform,
    *,
    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,
    auth: BearerTokenAuth | None = None,
    asgi_middleware: Sequence[ASGIMiddlewareEntry] | None = None,
    context_factory: ContextFactory | None = None,
    middleware: Sequence[SkillMiddleware] | None = None,
    streaming_responses: bool = False,
    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,
    **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.

    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. ``asgi_middleware`` outermost — CORS, tenant resolution, custom
       auth, etc.

    :param platform: The :class:`DecisioningPlatform` instance under
        test.
    :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: Forwarded to
        :func:`create_adcp_server_from_platform`. Default ``False`` for
        test ergonomics — production :func:`serve` defaults to ``True``.
    :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 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 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.
    """
    from adcp.decisioning.serve import create_adcp_server_from_platform
    from adcp.server.serve import (
        _apply_asgi_middleware,
        _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__
    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,
        enable_dns_rebinding_protection=enable_dns_rebinding_protection,
        validation=validation,
        pre_validation_hooks=pre_validation_hooks,
        response_enhancer=response_enhancer,
    )
    # 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,
        )
    app = _wrap_with_size_limit(app, max_request_size)
    app = _apply_asgi_middleware(app, asgi_middleware)
    return app

Build 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 with httpx.ASGITransport, starlette.testclient.TestClient, or any ASGI test harness.

The wrapping order mirrors production (_run_mcp_http in serve()):

auth innermost — body-peeks the JSON-RPC payload before the path normalizer reshapes it (discovery bypass).
Path normalizer — strips trailing slashes so /mcp/ → /mcp without a 307 round-trip.
Discovery wrapper (only when discovery_base_url is provided) — serves /.well-known/adcp-agents.json.
Size cap (max_request_size; None → 10 MB default).
asgi_middleware outermost — CORS, tenant resolution, custom auth, etc.

:param platform: The :class:DecisioningPlatform instance under test. :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: Forwarded to :func:create_adcp_server_from_platform. Default False for test ergonomics — production :func:serve defaults to True. :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 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: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 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', name: str | None = None, advertise_all: bool = False, auto_emit_completion_webhooks: bool = False, follow_redirects: bool = True, headers: Mapping[str, str] | 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, 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, **factory_kwargs: Any) ‑> AsyncIterator[httpx.AsyncClient]

Expand source code

@asynccontextmanager
async def build_test_client(
    platform: DecisioningPlatform,
    *,
    base_url: str = "http://test",
    name: str | None = None,
    advertise_all: bool = False,
    auto_emit_completion_webhooks: bool = False,
    follow_redirects: bool = True,
    headers: Mapping[str, str] | 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,
    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,
    **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 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 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 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,
        name=name,
        advertise_all=advertise_all,
        auto_emit_completion_webhooks=auto_emit_completion_webhooks,
        allowed_hosts=[hostname],
        allowed_origins=allowed_origins,
        auth=auth,
        asgi_middleware=asgi_middleware,
        context_factory=context_factory,
        middleware=middleware,
        streaming_responses=streaming_responses,
        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,
        **factory_kwargs,
    )
    async with LifespanManager(app):
        async with _httpx.AsyncClient(
            transport=_httpx.ASGITransport(app=app),
            base_url=base_url,
            headers=headers,
            follow_redirects=follow_redirects,
        ) as client:
            yield client

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 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 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 factory_kwargs: Forwarded to :func:create_adcp_server_from_platform via :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"
)

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: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: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

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 = None

Typed container for a wire adcp_error dict.

Maps to the AdCP transport-errors spec §MCP Binding fields.

Instance variables

var code : str
var details : dict[str, typing.Any] | None
var field : str | None
var message : str
var recovery : str | None
var retry_after : int | None
var 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 item

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.

Instance variables

var content_type : str

prop 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 for endpoint_pattern matching.

var host : str

var method : Literal['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS']

var path : str

var payload : Any

var payload_length : int

var principal : str

var purpose : Literal['platform_primary', 'measurement', 'attribution', 'creative_serving', 'identity', 'other'] | None

var status_code : int | None

var timestamp : datetime.datetime

var 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 item

Project 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.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.

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.

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 — 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.

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.structuredContent 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.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: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.

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.

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.structuredContent 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.ok for success/failure and :attr:~ToolInvokeResult.adcp_error for 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 None

Result of a :meth:SellerTestClient.invoke() call.

Instance variables

var adcp_error : AdcpErrorPayload | None

var data : dict[str, typing.Any] | None

prop 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 None

True 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: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.

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_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.

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 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.

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 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).

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 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).

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 call

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.

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 coro

Await coro with principal bound 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.datetime
var total_count : int
var 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 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.