fix(server): fix streamable-http ASGI error, host binding, and AdCP 3.0.1 scenario gaps (#296)

* fix(server): fix streamable-http ASGI error, host binding, and AdCP 3.0.1 scenario gaps

Closes #295

serve() and create_mcp_server() now default stateless_http=True, which
sets FastMCP's stateless_http+json_response flags. The streaming-SSE
default caused "ASGI callable returned without completing response" and
made examples/seller_agent.py report overall_status: "unreachable" to
the AdCP storyboard runner. Expose host kwarg (ADCP_HOST env → 0.0.0.0
default) so container deployments (Fly.io, k8s, Cloud Run) bind the
external interface instead of loopback.

Add five AdCP 3.0.1 seed_* scenarios (seed_product, seed_pricing_option,
seed_creative, seed_plan, seed_media_buy) to SCENARIOS, TestControllerStore
stubs, _handle_test_controller dispatch, and both inputSchema locations.
Add account field to comply_test_controller schema so runner detection
works. Emit a WARNING log from capabilities_response() when compliance_
testing is declared without idempotency, surfacing the v2-mode downgrade.

https://claude.ai/code/session_01RJtJkQ8rooA7monEAB6ZVd

* refactor(server): rename stateless_http to streaming_responses

The FastMCP-internal flag name ``stateless_http`` reads like "disable
streamable-http" to a docstring-skimmer, but actually controls one
mode within the streamable-http transport (synchronous JSON response
vs. SSE-streamed response). Rename the SDK parameter to
``streaming_responses`` and flip the polarity so ``True`` means
"enable streaming" rather than "disable a feature".

Default is now ``False`` (one JSON response per request) which matches
what AdCP tools actually emit today — none of them produce progress
events. The FastMCP SSE-internal mode also has an upstream bug that
drops the ASGI response, blocking the storyboard runner; the new
default sidesteps that. Set ``streaming_responses=True`` if/when AdCP
adds genuinely streamed tools.

Internal call to ``mcp.settings.stateless_http = True`` is kept (that's
FastMCP's own setting name; we negate our parameter at the boundary).

Acting on user feedback that ``stateless_http=True`` was a confusing
name for the flag's user-visible behavior.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: bokelley

src/adcp/server/mcp_tools.py

@@ -839,6 +839,7 @@
         "inputSchema": {
             "type": "object",
             "properties": {
+                "account": {"type": "object"},
                 "scenario": {
                     "type": "string",
                     "enum": [
@@ -849,6 +850,11 @@
                         "force_session_status",
                         "simulate_delivery",
                         "simulate_budget_spend",
+                        "seed_product",
+                        "seed_pricing_option",
+                        "seed_creative",
+                        "seed_plan",
+                        "seed_media_buy",
                     ],
                 },
                 "params": {"type": "object"},

src/adcp/server/responses.py

@@ -22,11 +22,14 @@ async def get_products():
 
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timezone
 from typing import Any
 
 from adcp.server.helpers import valid_actions_for_status
 
+_logger = logging.getLogger("adcp.server")
+
 
 def _serialize(items: list[Any]) -> list[Any]:
     """Serialize a list of dicts or Pydantic models to plain dicts."""
@@ -98,6 +101,14 @@ def capabilities_response(
             idempotency=store.capability(),
         )
     """
+    if compliance_testing is not None and not idempotency:
+        _logger.warning(
+            "capabilities_response: adcp.idempotency not declared. "
+            "The AdCP 3.0.1 storyboard runner may downgrade to v2 mode and "
+            "cascade failures across idempotency-dependent tracks. "
+            "Pass idempotency={'supported': False} to declare non-support, "
+            "or idempotency=store.capability() to declare support."
+        )
     adcp_info: dict[str, Any] = {"major_versions": major_versions or [3]}
     if supported_versions is None and adcp_version is not None:
         supported_versions = [adcp_version]

src/adcp/server/serve.py

@@ -305,6 +305,7 @@ def serve(
     *,
     name: str = "adcp-agent",
     port: int | None = None,
+    host: str | None = None,
     transport: str = "streamable-http",
     instructions: str | None = None,
     test_controller: TestControllerStore | None = None,
@@ -315,6 +316,7 @@ def serve(
     message_parser: MessageParser | None = None,
     advertise_all: bool = False,
     max_request_size: int | None = None,
+    streaming_responses: bool = False,
 ) -> None:
     """Start an MCP or A2A server from an ADCP handler or server builder.
 
@@ -371,6 +373,23 @@ def serve(
             entirely (not recommended — the cap is the only guard
             against adversarial payloads exhausting Pydantic validation
             CPU/memory). See :mod:`adcp.server._size_limit`.
+        host: Network interface to bind to (MCP transports only). Defaults
+            to the ``ADCP_HOST`` environment variable, then ``"0.0.0.0"``
+            (all interfaces). Use ``"127.0.0.1"`` for local-only
+            development. Container deployments (Fly.io, k8s, Cloud Run)
+            require ``"0.0.0.0"`` so the process listens on the
+            container's external interface.
+        streaming_responses: When ``False`` (default), the streamable-http
+            transport returns one ``application/json`` response per
+            request. AdCP tools today don't emit progress events, and
+            FastMCP's SSE-internal streaming default has an upstream bug
+            that drops the ASGI response without completing — making the
+            storyboard runner report ``overall_status: "unreachable"``.
+            Set to ``True`` only if your tools genuinely emit progress
+            notifications and your clients consume the SSE stream
+            (MCP transports only). Note: the legacy ``transport="sse"``
+            is a separate (deprecated) MCP transport, unrelated to this
+            flag.
 
     Security:
         This function does NOT configure authentication. In production,
@@ -428,13 +447,15 @@ async def force_account_status(self, account_id, status):
             handler,
             name=name,
             port=port,
+            host=host,
             transport=transport,
             instructions=instructions,
             test_controller=test_controller,
             context_factory=context_factory,
             middleware=middleware,
             advertise_all=advertise_all,
             max_request_size=max_request_size,
+            streaming_responses=streaming_responses,
         )
     else:
         valid = ", ".join(sorted(("a2a", "streamable-http", "sse", "stdio")))
@@ -523,24 +544,28 @@ def _serve_mcp(
     *,
     name: str,
     port: int | None,
+    host: str | None = None,
     transport: str,
     instructions: str | None,
     test_controller: TestControllerStore | None,
     context_factory: ContextFactory | None = None,
     middleware: Sequence[SkillMiddleware] | None = None,
     advertise_all: bool = False,
     max_request_size: int | None = None,
+    streaming_responses: bool = False,
 ) -> None:
     """Start an MCP server."""
     mcp = create_mcp_server(
         handler,
         name=name,
         port=port,
+        host=host,
         instructions=instructions,
         include_test_controller=test_controller is not None,
         context_factory=context_factory,
         middleware=middleware,
         advertise_all=advertise_all,
+        streaming_responses=streaming_responses,
     )
 
     if test_controller is not None:
@@ -644,11 +669,13 @@ def create_mcp_server(
     *,
     name: str = "adcp-agent",
     port: int | None = None,
+    host: str | None = None,
     instructions: str | None = None,
     include_test_controller: bool = False,
     context_factory: ContextFactory | None = None,
     middleware: Sequence[SkillMiddleware] | None = None,
     advertise_all: bool = False,
+    streaming_responses: bool = False,
 ) -> Any:
     """Create a FastMCP server from an ADCP handler without starting it.
 
@@ -692,6 +719,17 @@ def create_mcp_server(
             :func:`~adcp.server.get_tools_for_handler` for semantics;
             use ``True`` for spec-compliance storyboards or when you
             deliberately want to expose a ``not_supported`` tool.
+        host: Network interface to bind to. Defaults to the ``ADCP_HOST``
+            environment variable, then ``"0.0.0.0"`` (all interfaces).
+            Use ``"127.0.0.1"`` for local-only development.
+        streaming_responses: When ``False`` (default), the streamable-http
+            transport returns one ``application/json`` response per
+            request — the right shape for AdCP tools today (none of which
+            emit progress events). The FastMCP SSE-internal streaming
+            default also has an upstream bug that drops the ASGI response
+            without completing, blocking the storyboard runner. Set to
+            ``True`` only if your tools genuinely emit progress
+            notifications and your clients consume the SSE stream.
 
     Returns:
         A configured FastMCP server instance. Call ``mcp.run()`` to start,
@@ -744,7 +782,15 @@ def create_mcp_server(
     from mcp.server.fastmcp import FastMCP
 
     resolved_port = port or int(os.environ.get("PORT", "3001"))
+    resolved_host = host if host is not None else (os.environ.get("ADCP_HOST") or "0.0.0.0")
     mcp = FastMCP(name, instructions=instructions, port=resolved_port)
+    mcp.settings.host = resolved_host
+    if not streaming_responses:
+        # FastMCP's SSE-internal default has an upstream bug; switching to
+        # stateless JSON-response mode is also semantically correct for
+        # AdCP tools, which return one complete envelope per request.
+        mcp.settings.stateless_http = True
+        mcp.settings.json_response = True
     _register_handler_tools(
         mcp,
         handler,

src/adcp/server/test_controller.py

@@ -52,6 +52,12 @@ async def force_account_status(self, account_id, status):
     "force_session_status",
     "simulate_delivery",
     "simulate_budget_spend",
+    # seed_* scenarios pre-populate storyboard fixtures (AdCP 3.0.1)
+    "seed_product",
+    "seed_pricing_option",
+    "seed_creative",
+    "seed_plan",
+    "seed_media_buy",
 ]
 
 
@@ -191,6 +197,77 @@ async def simulate_budget_spend(
         """
         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
+
 
 def _list_scenarios(store: TestControllerStore) -> list[str]:
     """Detect which scenarios a store actually implements.
@@ -353,6 +430,37 @@ async def _handle_test_controller(
                 media_buy_id=scenario_params.get("media_buy_id"),
                 **extra,
             )
+        elif scenario == "seed_product":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                product_id=scenario_params.get("product_id"),
+                **extra,
+            )
+        elif scenario == "seed_pricing_option":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                product_id=scenario_params.get("product_id"),
+                pricing_option_id=scenario_params.get("pricing_option_id"),
+                **extra,
+            )
+        elif scenario == "seed_creative":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                creative_id=scenario_params.get("creative_id"),
+                **extra,
+            )
+        elif scenario == "seed_plan":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                plan_id=scenario_params.get("plan_id"),
+                **extra,
+            )
+        elif scenario == "seed_media_buy":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                media_buy_id=scenario_params.get("media_buy_id"),
+                **extra,
+            )
         else:
             return _controller_error("UNKNOWN_SCENARIO", f"Unknown scenario: {scenario}")
     except TestControllerError as e:
@@ -442,6 +550,7 @@ async def comply_test_controller(**kwargs: Any) -> str:
     tool.parameters = {
         "type": "object",
         "properties": {
+            "account": {"type": "object"},
             "scenario": {
                 "type": "string",
                 "enum": [
@@ -452,6 +561,11 @@ async def comply_test_controller(**kwargs: Any) -> str:
                     "force_session_status",
                     "simulate_delivery",
                     "simulate_budget_spend",
+                    "seed_product",
+                    "seed_pricing_option",
+                    "seed_creative",
+                    "seed_plan",
+                    "seed_media_buy",
                 ],
             },
             "params": {"type": "object"},