send method
Send a message through the transport
Implementation
@override
void send(dynamic message) {
if (_isClosed) return;
try {
_logger.debug('StreamableHTTP send() called with message: $message');
// 2026-07-28 `subscriptions/listen` (SEP-2577) routing. Route to the
// long-lived SSE stream keyed by subscriptionId, BEFORE the normal
// response/notification handling below. Legacy paths are unaffected — a
// message only matches when its id (terminal result) or its
// `_meta.subscriptionId` (stream notification) names an open subscription.
if (message is Map && _statelessSubscriptionStreams.isNotEmpty) {
if (_routeStatelessSubscription(message)) return;
}
// A message is a JSON-RPC RESPONSE only when it has an id AND no
// method (i.e. it carries `result` or `error`). A server-INITIATED
// request (e.g., `sampling/createMessage`, `elicitation/create`)
// also has an id but is NOT a response — it must be routed to the
// standalone GET stream so the client can pick it up. The earlier
// `containsKey('id')` check sent every id-bearing message into
// the response-completion path, which dropped outbound
// server-initiated requests on the floor with `No pending request
// found for response with ID: srv-N`.
final isResponse = message is Map &&
message.containsKey('id') &&
!message.containsKey('method');
if (isResponse) {
final requestId = message['id'];
// Session half of the in-flight key, stamped by `Server._sendResponse`
// / `_sendErrorResponse`. Without it a response cannot be attributed to
// a session, and matching on the bare id would be exactly the collision
// this key exists to prevent — so drop it rather than guess.
final targetSessionId = message['_targetSessionId'];
if (targetSessionId is! String) {
_logger.error(
'Dropping response id=$requestId: no _targetSessionId. A response '
'must be routed by (session, id); matching on the bare id would '
'cross sessions.');
return;
}
final inflightKey = _inflightKey(targetSessionId, requestId);
_logger.debug('Response $inflightKey (id type: ${requestId.runtimeType})');
// Internal routing metadata never reaches the wire.
final outbound = Map<String, dynamic>.from(message)
..remove('_targetSessionId');
// 2026-07-28 stateless: resolve the one-shot completer independent of
// the JSON/SSE response-mode config. Checked first so a stateless
// reply never falls into the session-scoped SSE/JSON routing below.
final statelessCompleter = _statelessCompleters.remove(inflightKey);
if (statelessCompleter != null) {
if (!statelessCompleter.isCompleted) {
statelessCompleter.complete(outbound);
}
return;
}
// JSON-RPC batch entry: resolve the batch completer independent of the
// response-mode config (the batch is answered as one JSON array in
// `_handleBatchRequest`). Checked before the mode-scoped routing below.
final batchCompleter = _batchCompleters[inflightKey];
if (batchCompleter != null) {
if (!batchCompleter.isCompleted) {
batchCompleter.complete(outbound);
}
return;
}
// Generate event ID for resumability
final eventId = (_eventIdCounter++).toString();
// Store event for resumability
_eventStore[eventId] = EventMessage(
message: outbound,
eventId: eventId,
);
// Handle based on mode
if (config.isJsonResponseEnabled) {
if (config.jsonResponseMode == 'sync') {
// Synchronous JSON mode: complete the pending completer
final completer = _pendingCompleters.remove(inflightKey);
if (completer != null) {
try {
completer.complete(outbound);
_logger.debug('Completed completer for request $inflightKey');
} catch (e, stackTrace) {
_logger.error('Error completing completer for $inflightKey: $e');
_logger.debug('Stack trace: $stackTrace');
}
} else {
_logger.warning('No pending completer for response $inflightKey');
_logger.debug('Available completers: ${_pendingCompleters.keys.toList()}');
}
} else {
// Asynchronous JSON mode: store response for polling. The store key
// and the in-flight key are the same `<session>:<id>` shape.
if (_pendingRequests.remove(inflightKey) != null) {
_responseStore[inflightKey] = outbound;
_responseTimestamps[inflightKey] = DateTime.now();
} else {
_logger.warning('No pending request for async response $inflightKey');
}
}
} else if (_sseStreams.containsKey(inflightKey)) {
// SSE response for specific request
final stream = _sseStreams[inflightKey]!;
_sendSseEvent(stream.controller, outbound, eventId: eventId);
// If this is a response or error, close the stream
if (message.containsKey('result') || message.containsKey('error')) {
stream.controller.close();
_sseStreams.remove(inflightKey);
_messageRouters.remove(inflightKey)?.close();
}
} else {
// Log when we can't find a pending request for a response
_logger.warning('No pending request found for response $inflightKey');
_logger.debug('Current pending requests: ${_pendingRequests.keys.toList()}');
_logger.debug('Current SSE streams: ${_sseStreams.keys.toList()}');
}
} else {
// Notification or server-initiated message
final eventId = (_eventIdCounter++).toString();
// Remove internal metadata before storing and sending
final cleanMessage = Map<String, dynamic>.from(message);
final targetSessionId = cleanMessage.remove('_targetSessionId') as String?;
// GET-stream event (notification / broadcast) — eligible for replay
// on a `Last-Event-ID` reconnect. Broadcasts carry a null target.
_eventStore[eventId] = EventMessage(
message: cleanMessage,
eventId: eventId,
forGetStream: true,
targetSessionId: targetSessionId,
);
var sent = false;
// If target session is specified, send only to that session's GET stream
if (targetSessionId != null && _getStreams.containsKey(targetSessionId)) {
_logger.debug('📤 Sending notification to target session: $targetSessionId');
_sendSseEvent(_getStreams[targetSessionId]!.controller, cleanMessage, eventId: eventId);
sent = true;
} else if (targetSessionId == null) {
// Broadcast mode: send to all GET streams
for (final entry in _getStreams.entries) {
_logger.debug('📤 Broadcasting notification to GET stream (session: ${entry.key})');
_sendSseEvent(entry.value.controller, cleanMessage, eventId: eventId);
sent = true;
}
// If no GET stream available, send to all active POST SSE streams
if (!sent && _sseStreams.isNotEmpty) {
_logger.debug('No GET stream available, sending to ${_sseStreams.length} active POST SSE streams');
for (final entry in _sseStreams.entries) {
_logger.debug('📤 Broadcasting notification to POST SSE stream (requestId: ${entry.key})');
_sendSseEvent(entry.value.controller, cleanMessage, eventId: eventId);
sent = true;
}
}
} else {
_logger.warning('Target session $targetSessionId not found or no GET stream available');
}
if (!sent) {
_logger.debug('No streams available to send notification');
}
}
} catch (e, stackTrace) {
_logger.error('Error sending message: $e');
_logger.debug('Stack trace: $stackTrace');
}
}