Module adcp.server.test_controller

Built-in comply_test_controller for ADCP servers.

Provides TestControllerStore and register_test_controller() so that storyboard tests can manipulate server state (force status transitions, simulate delivery, etc.) without agents needing to implement the comply_test_controller tool by hand.

Usage

from adcp.server import serve, ADCPHandler from adcp.server.test_controller import TestControllerStore, register_test_controller

class MyStore(TestControllerStore): async def force_account_status(self, account_id, status): old = self.accounts[account_id]["status"] self.accounts[account_id]["status"] = status return {"previous_state": old, "current_state": status}

store = MyStore() serve(MySeller(), name="my-agent", test_controller=store)

Header-driven compatibility: Store methods MAY accept a keyword-only context: ToolContext | None parameter. When the server was configured with a context_factory, the dispatcher calls the factory per request and threads the resulting ToolContext into the store method. This lets sellers whose test runtime reads request headers (e.g. AdCPTestContext.from_headers(request.headers)) compose the storyboard-driven comply_test_controller skill with their existing header-driven mock state — populate the test context in the context_factory (from a ContextVar set by your HTTP middleware) and read it off context.metadata inside the store. Stores that don't declare context on a method keep working unchanged — the dispatcher only passes context to methods whose signature accepts it.

Functions

def register_test_controller(mcp: Any,
store: TestControllerStore,
*,
context_factory: ContextFactory | None = None,
account_resolver: _AccountResolver | None = None) ‑> None

Expand source code

def register_test_controller(
    mcp: Any,
    store: TestControllerStore,
    *,
    context_factory: ContextFactory | None = None,
    account_resolver: _AccountResolver | None = None,
) -> None:
    """Register the comply_test_controller tool on an MCP server.

    This is the Python equivalent of the JS SDK's registerTestController().
    It adds the comply_test_controller MCP tool backed by your TestControllerStore.

    Args:
        mcp: A FastMCP server instance.
        store: Your TestControllerStore implementation.
        context_factory: Optional ``ContextFactory`` invoked per call to
            build a :class:`ToolContext`. When set, the context is
            threaded into store methods that declare a ``context``
            keyword — which is how sellers whose test runtime reads
            request headers (``AdCPTestContext.from_headers``) combine
            header-driven mock state with the storyboard-driven
            ``comply_test_controller`` skill. Wire the same factory you
            pass to :func:`create_mcp_server` so both paths see the
            same per-request context.
        account_resolver: Async-or-sync callable that resolves a wire
            account ref to a framework :class:`Account`, OR the
            :data:`INSECURE_ALLOW_ALL` sentinel for tests that opt out
            of the gate. The comply controller applies the Phase 1
            sandbox-authority gate against the resolved account: only
            accounts with ``mode in {'sandbox', 'mock'}`` (or legacy
            ``sandbox=True``) are admitted; ``mode='live'`` is denied
            regardless of wire signals. v6 :class:`DecisioningPlatform`
            adopters get this hooked automatically by
            ``decisioning.serve``. Adopters wiring the controller
            manually pass a closure over their own account store.

            **Default fail-closed.** When ``None`` AND ``ADCP_SANDBOX``
            is unset, every comply call is denied — manually-wired
            ``ADCPHandler`` / :class:`ComplianceHandler` deployments are
            protected by default. Tests that intentionally bypass the
            gate pass ``account_resolver=INSECURE_ALLOW_ALL``; dev
            servers can set ``ADCP_SANDBOX=1`` instead.

            See ``docs/proposals/lifecycle-state-and-sandbox-authority.md``.

    Example:
        from adcp.server.test_controller import TestControllerStore, register_test_controller

        class MyStore(TestControllerStore):
            async def force_account_status(self, account_id, status):
                old = self.accounts[account_id]["status"]
                self.accounts[account_id]["status"] = status
                return {"previous_state": old, "current_state": status}

        mcp = create_mcp_server(MySeller(), name="my-agent")
        register_test_controller(mcp, MyStore())
        mcp.run(transport="streamable-http")
    """

    from mcp.server.fastmcp.tools import Tool
    from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata
    from pydantic import ConfigDict

    from adcp.server.base import ToolContext as _ToolContext
    from adcp.server.serve import RequestMetadata as _RequestMetadata

    async def comply_test_controller(**kwargs: Any) -> dict[str, Any]:
        context: _ToolContext | None = None
        if context_factory is not None:
            meta = _RequestMetadata(tool_name="comply_test_controller", transport="mcp")
            context = context_factory(meta)
            if not isinstance(context, _ToolContext):
                raise TypeError(
                    "context_factory for comply_test_controller returned "
                    f"{type(context).__name__}, not a ToolContext instance"
                )
        return await _handle_test_controller(
            store,
            kwargs,
            context=context,
            account_resolver=account_resolver,
        )

    tool = Tool.from_function(
        comply_test_controller,
        name="comply_test_controller",
        description="Compliance test controller. Sandbox only, not for production use.",
    )

    # Override schema with the proper comply_test_controller inputSchema.
    # Derived from SCENARIOS so it can't drift from the dispatcher.
    tool.parameters = {
        "type": "object",
        "properties": {
            "account": {"type": "object"},
            "scenario": {
                "type": "string",
                # Derived from SCENARIOS so the enum never drifts from the dispatcher.
                "enum": ["list_scenarios"] + SCENARIOS,
            },
            "params": {"type": "object"},
            "context": {"type": "object"},
        },
        "required": ["scenario"],
    }

    # Override fn_metadata with a permissive model
    class _ControllerArgs(ArgModelBase):
        model_config = ConfigDict(extra="allow")

        def model_dump_one_level(self) -> dict[str, Any]:
            result: dict[str, Any] = {}
            for field_name in self.__class__.model_fields:
                result[field_name] = getattr(self, field_name)
            if self.model_extra:
                result.update(self.model_extra)
            return result

    tool.fn_metadata = FuncMetadata(
        arg_model=_ControllerArgs,
        output_schema=tool.fn_metadata.output_schema,
        output_model=tool.fn_metadata.output_model,
        wrap_output=tool.fn_metadata.wrap_output,
    )

    mcp._tool_manager._tools["comply_test_controller"] = tool

Register the comply_test_controller tool on an MCP server.

This is the Python equivalent of the JS SDK's registerTestController(). It adds the comply_test_controller MCP tool backed by your TestControllerStore.

Args

mcp

A FastMCP server instance.

store

Your TestControllerStore implementation.

context_factory

Optional ContextFactory invoked per call to build a :class:ToolContext. When set, the context is threaded into store methods that declare a context keyword — which is how sellers whose test runtime reads request headers (AdCPTestContext.from_headers) combine header-driven mock state with the storyboard-driven comply_test_controller skill. Wire the same factory you pass to :func:create_mcp_server so both paths see the same per-request context.

account_resolver

Async-or-sync callable that resolves a wire account ref to a framework :class:Account, OR the :data:INSECURE_ALLOW_ALL sentinel for tests that opt out of the gate. The comply controller applies the Phase 1 sandbox-authority gate against the resolved account: only accounts with mode in {'sandbox', 'mock'} (or legacy sandbox=True) are admitted; mode='live' is denied regardless of wire signals. v6 :class:DecisioningPlatform adopters get this hooked automatically by decisioning.serve. Adopters wiring the controller manually pass a closure over their own account store.

Default fail-closed. When None AND ADCP_SANDBOX is unset, every comply call is denied — manually-wired ADCPHandler / :class:ComplianceHandler deployments are protected by default. Tests that intentionally bypass the gate pass account_resolver=INSECURE_ALLOW_ALL; dev servers can set ADCP_SANDBOX=1 instead.

See docs/proposals/lifecycle-state-and-sandbox-authority.md.

Example

from adcp.server.test_controller import TestControllerStore, register_test_controller

class MyStore(TestControllerStore): async def force_account_status(self, account_id, status): old = self.accounts[account_id]["status"] self.accounts[account_id]["status"] = status return {"previous_state": old, "current_state": status}

mcp = create_mcp_server(MySeller(), name="my-agent") register_test_controller(mcp, MyStore()) mcp.run(transport="streamable-http")

Classes

class TestControllerError (code: str, message: str, current_state: str | None = None)

Expand source code

class TestControllerError(Exception):
    """Typed error for test controller store methods.

    Raise this from your TestControllerStore methods to return structured
    error responses. The dispatcher catches it and converts to the AdCP
    comply_test_controller error format.

    Example:
        async def force_media_buy_status(self, media_buy_id, status, rejection_reason=None):
            prev = self.media_buys.get(media_buy_id)
            if prev is None:
                raise TestControllerError("NOT_FOUND", f"Media buy {media_buy_id} not found")
            if prev in ("completed", "rejected", "canceled"):
                raise TestControllerError(
                    "INVALID_TRANSITION",
                    f"Cannot transition from {prev}",
                    current_state=prev,
                )
            self.media_buys[media_buy_id] = status
            return {"previous_state": prev, "current_state": status}
    """

    def __init__(self, code: str, message: str, current_state: str | None = None):
        super().__init__(message)
        self.code = code
        self.current_state = current_state

Typed error for test controller store methods.

Raise this from your TestControllerStore methods to return structured error responses. The dispatcher catches it and converts to the AdCP comply_test_controller error format.

Example

async def force_media_buy_status(self, media_buy_id, status, rejection_reason=None): prev = self.media_buys.get(media_buy_id) if prev is None: raise TestControllerError("NOT_FOUND", f"Media buy {media_buy_id} not found") if prev in ("completed", "rejected", "canceled"): raise TestControllerError( "INVALID_TRANSITION", f"Cannot transition from {prev}", current_state=prev, ) self.media_buys[media_buy_id] = status return {"previous_state": prev, "current_state": status}

Ancestors

• builtins.Exception
• builtins.BaseException

class TestControllerStore

Expand source code

