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');
final noResponse = outbound.remove(noResponseKey) == true;
if (noResponse) {
_endWithoutResponse(inflightKey);
return;
}
// 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) {
final exchange = _statelessExchanges[inflightKey];
if (exchange?.stream != null) {
// Already streaming: the response is the stream's last event.
exchange!.answered = true;
exchange.keepAlive?.cancel();
_statelessExchanges.remove(inflightKey);
if (!identical(outbound[noResponseKey], true)) {
_sendSseEvent(exchange.stream!, outbound);
}
exchange.stream!.close();
if (!statelessCompleter.isCompleted) {
statelessCompleter.complete(_streamed);
}
} else 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;
}
// 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 (_sseLogByKey.containsKey(inflightKey)) {
// SSE response for specific request; logged even when the client
// is momentarily disconnected, so a reconnect can still get it.
final log = _logFor(inflightKey, targetSessionId);
final stream = _sseStreams.remove(inflightKey);
_sendLogged(log, stream?.controller, outbound);
log.endedAt = DateTime.now();
// The response ends the request's stream.
stream?.controller.close();
_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 request. One stream only: the
// stream of the request it belongs to while that is open, otherwise
// the session's GET stream (logged for replay when not connected).
final cleanMessage = Map<String, dynamic>.from(message);
final targetSessionId =
cleanMessage.remove('_targetSessionId') as String?;
final related = cleanMessage.remove('_relatedRequestId');
if (targetSessionId != null && related != null) {
final key = _inflightKey(targetSessionId, related);
// A 2026-07-28 request: its notifications turn the answer into a
// stream and go on it.
if (_statelessExchanges.containsKey(key)) {
_openStatelessStream(key);
final exchange = _statelessExchanges[key];
if (exchange?.stream != null) {
_sendSseEvent(exchange!.stream!, cleanMessage);
}
return;
}
final stream = _sseStreams[key];
if (stream != null) {
_sendLogged(
_logFor(key, targetSessionId),
stream.controller,
cleanMessage,
);
return;
}
}
final sessions =
targetSessionId != null
? [targetSessionId]
: _activeSessions.toList();
for (final session in sessions) {
if (!_activeSessions.contains(session)) continue;
_sendLogged(
_logFor('get:$session', session),
_getStreams[session]?.controller,
cleanMessage,
);
}
}
} catch (e, stackTrace) {
_logger.error('Error sending message: $e');
_logger.debug('Stack trace: $stackTrace');
}
}