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

> Discover how ChatMCP implements SSE for MCP communication using a dedicated SSEClient. Learn about stream establishment endpoint discovery and bidirectional JSON-RPC messaging.

- Repository: [刀刀/chatmcp](https://github.com/daodao97/chatmcp)
- Tags: architecture
- Published: 2026-02-28

---

**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`:

```dart
// 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:

```dart
// 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:

```dart
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:

```dart
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:

```dart
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:

```dart
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.