Copilot requires absolute URL for messages endpoint

[lancegliser]

The current MCP documentation has some known issues.

Currently, the endpoint returned in the Open SSE connection call must be a full URI. | Not applicable

There is a message_path parameter in ServerSettings, but using it to provide an absolute error throws an error:

                  ^^^^^^^^^^^^^^^^^^^^
venv/lib/python3.12/site-packages/fastmcp/server/server.py", line 762, in sse_app
    Mount(self.settings.message_path, app=sse.handle_post_message),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.12/site-packages/starlette/routing.py", line 381, in __init__
    assert path == "" or path.startswith("/"), "Routed paths must start with '/'"
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: Routed paths must start with '/'

Seems totally reasonable as we're using it for routing. Appears we need a second parameter to allow SseServerTransport.connect_see's session_uri with a prefix.

[lancegliser]

Not directly related, but if someone comes here looking for Copilot solutions, you'll also need to patch the headers sent by the server. You can do so without changes to MCP.

Make some Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request


class RemoveSseCharsetMiddleware(BaseHTTPMiddleware):
    """
    Middleware for SSE.

    Due to failed Microsoft Copilot implementations we must remove `; charset=utf-8`
    """

    async def dispatch(self, request: Request, call_next):
        """Remove '; charset=utf-8' from content-type headers."""
        response = await call_next(request)
        if "content-type" in response.headers:
            content_type = response.headers["content-type"]
            if "text/event-stream" in content_type and "; charset=utf-8" in content_type:
                response.headers["content-type"] = content_type.replace("; charset=utf-8", "")
        return response

Use it when mounting your sse_app:

    app.router.routes.append(
        Mount(
            "/",
            app=mcp_sse_app,
            middleware=[
                Middleware(McpApiKeyMiddleware),
                Middleware(RemoveSseCharsetMiddleware),
            ],
        )
    )

[lancegliser]

Digging in, SseServerTransport is in https://github.com/modelcontextprotocol/python-sdk. I'll move this conversation there.

[lancegliser]

This thread provides a couple details that will help:

• The standard has been updated. It's going to eventually be merged into a single endpoint that does both transactional messages and streaming.
• @dwayn provided some middleware to ensure mounted apps take mounting into effect. It's at https://github.com/dwayn/fastmcp-mount/tree/master.

That however doesn't take into account the need for full base url in messages. Adapting his work, I just added base_url from the scope:

import logging
import re

from starlette.types import ASGIApp, Message, Receive, Scope, Send

logger = logging.getLogger(__name__)


class MountFastMCP:
    """
    Prefix the path in MCP 'event: endpoint' with additional root_path from the ASGI scope.

    This corrects the issue where the MCP server (like fastmcp) generates
    endpoint URLs (e.g., '/messages/') relative to its own root, without
    knowledge of the path it's mounted under (e.g., '/mcp/server1').
    This middleware ensures the client receives the correctly prefixed path
    (e.g., '/mcp/server1/messages/').

    Args:
        app: The ASGI application to wrap (typically the result of
             `FastMCP().sse_app()`).
        uri_prefix: An additional bytes to add before the relative path to the messages endpoint.
    """

    def __init__(self, app: ASGIApp, uri_prefix: bytes = b"") -> None:
        self.app = app
        self.uri_prefix = uri_prefix
        # Regex to find the endpoint event and capture relevant parts
        # Group 1: 'event: endpoint\r\ndata: '
        # Group 2: The relative path (e.g., /messages/ or just /) - Must start with /
        # Group 3: The remainder including optional query string and CRLFs
        #          (e.g., ?session_id=value\r\n\r\n or just \r\n\r\n)
        self._endpoint_event_regex = re.compile(
            # Match 'event: endpoint\r\ndata: ' literally
            rb"^(event: endpoint\r\ndata: )"
            # Match the path: must start with '/',
            # followed by zero or more non-'?' chars
            rb"(/[^?]*)"
            # Match the rest: optional query string starting with '?'
            # up to the end CRLFs
            rb"(\?.*?\r\n\r\n.*)$",
            re.DOTALL,  # Allow '.' to match newline characters if needed within group 3
        )

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            # Pass through non-HTTP scopes directly (e.g., 'lifespan', 'websocket')
            await self.app(scope, receive, send)
            return

        # Store the root_path from the scope this middleware instance receives.
        # This represents the path *up to* where this middleware (and the app it wraps)
        # is mounted.
        # Ensure it doesn't end with a slash IF it's not the root '/' itself.
        raw_root_path = scope.get("root_path", "")
        if raw_root_path != "/" and raw_root_path.endswith("/"):
            root_path = raw_root_path.rstrip("/")
        else:
            root_path = raw_root_path

        root_path_bytes = root_path.encode("utf-8")

        async def wrapped_send(message: Message) -> None:
            if message["type"] == "http.response.body":
                body = message.get("body", b"")
                # Optimization: Only attempt regex match if the prefix is present
                if body.startswith(b"event: endpoint\r\ndata: "):
                    # logger.debug(
                    #     f"[MountFastMCP] Found potential endpoint event: "
                    #     f"{body[:150]}..."
                    # )
                    match = self._endpoint_event_regex.match(body)
                    if match:
                        prefix, relative_path, remainder = match.groups()

                        # Construct full path, avoid double slashes if root_path is '/'
                        # and relative_path starts with '/' (which it should).
                        if root_path_bytes == b"/":
                            # If mounted at root, the relative path is already correct
                            full_path = relative_path
                        else:
                            # Otherwise, prepend the root path
                            full_path = root_path_bytes + relative_path
                        base_url = scope.get("base_url") or ""

                        # Reconstruct the message body
                        new_body = prefix + base_url.encode("utf-8") + full_path + remainder
                        message["body"] = new_body  # Modify the message in place
                        logger.info(
                            f"[MountFastMCP] Rewrote endpoint path. "
                            f"Original relative: {relative_path.decode()}, "
                            f"New full: {full_path.decode()}"
                        )
                        # logger.debug(
                        #     f"[MountFastMCP] Rewritten body: {new_body[:150]}..."
                        # )
                    # else:
                    # Log if the prefix matched but the full regex
                    # didn't (might indicate unexpected format)
                    # logger.warning(
                    #     f"[MountFastMCP] Endpoint event prefix found but "
                    #     f"regex failed detailed match. Body: {body[:150]}..."
                    # )

            # Send the original or modified message
            await send(message)

        # Call the wrapped application with the original scope/receive,
        # but using our wrapped_send function.
        await self.app(scope, receive, wrapped_send)