Class SseServerTransport

src/mcp/server/sse.py:64–262

SSE server transport for MCP. This class provides two ASGI applications, suitable for use with a framework like Starlette and a server like Hypercorn: 1. connect_sse() is an ASGI application which receives incoming GET requests, and sets up a new SSE stream to send server mes

class SseServerTransport:
    """SSE server transport for MCP. This class provides two ASGI applications,
    suitable for use with a framework like Starlette and a server like Hypercorn:

        1. connect_sse() is an ASGI application which receives incoming GET requests,
           and sets up a new SSE stream to send server messages to the client.
        2. handle_post_message() is an ASGI application which receives incoming POST
           requests, which should contain client messages that link to a
           previously-established SSE session.
    """

    _endpoint: str
    _read_stream_writers: dict[UUID, ContextSendStream[SessionMessage | Exception]]
    # Identity of the credential that created each session; requests for a
    # session must present the same credential.
    _session_owners: dict[UUID, AuthorizationContext]
    _security: TransportSecurityMiddleware

    def __init__(self, endpoint: str, security_settings: TransportSecuritySettings | None = None) -> None:
        """Creates a new SSE server transport, which will direct the client to POST
        messages to the relative path given.

        Args:
            endpoint: A relative path where messages should be posted
                    (e.g., "/messages/").
            security_settings: Optional security settings for DNS rebinding protection.

        Note:
            We use relative paths instead of full URLs for several reasons:
            1. Security: Prevents cross-origin requests by ensuring clients only connect
               to the same origin they established the SSE connection with
            2. Flexibility: The server can be mounted at any path without needing to
               know its full URL
            3. Portability: The same endpoint configuration works across different
               environments (development, staging, production)

        Raises:
            ValueError: If the endpoint is a full URL instead of a relative path
        """

        super().__init__()

        # Validate that endpoint is a relative path and not a full URL
        if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
            raise ValueError(
                f"Given endpoint: {endpoint} is not a relative path (e.g., '/messages/'), "
                "expecting a relative path (e.g., '/messages/')."
            )

        # Ensure endpoint starts with a forward slash
        if not endpoint.startswith("/"):
            endpoint = "/" + endpoint

        self._endpoint = endpoint
        self._read_stream_writers = {}
        self._session_owners = {}
        self._security = TransportSecurityMiddleware(security_settings)
        logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")