class TestControllerStore:
    """Base class for test controller state management.

    Subclass this and override the methods for scenarios your agent supports.
    Methods you don't override will be reported as unsupported scenarios
    and excluded from list_scenarios.

    Raise TestControllerError for structured error responses.

    Methods MAY declare an optional keyword-only ``context: ToolContext |
    None = None`` parameter. When present, the dispatcher threads the
    ``ToolContext`` built by the server's ``context_factory`` into the
    call — header-driven mock state (e.g. ``AdCPTestContext.from_headers``)
    populated in the factory is readable off ``context.metadata``.
    Stores that don't declare ``context`` keep working unchanged.
    """

    async def force_creative_status(
        self,
        creative_id: str,
        status: str,
        rejection_reason: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Force a creative to a given status.

        Returns:
            {"previous_state": str, "current_state": str}
        """
        raise NotImplementedError

    async def force_account_status(
        self,
        account_id: str,
        status: str,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Force an account to a given status.

        Returns:
            {"previous_state": str, "current_state": str}
        """
        raise NotImplementedError

    async def force_media_buy_status(
        self,
        media_buy_id: str,
        status: str,
        rejection_reason: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Force a media buy to a given status.

        Returns:
            {"previous_state": str, "current_state": str}
        """
        raise NotImplementedError

    async def force_session_status(
        self,
        session_id: str,
        status: str,
        termination_reason: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Force a session to a given status.

        Returns:
            {"previous_state": str, "current_state": str}
        """
        raise NotImplementedError

    async def force_create_media_buy_arm(
        self,
        arm: str,
        task_id: str | None = None,
        message: str | None = None,
        *,
        account: dict[str, Any] | None = None,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Register a single-shot directive for the next create_media_buy call.

        The directive is consumed by the next create_media_buy call from the
        same authenticated sandbox account, then cleared. A second registration
        before consumption overwrites the first.

        Args:
            arm: Response arm — ``'submitted'`` or ``'input-required'``.
            task_id: Required when ``arm='submitted'``. The seller MUST emit
                this exact value on the next create_media_buy task envelope
                and accept it on subsequent tasks/get calls within the same
                sandbox account. Max 128 chars.
            message: Optional plain-text note surfaced on the response.
                Max 2000 chars.
            account: Caller-supplied account object from the MCP request.
                Implementations use this for single-shot-per-account isolation.
            context: Optional ToolContext from the server's context_factory.

        Returns:
            ForcedDirectiveSuccess::

                {"success": True, "forced": {"arm": str, "task_id"?: str}}

        Raises:
            TestControllerError: with code ``"NOT_FOUND"`` if the caller
                account is not recognized, or ``"INVALID_PARAMS"`` on
                validation failure.
        """
        raise NotImplementedError

    async def force_task_completion(
        self,
        task_id: str,
        result: dict[str, Any],
        *,
        account: dict[str, Any] | None = None,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Resolve a previously-submitted task to ``'completed'``.

        Isolation and idempotency contract:

        - **Cross-account replay** — raise ``TestControllerError("NOT_FOUND", ...)``
          when the task_id was registered by a different sandbox account.
        - **Identical-params replay** — idempotent; return the same
          ``StateTransitionSuccess``.
        - **Diverging-params replay** against a terminal task — raise
          ``TestControllerError("INVALID_TRANSITION", ...,
          current_state="completed")``.

        Args:
            task_id: Task handle to resolve. Max 128 chars.
            result: Completion payload (non-empty object). Implementations
                SHOULD validate it against the response branch for the task's
                original method and MUST reject payloads that fail that check
                with ``TestControllerError("INVALID_PARAMS", ...)``.
            account: Caller-supplied account object from the MCP request.
                Used for cross-account isolation.
            context: Optional ToolContext from the server's context_factory.

        Returns:
            StateTransitionSuccess::

                {"success": True, "previous_state": "submitted",
                 "current_state": "completed"}

        Raises:
            TestControllerError: with code ``"NOT_FOUND"`` if the task_id
                is unknown or owned by a different account,
                ``"INVALID_TRANSITION"`` if the task is already terminal and
                params diverge, or ``"INVALID_PARAMS"`` on validation failure.
        """
        raise NotImplementedError

    async def simulate_delivery(
        self,
        media_buy_id: str,
        impressions: int | None = None,
        clicks: int | None = None,
        conversions: int | None = None,
        reported_spend: dict[str, Any] | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Simulate delivery metrics for a media buy.

        Returns:
            {"simulated": {...}, "cumulative": {...} | None}
        """
        raise NotImplementedError

    async def simulate_budget_spend(
        self,
        spend_percentage: float,
        account_id: str | None = None,
        media_buy_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Simulate budget spend to a percentage.

        Returns:
            {"simulated": {...}}
        """
        raise NotImplementedError

    async def seed_product(
        self,
        fixture: dict[str, Any] | None = None,
        product_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a product fixture for storyboard tests (AdCP 3.0.1).

        Returns:
            {"product_id": str}
        """
        raise NotImplementedError

    async def seed_pricing_option(
        self,
        fixture: dict[str, Any] | None = None,
        product_id: str | None = None,
        pricing_option_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a pricing option fixture for storyboard tests (AdCP 3.0.1).

        Returns:
            {"pricing_option_id": str}
        """
        raise NotImplementedError

    async def seed_creative(
        self,
        fixture: dict[str, Any] | None = None,
        creative_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a creative fixture for storyboard tests (AdCP 3.0.1).

        Returns:
            {"creative_id": str}
        """
        raise NotImplementedError

    async def seed_plan(
        self,
        fixture: dict[str, Any] | None = None,
        plan_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a plan fixture for storyboard tests (AdCP 3.0.1).

        Returns:
            {"plan_id": str}
        """
        raise NotImplementedError

    async def seed_media_buy(
        self,
        fixture: dict[str, Any] | None = None,
        media_buy_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a media buy fixture for storyboard tests (AdCP 3.0.1).

        Returns:
            {"media_buy_id": str}
        """
        raise NotImplementedError

    async def seed_creative_format(
        self,
        fixture: dict[str, Any] | None = None,
        format_id: str | None = None,
        *,
        context: ToolContext | None = None,
    ) -> dict[str, Any]:
        """Pre-populate a creative format fixture for storyboard tests (AdCP 3.0.1).

        The seller MUST expose the seeded format_id in list_creative_formats
        responses for the duration of the compliance session.

        Returns:
            {"format_id": str}
        """
        raise NotImplementedError

Base class for test controller state management.

Subclass this and override the methods for scenarios your agent supports. Methods you don't override will be reported as unsupported scenarios and excluded from list_scenarios.

Raise TestControllerError for structured error responses.

Methods MAY declare an optional keyword-only context: ToolContext | None = None parameter. When present, the dispatcher threads the ToolContext built by the server's context_factory into the call — header-driven mock state (e.g. AdCPTestContext.from_headers) populated in the factory is readable off context.metadata. Stores that don't declare context keep working unchanged.

Methods

async def force_account_status(self, account_id: str, status: str, *, context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_account_status(
    self,
    account_id: str,
    status: str,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Force an account to a given status.

    Returns:
        {"previous_state": str, "current_state": str}
    """
    raise NotImplementedError

Force an account to a given status.

Returns

{"previous_state": str, "current_state": str}

async def force_create_media_buy_arm(self,
arm: str,
task_id: str | None = None,
message: str | None = None,
*,
account: dict[str, Any] | None = None,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_create_media_buy_arm(
    self,
    arm: str,
    task_id: str | None = None,
    message: str | None = None,
    *,
    account: dict[str, Any] | None = None,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Register a single-shot directive for the next create_media_buy call.

    The directive is consumed by the next create_media_buy call from the
    same authenticated sandbox account, then cleared. A second registration
    before consumption overwrites the first.

    Args:
        arm: Response arm — ``'submitted'`` or ``'input-required'``.
        task_id: Required when ``arm='submitted'``. The seller MUST emit
            this exact value on the next create_media_buy task envelope
            and accept it on subsequent tasks/get calls within the same
            sandbox account. Max 128 chars.
        message: Optional plain-text note surfaced on the response.
            Max 2000 chars.
        account: Caller-supplied account object from the MCP request.
            Implementations use this for single-shot-per-account isolation.
        context: Optional ToolContext from the server's context_factory.

    Returns:
        ForcedDirectiveSuccess::

            {"success": True, "forced": {"arm": str, "task_id"?: str}}

    Raises:
        TestControllerError: with code ``"NOT_FOUND"`` if the caller
            account is not recognized, or ``"INVALID_PARAMS"`` on
            validation failure.
    """
    raise NotImplementedError

Register a single-shot directive for the next create_media_buy call.

The directive is consumed by the next create_media_buy call from the same authenticated sandbox account, then cleared. A second registration before consumption overwrites the first.

Args

arm

Response arm — 'submitted' or 'input-required'.

task_id

Required when arm='submitted'. The seller MUST emit this exact value on the next create_media_buy task envelope and accept it on subsequent tasks/get calls within the same sandbox account. Max 128 chars.

message

Optional plain-text note surfaced on the response. Max 2000 chars.

account

Caller-supplied account object from the MCP request. Implementations use this for single-shot-per-account isolation.

context

Optional ToolContext from the server's context_factory.

Returns

ForcedDirectiveSuccess::

{"success": True, "forced": {"arm": str, "task_id"?: str}}

Raises

TestControllerError

with code "NOT_FOUND" if the caller account is not recognized, or "INVALID_PARAMS" on validation failure.

async def force_creative_status(self,
creative_id: str,
status: str,
rejection_reason: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_creative_status(
    self,
    creative_id: str,
    status: str,
    rejection_reason: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Force a creative to a given status.

    Returns:
        {"previous_state": str, "current_state": str}
    """
    raise NotImplementedError

Force a creative to a given status.

Returns

{"previous_state": str, "current_state": str}

async def force_media_buy_status(self,
media_buy_id: str,
status: str,
rejection_reason: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_media_buy_status(
    self,
    media_buy_id: str,
    status: str,
    rejection_reason: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Force a media buy to a given status.

    Returns:
        {"previous_state": str, "current_state": str}
    """
    raise NotImplementedError

Force a media buy to a given status.

Returns

{"previous_state": str, "current_state": str}

async def force_session_status(self,
session_id: str,
status: str,
termination_reason: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_session_status(
    self,
    session_id: str,
    status: str,
    termination_reason: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Force a session to a given status.

    Returns:
        {"previous_state": str, "current_state": str}
    """
    raise NotImplementedError

Force a session to a given status.

Returns

{"previous_state": str, "current_state": str}

async def force_task_completion(self,
task_id: str,
result: dict[str, Any],
*,
account: dict[str, Any] | None = None,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def force_task_completion(
    self,
    task_id: str,
    result: dict[str, Any],
    *,
    account: dict[str, Any] | None = None,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Resolve a previously-submitted task to ``'completed'``.

    Isolation and idempotency contract:

    - **Cross-account replay** — raise ``TestControllerError("NOT_FOUND", ...)``
      when the task_id was registered by a different sandbox account.
    - **Identical-params replay** — idempotent; return the same
      ``StateTransitionSuccess``.
    - **Diverging-params replay** against a terminal task — raise
      ``TestControllerError("INVALID_TRANSITION", ...,
      current_state="completed")``.

    Args:
        task_id: Task handle to resolve. Max 128 chars.
        result: Completion payload (non-empty object). Implementations
            SHOULD validate it against the response branch for the task's
            original method and MUST reject payloads that fail that check
            with ``TestControllerError("INVALID_PARAMS", ...)``.
        account: Caller-supplied account object from the MCP request.
            Used for cross-account isolation.
        context: Optional ToolContext from the server's context_factory.

    Returns:
        StateTransitionSuccess::

            {"success": True, "previous_state": "submitted",
             "current_state": "completed"}

    Raises:
        TestControllerError: with code ``"NOT_FOUND"`` if the task_id
            is unknown or owned by a different account,
            ``"INVALID_TRANSITION"`` if the task is already terminal and
            params diverge, or ``"INVALID_PARAMS"`` on validation failure.
    """
    raise NotImplementedError

Resolve a previously-submitted task to 'completed'.

Isolation and idempotency contract:

• Cross-account replay — raise TestControllerError("NOT_FOUND", ...) when the task_id was registered by a different sandbox account.
• Identical-params replay — idempotent; return the same StateTransitionSuccess.
• Diverging-params replay against a terminal task — raise TestControllerError("INVALID_TRANSITION", ..., current_state="completed").

Args

task_id

Task handle to resolve. Max 128 chars.

result

Completion payload (non-empty object). Implementations SHOULD validate it against the response branch for the task's original method and MUST reject payloads that fail that check with TestControllerError("INVALID_PARAMS", ...).

account

Caller-supplied account object from the MCP request. Used for cross-account isolation.

context

Optional ToolContext from the server's context_factory.

Returns

StateTransitionSuccess::

{"success": True, "previous_state": "submitted",
 "current_state": "completed"}

Raises

TestControllerError

with code "NOT_FOUND" if the task_id is unknown or owned by a different account, "INVALID_TRANSITION" if the task is already terminal and params diverge, or "INVALID_PARAMS" on validation failure.

async def seed_creative(self,
fixture: dict[str, Any] | None = None,
creative_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_creative(
    self,
    fixture: dict[str, Any] | None = None,
    creative_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a creative fixture for storyboard tests (AdCP 3.0.1).

    Returns:
        {"creative_id": str}
    """
    raise NotImplementedError

Pre-populate a creative fixture for storyboard tests (AdCP 3.0.1).

Returns

{"creative_id": str}

async def seed_creative_format(self,
fixture: dict[str, Any] | None = None,
format_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_creative_format(
    self,
    fixture: dict[str, Any] | None = None,
    format_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a creative format fixture for storyboard tests (AdCP 3.0.1).

    The seller MUST expose the seeded format_id in list_creative_formats
    responses for the duration of the compliance session.

    Returns:
        {"format_id": str}
    """
    raise NotImplementedError

Pre-populate a creative format fixture for storyboard tests (AdCP 3.0.1).

The seller MUST expose the seeded format_id in list_creative_formats responses for the duration of the compliance session.

Returns

{"format_id": str}

async def seed_media_buy(self,
fixture: dict[str, Any] | None = None,
media_buy_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_media_buy(
    self,
    fixture: dict[str, Any] | None = None,
    media_buy_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a media buy fixture for storyboard tests (AdCP 3.0.1).

    Returns:
        {"media_buy_id": str}
    """
    raise NotImplementedError

Pre-populate a media buy fixture for storyboard tests (AdCP 3.0.1).

Returns

{"media_buy_id": str}

async def seed_plan(self,
fixture: dict[str, Any] | None = None,
plan_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_plan(
    self,
    fixture: dict[str, Any] | None = None,
    plan_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a plan fixture for storyboard tests (AdCP 3.0.1).

    Returns:
        {"plan_id": str}
    """
    raise NotImplementedError

Pre-populate a plan fixture for storyboard tests (AdCP 3.0.1).

Returns

{"plan_id": str}

async def seed_pricing_option(self,
fixture: dict[str, Any] | None = None,
product_id: str | None = None,
pricing_option_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_pricing_option(
    self,
    fixture: dict[str, Any] | None = None,
    product_id: str | None = None,
    pricing_option_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a pricing option fixture for storyboard tests (AdCP 3.0.1).

    Returns:
        {"pricing_option_id": str}
    """
    raise NotImplementedError

Pre-populate a pricing option fixture for storyboard tests (AdCP 3.0.1).

Returns

{"pricing_option_id": str}

async def seed_product(self,
fixture: dict[str, Any] | None = None,
product_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def seed_product(
    self,
    fixture: dict[str, Any] | None = None,
    product_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Pre-populate a product fixture for storyboard tests (AdCP 3.0.1).

    Returns:
        {"product_id": str}
    """
    raise NotImplementedError

Pre-populate a product fixture for storyboard tests (AdCP 3.0.1).

Returns

{"product_id": str}

async def simulate_budget_spend(self,
spend_percentage: float,
account_id: str | None = None,
media_buy_id: str | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def simulate_budget_spend(
    self,
    spend_percentage: float,
    account_id: str | None = None,
    media_buy_id: str | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Simulate budget spend to a percentage.

    Returns:
        {"simulated": {...}}
    """
    raise NotImplementedError

Simulate budget spend to a percentage.

Returns

{"simulated": {…}}

async def simulate_delivery(self,
media_buy_id: str,
impressions: int | None = None,
clicks: int | None = None,
conversions: int | None = None,
reported_spend: dict[str, Any] | None = None,
*,
context: ToolContext | None = None) ‑> dict[str, Any]

Expand source code

async def simulate_delivery(
    self,
    media_buy_id: str,
    impressions: int | None = None,
    clicks: int | None = None,
    conversions: int | None = None,
    reported_spend: dict[str, Any] | None = None,
    *,
    context: ToolContext | None = None,
) -> dict[str, Any]:
    """Simulate delivery metrics for a media buy.

    Returns:
        {"simulated": {...}, "cumulative": {...} | None}
    """
    raise NotImplementedError

Simulate delivery metrics for a media buy.

Returns

{"simulated": {…}, "cumulative": {…} | None}