How SSE Transport Handles Endpoint Discovery and Message Routing in the MCP Protocol

The MCP SSE transport implements a two-step handshake where the client first connects to an SSE stream to discover the JSON-RPC endpoint via an endpoint event, then routes messages asynchronously by correlating JSON-RPC request IDs with incoming message events.

The junjiem/dify-plugin-tools-mcp_sse repository provides a Python implementation of the Message Control Protocol (MCP) using Server-Sent Events (SSE) as the transport layer. Understanding how SSE transport handles endpoint discovery and message routing in the MCP protocol requires examining the McpSseClient class in utils/mcp_client.py, which manages the asynchronous handshake, URL validation, and request-response correlation through event-driven architecture.

Endpoint Discovery Mechanism

The MCP protocol over SSE uses a dynamic endpoint discovery pattern that separates the SSE connection URL from the actual JSON-RPC message endpoint. This design allows servers to allocate ephemeral messaging endpoints while maintaining a stable SSE stream.

Establishing the SSE Connection

When McpSseClient is instantiated, it immediately opens a persistent SSE connection to the user-supplied URL (self.url). The _listen_messages coroutine begins watching the stream for specific event types. According to the source code in utils/mcp_client.py, the client awaits an event named endpoint that signals the completion of the handshake.

case "endpoint":
    # build absolute endpoint URL

    self.endpoint_url = urljoin(self.url.rstrip("/"), sse.data.lstrip("/"))
    logger.info(f"{self.name} - Received endpoint URL: {self.endpoint_url}")
    self._connected.set()

URL Construction and Validation

The client constructs the absolute endpoint URL using urljoin(self.url.rstrip("/"), sse.data.lstrip("/")), ensuring proper handling of relative paths returned by the server. The implementation validates that the discovered endpoint shares the same origin as the original SSE connection to prevent cross-origin attacks.

# verify same origin

url_parsed = urlparse(self.url)
endpoint_parsed = urlparse(self.endpoint_url)
if (url_parsed.netloc != endpoint_parsed.netloc
        or url_parsed.scheme != endpoint_parsed.scheme):
    raise ValueError(...)

This security check, implemented in lines 22-32 of utils/mcp_client.py, ensures that the JSON-RPC endpoint matches the scheme and host of the SSE connection before the client proceeds with message routing.

Asynchronous Message Routing Architecture

After endpoint discovery completes, the transport uses an asynchronous routing system that decouples request sending from response handling through the SSE stream.

Sending JSON-RPC Requests

The send_message method posts JSON-RPC requests containing an "id" field to the discovered self.endpoint_url. The method then blocks on the response_ready threading event, waiting for the corresponding response to arrive via the SSE stream.

if "id" in data:
    message_id = data["id"]
    while True:
        self.response_ready.wait()
        self.response_ready.clear()
        if message_id in self.message_dict:
            message = self.message_dict.pop(message_id, None)
            if message and message.get("method") == "ping":
                continue
            return message

This implementation, found in lines 66-78 of utils/mcp_client.py, demonstrates how the client handles asynchronous response matching while filtering out protocol-level ping messages.

Correlating Responses via Event Stream

The _listen_messages coroutine processes incoming SSE events continuously. When it encounters a message event, it parses the JSON payload and stores the message in self.message_dict keyed by the message's "id", then signals that a response is ready.

case "message":
    message = json.loads(sse.data)
    logger.debug(f"{self.name} - Received server message: {message}")
    self.message_dict[message["id"]] = message
    self.response_ready.set()

This routing mechanism, defined in lines 35-39 of utils/mcp_client.py, enables the client to match out-of-order responses to their original requests using the JSON-RPC ID field as the correlation key.

Connection Lifecycle and Error Handling

The McpSseClient manages connection state through threading primitives and comprehensive error propagation.

State Management

The client uses two primary synchronization primitives: _connected, which signals successful endpoint discovery, and response_ready, which indicates that a message has been received. These events coordinate between the listener thread (reading SSE events) and the main thread (sending requests).

Exception Propagation

Any exception occurring in the _listen_messages thread is captured in _thread_exception and propagated to callers of send_message or connect. The close method stops the listener thread, closes the underlying httpx.Client, and cleans up synchronization primitives to prevent resource leaks.

Practical Implementation Example

The following implementation demonstrates the complete workflow of initializing an MCP client, performing endpoint discovery, and routing messages:

from utils.mcp_client import McpSseClient

# Create an SSE-based MCP client

client = McpSseClient(
    name="MyPlugin",
    url="https://example.com/mcp/sse",   # SSE endpoint provided by the server

    headers={"Authorization": "Bearer <token>"}
)

# Initialise the MCP session (sends initialize + notifications/initialized)

client.initialize()

# Call an MCP method, e.g., list available tools

tools = client.list_tools()
print(tools)

# Clean up when finished

client.close()

This pattern is utilized throughout the repository, including in provider/mcp_tool.py for Dify plugin integration, and in specific tool implementations like tools/mcp_list_tools.py and tools/mcp_call_tool.py.

Summary

• Two-step handshake: The SSE transport first establishes a stream to discover the JSON-RPC endpoint via an endpoint event, then uses that endpoint for subsequent messages.
• Dynamic URL construction: The client builds absolute URLs using urljoin and enforces same-origin policy by validating scheme and host in utils/mcp_client.py.
• ID-based correlation: Messages are routed by matching JSON-RPC "id" fields between sent requests and received message events stored in self.message_dict.
• Asynchronous synchronization: The implementation uses threading events (_connected, response_ready) to coordinate between the SSE listener and request sender.
• Lifecycle management: Proper cleanup via close() prevents resource leaks and ensures thread termination.

Frequently Asked Questions

How does the MCP SSE client discover the JSON-RPC endpoint?

The client opens an SSE connection to the provided URL and listens for an event named endpoint. The data payload contains the relative path to the JSON-RPC endpoint, which the client resolves to an absolute URL using urljoin while enforcing same-origin security checks in utils/mcp_client.py.

What happens if the discovered endpoint has a different origin than the SSE connection?

The client validates that the discovered endpoint shares the same scheme and host (netloc) as the original SSE URL. If validation fails, the client raises a ValueError and aborts the connection, preventing potential cross-origin request forgery attacks.

How does the client match responses to requests in the SSE transport?

The send_message method generates a unique message ID for each JSON-RPC request. The _listen_messages coroutine stores incoming messages in a dictionary keyed by their ID and signals response_ready. The sender waits for this signal and retrieves the matching response from the dictionary, filtering out ping messages automatically.

What is the purpose of ping messages in the MCP protocol?

Ping messages serve as keep-alive signals in the JSON-RPC stream. The client detects these by checking if message.get("method") == "ping" and automatically continues waiting for the actual response, ensuring connection vitality without exposing protocol-level heartbeats to the application layer.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}