How ChatMCP Implements SSE for MCP Communication: A Deep Dive into the Dart Client Architecture

ChatMCP uses a dedicated SSEClient class to establish persistent Server-Sent Events connections with MCP servers, implementing a three-step workflow of stream establishment, endpoint discovery, and bidirectional JSON-RPC messaging over HTTP.

The daodao97/chatmcp repository provides a robust Model Context Protocol (MCP) client implementation in Dart that supports multiple transport mechanisms. For server-sent events specifically, the codebase leverages the EventFlux library to maintain long-lived connections while handling the complexities of endpoint negotiation, message correlation, and automatic reconnection.

SSE Connection Architecture and Endpoint Discovery

The ChatMCP SSE implementation begins with establishing a unidirectional event stream before upgrading to full bidirectional communication.

Establishing the SSE Stream with EventFlux

In lib/mcp/sse/sse_client.dart, the SSEClient initializes connections using the EventFlux library. The client spawns a new isolate and establishes a GET request to the URL defined in ServerConfig.command:

// From sse_client.dart - Connection establishment
await EventFlux.spawn();
await EventFlux.connect(
  sseEndpoint,
  onSuccessCallback: (stream) {
    _onConnected();
  },
  onError: (error) {
    _scheduleReconnection();
  },
  reconnectConfig: ReconnectConfig(
    autoReconnect: true,
    max attempts: 5,
    reconnectInterval: Duration(seconds: 3),
  ),
);

The connection includes automatic reconfiguration through ReconnectConfig, which implements exponential back-off to handle transient network failures without manual intervention.

Endpoint Discovery and Message URL Resolution

Once the SSE stream is active, the client enters a waitingForEndpoint state. MCP servers send a special endpoint event containing the HTTP POST URL for subsequent RPC calls:

// From sse_client.dart - Endpoint event handler
void _handleEndpointEvent(String data) {
  final uri = Uri.parse(data);
  final baseUri = Uri.parse(_serverConfig.command);
  final normalizedUri = baseUri.resolveUri(uri);
  
  _messageEndpoint = normalizedUri.toString();
  _endpointConfirmedCompleter.complete();
  
  // Transition to connected state
  _connectionState = ConnectionState.connected;
}

This discovery mechanism allows the server to dynamically assign message endpoints while the client maintains a single persistent SSE stream for incoming data. The implementation uses a Completer (_endpointConfirmedCompleter) with a configurable timeout (_endpointTimeout) to prevent indefinite blocking if the endpoint event never arrives.

JSON-RPC Message Exchange Over SSE

With the endpoint discovered, ChatMCP implements bidirectional communication by separating transport directions: SSE for server-to-client messages, HTTP POST for client-to-server requests.

Sending Requests via HTTP POST

All outgoing JSON-RPC messages—including initialize, ping, and tool calls—are sent via the _sendHttpPost method defined at lines 48-66 of sse_client.dart. This method acquires a write lock to prevent race conditions during transmission:

Future<void> _sendHttpPost(Map<String, dynamic> json) async {
  await _writeLock.synchronized(() async {
    if (_messageEndpoint == null) {
      throw StateError('Message endpoint not available');
    }
    
    final response = await http.post(
      Uri.parse(_messageEndpoint),
      headers: _headers,
      body: jsonEncode(json),
    );
    
    if (response.statusCode != 200) {
      throw Exception('HTTP ${response.statusCode}: ${response.body}');
    }
  });
}

The client maintains a _pendingRequests map keyed by message ID to correlate asynchronous responses with their originating requests.

Handling Responses on the SSE Stream

Incoming SSE events are parsed by _handleSSEEvent (lines 60-76), which deserializes JSON-RPC payloads and routes them appropriately:

void _handleSSEEvent(String data) {
  final message = JSONRPCMessage.fromJson(jsonDecode(data));
  
  // Match to pending request
  if (message.id != null && _pendingRequests.containsKey(message.id)) {
    final completer = _pendingRequests.remove(message.id);
    completer?.complete(message);
  } else {
    // Forward notifications to callback
    onMessage?.call(message);
  }
}

This dual-transport approach allows the client to maintain the SSE stream's server-push capability while utilizing standard HTTP POST semantics for request delivery.

Connection State Management and Reconnection

ChatMCP implements sophisticated state tracking to ensure reliable MCP communication across unstable network conditions.

Robust Reconnection with Exponential Back-off

The ConnectionState enum tracks progression through discrete phases: disconnected → connecting → waitingForEndpoint → connected → reconnecting. When EventFlux detects a stream closure or error, it automatically attempts reconnection based on the ReconnectConfig parameters.

For the alternative streamable client implementation (lib/mcp/streamable/streamable_client.dart), manual reconnection is handled via _scheduleReconnection, which clears the pending request map to prevent memory leaks from orphaned futures:

void _scheduleReconnection() {
  _pendingRequests.forEach((id, completer) {
    completer.completeError('Connection lost');
  });
  _pendingRequests.clear();
  
  Future.delayed(Duration(seconds: 5), () {
    if (_connectionState != ConnectionState.connected) {
      _connect();
    }
  });
}

OAuth Authentication in SSE Headers

The SSEClient supports authenticated MCP servers through Bearer token injection. The _headers getter (lines 37-51) conditionally adds authorization headers to both the initial SSE request and subsequent HTTP POSTs:

Map<String, String> get _headers {
  final headers = {
    'Content-Type': 'application/json; charset=utf-8',
    'Accept': 'application/json; charset=utf-8',
  };
  
  if (_serverConfig.oauth?.enabled == true &&
      _serverConfig.oauth?.accessToken != null &&
      _serverConfig.oauth!.isTokenValid) {
    headers['Authorization'] = 'Bearer ${_serverConfig.oauth!.accessToken}';
  }
  
  return headers;
}

This ensures that authentication credentials propagate consistently across both transport channels.

Summary

• EventFlux Integration: ChatMCP uses the EventFlux library in lib/mcp/sse/sse_client.dart to manage persistent SSE connections with automatic reconnection capabilities.
• Endpoint Discovery: The client resolves relative endpoint URLs from SSE events to establish the HTTP POST destination for JSON-RPC requests.
• Split Transport Architecture: Server-sent events handle incoming messages while standard HTTP POSTs handle outgoing requests, maintaining bidirectional communication.
• Robust State Management: Connection states are tracked explicitly with Completer objects for async coordination and automatic cleanup of pending requests during reconnections.
• OAuth Support: Bearer tokens are injected into both SSE and HTTP headers when ServerConfig.oauth.enabled is true, supporting authenticated MCP servers.

Frequently Asked Questions

What is the role of EventFlux in ChatMCP's SSE client?

EventFlux serves as the underlying Dart library that manages the Server-Sent Events protocol implementation in lib/mcp/sse/sse_client.dart. It handles the HTTP GET request to establish the text/event-stream connection, parses incoming server events, and provides the ReconnectConfig mechanism for automatic reconnection with exponential back-off when network interruptions occur.

How does ChatMCP handle SSE reconnection when the connection drops?

The implementation uses a two-tier reconnection strategy. First, EventFlux's built-in ReconnectConfig automatically attempts reconnection with configurable intervals. Second, the SSEClient maintains a _pendingRequests map that gets cleared during reconnection events to prevent memory leaks, with each pending Completer receiving an error notification so calling code can handle the failure appropriately.

Can ChatMCP SSE client support authenticated MCP servers?

Yes, the SSEClient fully supports OAuth 2.0 Bearer token authentication. When ServerConfig.oauth.enabled is true and a valid access token is present, the _headers getter automatically includes Authorization: Bearer <token> in both the initial SSE connection request and all subsequent HTTP POST requests to the message endpoint, ensuring authenticated communication throughout the session.

What is the difference between SSEClient and StreamableClient in ChatMCP?

SSEClient in lib/mcp/sse/sse_client.dart implements the standard MCP SSE transport where the server pushes events and the client posts to a separate endpoint. StreamableClient in lib/mcp/streamable/streamable_client.dart provides an alternative HTTP-streaming implementation that shares similar event-handling logic but may differ in connection lifecycle management. Both extend the base client functionality and are instantiated through the factory in lib/mcp/mcp.dart based on the ServerConfig.type value.

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