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)

Functions

def register_test_controller(mcp: Any,
store: TestControllerStore) ‑> None

Expand source code

def register_test_controller(mcp: Any, store: TestControllerStore) -> 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.

    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

    async def comply_test_controller(**kwargs: Any) -> str:
        result = await _handle_test_controller(store, kwargs)
        return json.dumps(result)

    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
    tool.parameters = {
        "type": "object",
        "properties": {
            "scenario": {
                "type": "string",
                "enum": [
                    "list_scenarios",
                    "force_creative_status",
                    "force_account_status",
                    "force_media_buy_status",
                    "force_session_status",
                    "simulate_delivery",
                    "simulate_budget_spend",
                ],
            },
            "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.

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

    async def force_creative_status(
        self, creative_id: str, status: str, rejection_reason: str | 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) -> 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
    ) -> 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
    ) -> dict[str, Any]:
        """Force a session to a given status.

        Returns:
            {"previous_state": str, "current_state": str}
        """
        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,
    ) -> 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,
    ) -> dict[str, Any]:
        """Simulate budget spend to a percentage.

        Returns:
            {"simulated": {...}}
        """
        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

async def force_account_status(self, account_id: str, status: str) ‑> dict[str, typing.Any]

Expand source code

async def force_account_status(self, account_id: str, status: str) -> 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_creative_status(self, creative_id: str, status: str, rejection_reason: str | None = None) ‑> dict[str, typing.Any]

Expand source code

async def force_creative_status(
    self, creative_id: str, status: str, rejection_reason: str | 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) ‑> dict[str, typing.Any]

Expand source code

async def force_media_buy_status(
    self, media_buy_id: str, status: str, rejection_reason: str | 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) ‑> dict[str, typing.Any]

Expand source code

async def force_session_status(
    self, session_id: str, status: str, termination_reason: str | 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 simulate_budget_spend(self,
spend_percentage: float,
account_id: str | None = None,
media_buy_id: str | None = None) ‑> dict[str, typing.Any]

Expand source code

async def simulate_budget_spend(
    self,
    spend_percentage: float,
    account_id: str | None = None,
    media_buy_id: str | 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) ‑> dict[str, typing.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,
) -> 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}