Function in_process_client_factory

tests/shared/test_sse.py:52–72

An httpx_client_factory for sse_client whose clients are served in process by `app`.

(app: Starlette)

def in_process_client_factory(app: Starlette) -> McpHttpClientFactory:
    """An httpx_client_factory for sse_client whose clients are served in process by `app`."""

    def factory(
        headers: dict[str, str] | None = None,
        timeout: httpx.Timeout | None = None,
        auth: httpx.Auth | None = None,
    ) -> httpx.AsyncClient:
        # The SSE GET runs until it observes a disconnect, so the bridge must let the
        # application drain on close rather than cancelling it. follow_redirects matches
        # create_mcp_http_client, the factory this one stands in for.
        return httpx.AsyncClient(
            transport=StreamingASGITransport(app, cancel_on_close=False),
            base_url=BASE_URL,
            headers=headers,
            timeout=timeout,
            auth=auth,
            follow_redirects=True,
        )

    return factory


async def _handle_read_resource(ctx: ServerRequestContext, params: ReadResourceRequestParams) -> ReadResourceResult: