Function mounted_app

tests/interaction/_connect.py:171–221

Mount the server's streamable HTTP app on the in-process bridge and yield an httpx client. Yields the httpx client (rooted at the in-process origin) and the live session manager. Tests use this in two ways: for raw-httpx assertions (status codes, headers, SSE bytes) the test speaks HTTP

(
    server: Server | MCPServer,
    *,
    stateless_http: bool = False,
    json_response: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: TransportSecuritySettings | None = NO_DNS_REBINDING_PROTECTION,
    on_request: Callable[[httpx.Request], Awaitable[None]] | None = None,
    on_response: Callable[[httpx.Response], Awaitable[None]] | None = None,
    headers: dict[str, str] | None = None,
    auth: AuthSettings | None = None,
    token_verifier: TokenVerifier | None = None,
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
)

@asynccontextmanager
async def mounted_app(
    server: Server | MCPServer,
    *,
    stateless_http: bool = False,
    json_response: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: TransportSecuritySettings | None = NO_DNS_REBINDING_PROTECTION,
    on_request: Callable[[httpx.Request], Awaitable[None]] | None = None,
    on_response: Callable[[httpx.Response], Awaitable[None]] | None = None,
    headers: dict[str, str] | None = None,
    auth: AuthSettings | None = None,
    token_verifier: TokenVerifier | None = None,
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
) -> AsyncIterator[tuple[httpx.AsyncClient, StreamableHTTPSessionManager]]:
    """Mount the server's streamable HTTP app on the in-process bridge and yield an httpx client.

    Yields the httpx client (rooted at the in-process origin) and the live session manager. Tests
    use this in two ways: for raw-httpx assertions (status codes, headers, SSE bytes) the test
    speaks HTTP through the yielded client directly; for client-driven assertions the test wraps
    that client in `client_via_http(http)`, which lets several `Client`s share the one mounted
    session manager. `on_request` observes every outgoing HTTP request before it leaves the
    yielded client; `on_response` observes every HTTP response as its headers arrive (response
    bodies of SSE streams are not yet read at that point).

    DNS-rebinding protection is disabled by default; pass explicit settings (or `None` for the
    localhost auto-enable behaviour) to test the protection itself.
    """
    lowlevel = server._lowlevel_server if isinstance(server, MCPServer) else server
    app = lowlevel.streamable_http_app(
        stateless_http=stateless_http,
        json_response=json_response,
        event_store=event_store,
        retry_interval=retry_interval,
        transport_security=transport_security,
        auth=auth,
        token_verifier=token_verifier,
        auth_server_provider=auth_server_provider,
    )
    event_hooks: dict[str, list[Callable[..., Awaitable[None]]]] = {}
    if on_request is not None:
        event_hooks["request"] = [on_request]
    if on_response is not None:
        event_hooks["response"] = [on_response]
    async with (
        server.session_manager.run(),
        httpx.AsyncClient(
            transport=StreamingASGITransport(app), base_url=BASE_URL, event_hooks=event_hooks, headers=headers
        ) as http_client,
    ):
        yield http_client, server.session_manager


@asynccontextmanager