Repository navigation
feat(testing): apps from build_asgi_app / serve() can start their lifespan only once, because the MCP session manager is created at build time #1311
Description
Summary
On 8.0.0, an app built by adcp.testing.build_asgi_app (or by serve()'s builders) can run its lifespan only once. The SDK creates a single ADCPStreamableHTTPSessionManager when it builds the streamable app (_create_adcp_mcp_session_manager, called at adcp/server/serve.py:2811-2828), and every lifespan start reuses it (lifespan=lambda app: manager.run(), serve.py:2846). mcp's StreamableHTTPSessionManager.run() is single-use by design: the second entry raises RuntimeError, and the docstring says "Create a new instance if you need to run again". So stopping and restarting the same app object fails on the second startup. Maintainers closed the mcp-side request to change this as not planned (modelcontextprotocol/python-sdk#1692). FastMCP 4 handles it by creating a fresh session manager inside each lifespan (fastmcp/server/http.py).
Reproduction (adcp 8.0.0, mcp 2.2.0)
import warnings
warnings.simplefilter("ignore")
from starlette.testclient import TestClient
from adcp.decisioning import DecisioningCapabilities, DecisioningPlatform, SingletonAccounts
from adcp.testing import build_asgi_app
from adcp.decisioning.capabilities import Account as CapabilitiesAccount
class Seller(DecisioningPlatform):
capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"], channels=["display"],
pricing_models=["cpm"],
account=CapabilitiesAccount(supported_billing=["operator"]))
accounts = SingletonAccounts(account_id="acct_1")
def get_products(self, req, ctx): return {"products": []}
def create_media_buy(self, req, ctx): raise NotImplementedError
def update_media_buy(self, media_buy_id, patch, ctx): raise NotImplementedError
def sync_creatives(self, req, ctx): raise NotImplementedError
def get_media_buy_delivery(self, req, ctx): raise NotImplementedError
def get_media_buys(self, req, ctx): raise NotImplementedError
app = build_asgi_app(Seller(), transport="mcp")
for attempt in (1, 2):
try:
with TestClient(app):
print(f"lifespan {attempt}: started")
except Exception as exc:
print(f"lifespan {attempt}:", type(exc).__name__, "|", exc)Output:
lifespan 1: started
lifespan 2: RuntimeError | StreamableHTTPSessionManager .run() can only be called once per instance. Create a new instance if you need to run again.
Why an adopter hits this
Test harnesses commonly enter an app's lifespan more than once, for example one TestClient(app) per test against a shared, expensive-to-build app. Embedded Sales Agent builds its app once per process and can't restart it. Its harness, tests/harness/_asgi_app.py (150 LOC), does three things: it keeps one app per process, runs that app's lifespan on a dedicated daemon-thread event loop, and marshals every request onto that loop with run_coroutine_threadsafe. The last part is needed because the manager's anyio task group is bound to the loop that called run(). Rebuilding the app per test would work, but would also rebuild everything the seller wires at build time.
Ask
Create the MCP session manager inside the app's lifespan rather than at build time, as FastMCP 4 does. Each lifespan start then gets a fresh instance, and the app can be started and stopped repeatedly. If build-time construction must stay, a documented build_asgi_app(..., reusable_lifespan=True) would also work.
Activity
bokelley commentedon Oct 2, 2026
Triage
Classification: Bug
Bucket(s): client (server build surface —
adcp/server/serve.py,adcp.testing)Status: ready-to-implement
What the experts said:
StreamableHTTPSessionManager.run()is single-use by upstream design; the fix is correct in principle but must use a mutable holder (_live) becauseADCPStreamableHTTPASGIApp.__call__closes over the build-timestreamable_httpreference;self._session_managermust also be assigned inside the lifespan (not at build time).reusable_lifespan=Trueopt-in — the testing module's job is to remove infrastructure plumbing from adopter code; the docstring's "drop-in compatible withTestClient" claim is a false promise until fixed; a regression test covering two consecutive lifespan cycles is required alongside the fix.My take: Both experts converge. The fix is a well-scoped, non-breaking closure refactor in one function plus one new regression test.
Implementation scope:
src/adcp/server/serve.py,streamable_http_app()(~line 2811–2847): Replace the build-time_create_adcp_mcp_session_manager(...)+lifespan=lambda app: manager.run()with an@asynccontextmanagerlifespan that creates a fresh manager andStreamableHTTPASGIAppon each start. Use a one-element mutable list_live: list[tuple[Any, Any]](holding(manager, streamable_http_instance)) shared between the lifespan andADCPStreamableHTTPASGIApp.__call__; deref via_live[0]in__call__. Setself._session_manager = managerinside the lifespan and clear on exit.tests/: Add a regression test that enters the app's lifespan twice on the same object (e.g. twowith TestClient(app):blocks, mirroring the issue's reproduction) and asserts noRuntimeError._session_manageris underscore-prefixed internal;streamable_http_app()return type (Starlette) unchanged.pytest tests/test_testing_decisioning.py,mypy src/adcp/server/serve.py,ruff check .Triaged by Claude Code. Session: https://claude.ai/code/session_011sjGbaoJjJiSmfEuGPAFd6
Generated by Claude Code
bokelley commentedon Oct 3, 2026
Fixed by #1333 (squash a6e7673): serve() binds a fresh MCP session manager and endpoints for each lifespan, so the same app can complete repeated startup/request/shutdown cycles.