Function sse_client

src/mcp/client/sse.py:31–161

Client transport for SSE. `sse_read_timeout` determines how long (in seconds) the client will wait for a new event before disconnecting. All other HTTP operations are controlled by `timeout`. Args: url: The SSE endpoint URL. headers: Optional headers to include in reque

(
    url: str,
    headers: dict[str, Any] | None = None,
    timeout: float = 5.0,
    sse_read_timeout: float = 300.0,
    httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
    auth: httpx.Auth | None = None,
    on_session_created: Callable[[str], None] | None = None,
)

@asynccontextmanager
async def sse_client(
    url: str,
    headers: dict[str, Any] | None = None,
    timeout: float = 5.0,
    sse_read_timeout: float = 300.0,
    httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
    auth: httpx.Auth | None = None,
    on_session_created: Callable[[str], None] | None = None,
):
    """Client transport for SSE.

    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
    event before disconnecting. All other HTTP operations are controlled by `timeout`.

    Args:
        url: The SSE endpoint URL.
        headers: Optional headers to include in requests.
        timeout: HTTP timeout for regular operations (in seconds).
        sse_read_timeout: Timeout for SSE read operations (in seconds).
        httpx_client_factory: Factory function for creating the HTTPX client.
        auth: Optional HTTPX authentication handler.
        on_session_created: Optional callback invoked with the session ID when received.
    """
    logger.debug(f"Connecting to SSE endpoint: {remove_request_params(url)}")
    async with httpx_client_factory(
        headers=headers, auth=auth, timeout=httpx.Timeout(timeout, read=sse_read_timeout)
    ) as client:
        async with aconnect_sse(client, "GET", url) as event_source:
            event_source.response.raise_for_status()
            logger.debug("SSE connection established")

            read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
            write_stream, write_stream_reader = create_context_streams[SessionMessage](0)

            async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED):
                try:
                    async for sse in event_source.aiter_sse():  # pragma: no branch
                        logger.debug(f"Received SSE event: {sse.event}")
                        match sse.event:
                            case "endpoint":
                                endpoint_url = urljoin(url, sse.data)
                                logger.debug(f"Received endpoint URL: {endpoint_url}")

                                url_parsed = urlparse(url)
                                endpoint_parsed = urlparse(endpoint_url)
                                if (  # pragma: no cover
                                    url_parsed.netloc != endpoint_parsed.netloc
                                    or url_parsed.scheme != endpoint_parsed.scheme
                                ):
                                    error_msg = (  # pragma: no cover
                                        f"Endpoint origin does not match connection origin: {endpoint_url}"
                                    )
                                    logger.error(error_msg)  # pragma: no cover
                                    raise ValueError(error_msg)  # pragma: no cover

                                if on_session_created:
                                    session_id = _extract_session_id_from_endpoint(endpoint_url)
                                    if session_id: