connect method
Connect the server to a transport
Implementation
void connect(ServerTransport transport) {
if (_transport != null) {
_logger.debug('Server already has a transport connected');
// Handle STDIO transport specifically
if (transport is StdioServerTransport &&
_transport is StdioServerTransport) {
_logger.debug(
'Attempting to connect another STDIO transport - reusing existing connection',
);
// Use existing transport instead of the new one
transport = _transport as StdioServerTransport;
} else {
throw McpError('Server is already connected to a transport');
}
}
_transport = transport;
// Determine if this is a multi-session transport (StreamableHTTP)
final isMultiSessionTransport = transport is StreamableHttpServerTransport;
// Wire RFC 9728 metadata into the streamable HTTP transport so the
// spec route `/.well-known/oauth-protected-resource` resolves to the
// current snapshot of `protectedResourceMetadata`. Using a callback
// (rather than a one-shot copy) so a later
// `configureProtectedResource(...)` is reflected without reconnect.
if (isMultiSessionTransport) {
transport.setProtectedResourceMetadataProvider(
() => protectedResourceMetadata,
);
// Authorization is decided per HTTP request, so a refusal is an HTTP
// 401/403 with a challenge a standard client acts on.
transport.setAuthorizer(_authorizeHttp);
// Let the transport version-gate JSON-RPC batching on the session's
// negotiated revision (batching valid only ≤ 2025-03-26). The transport
// assembles the batched array response; the gate stays anchored to the
// same negotiated version the dispatch path uses.
transport.setNegotiatedVersionResolver(
(sessionId) => _sessions[sessionId]?.negotiatedProtocolVersion,
);
}
// Create initial session only for single-session transports (for backward compatibility)
String? initialSessionId;
if (!isMultiSessionTransport) {
initialSessionId = _createSession(transport);
}
try {
transport.onMessage.listen(
(rawMessage) {
// 2026-07-28 stateless core: a request the transport flagged
// `_stateless` carries no persistent session. Handle it on a
// transient, per-request session (client info/caps sourced from
// `_meta`) that is torn down after the response — the server keeps
// no session state for it.
//
// SECURITY: `_stateless` is a transport-internal control key. Honor it
// ONLY when the connected transport genuinely has stateless enabled,
// so a client that forges `_stateless` in its request body cannot
// activate the 2026-07-28 path where it is off over any transport —
// stdio/SSE (no stateless support) and a StreamableHTTP transport with
// `enableStateless: false` both evaluate to false here. Transports also
// strip client-forged reserved keys at ingestion (defense in depth).
// A 2026-07-28 client closed its request's stream: that cancels it.
if (rawMessage is Map &&
rawMessage[StreamableHttpServerTransport.cancelInflightKey] ==
true) {
final sessionId = rawMessage['_sessionId'];
final requestId = '${rawMessage['id']}';
for (final op in _pendingOperations.values) {
if (op.sessionId == sessionId && op.requestId == requestId) {
op.isCancelled = true;
_logger.info(
'Request $requestId cancelled: the client closed its stream',
);
}
}
// A closed subscriptions/listen stream ends that subscription.
_statelessSubscriptions.remove(
_subscriptionKey('$sessionId', rawMessage['id']),
);
return;
}
final t = _transport;
final statelessAllowed =
t is StreamableHttpServerTransport && t.config.enableStateless;
if (rawMessage is Map &&
rawMessage['_stateless'] == true &&
statelessAllowed) {
unawaited(_handleStatelessInbound(rawMessage));
return;
}
// 2026-07-28 on a single-channel transport (stdio): a request whose
// `_meta` names its protocol version is modern and served
// statelessly; `initialize` keeps the legacy era for the process
// (versioning: a dual-era server selects from how the client opens).
if (t is! StreamableHttpServerTransport &&
rawMessage is Map &&
!rawMessage.containsKey('_sessionId')) {
if (_isModernRequest(rawMessage)) {
unawaited(_handleModernChannelRequest(rawMessage));
return;
}
if (_cancelModernChannelRequest(rawMessage)) return;
}
// Extract session ID from message metadata (if present)
String? messageSessionId;
if (rawMessage is Map && rawMessage.containsKey('_sessionId')) {
// Multi-session transport: use session ID from message
messageSessionId = rawMessage['_sessionId'] as String;
// Create session if it doesn't exist
if (!_sessions.containsKey(messageSessionId)) {
_logger.debug(
'Creating new session from transport message: $messageSessionId',
);
final session = ClientSession(
id: messageSessionId,
connectedAt: DateTime.now(),
transport:
transport, // IMPORTANT: Set transport reference for sending responses
);
session.capabilities = {};
_sessions[messageSessionId] = session;
// Update metrics
_metricsCollector
.gauge('mcp_connections_active')
.set(_sessions.length.toDouble());
_metricsCollector.counter('mcp_connections_total').increment();
// Emit the connect event so consumers can track sessions —
// the single-session path goes through `_createSession`
// which fires this; the multi-session inline path used by
// streamable HTTP / SSE was previously skipping it,
// leaving `server.onConnect` silent for those transports.
if (!_connectStreamController.isClosed) {
_connectStreamController.add(session);
}
}
} else {
// Single-session transport: use initial session ID
messageSessionId = initialSessionId;
}
if (messageSessionId == null) {
_logger.error('No session ID available for message: $rawMessage');
return;
}
_handleMessage(messageSessionId, rawMessage);
},
onError: (error) {
_logger.error('Error from transport message stream: $error');
},
);
transport.onClose
.then((_) {
// Remove all sessions associated with this transport
final sessionIds = List<String>.from(_sessions.keys);
for (final sid in sessionIds) {
_removeSession(sid);
}
_onDisconnect();
})
.catchError((error) {
_logger.error('Transport close error: $error');
// Remove all sessions associated with this transport
final sessionIds = List<String>.from(_sessions.keys);
for (final sid in sessionIds) {
_removeSession(sid);
}
_onDisconnect();
});
} catch (e) {
_logger.error('Failed to setup transport listeners: $e');
}
// Set up message processing with better error handling
_messageController.stream.listen(
(message) async {
try {
await _processMessage(message.sessionId, message);
} catch (e, stackTrace) {
_logger.error('Error processing message: $e');
_logger.debug('Stack trace: $stackTrace');
try {
_sendErrorResponse(
message.sessionId,
message.id,
ErrorCode.internalError,
'Internal error: $e',
);
} catch (sendError) {
_logger.error('Failed to send error response: $sendError');
}
}
},
onError: (error) {
_logger.error('Error in message stream: $error');
},
);
}