connect method

Future<void> connect(ClientTransport transport, { ProtocolEra era = ProtocolEra.auto, @Deprecated('Use era: ProtocolEra.modern (true) or ProtocolEra.legacy (false).') bool? statelessMode, })

Connect the client to a transport.

era picks the revision family:

• ProtocolEra.auto (default): probe with server/discover (waiting eraProbeTimeout); a modern answer selects 2026-07-28, anything else the initialize handshake.
• ProtocolEra.modern: 2026-07-28 — no handshake; every request carries its _meta (protocol version, client info, capabilities) and, over HTTP, MCP-Protocol-Version: 2026-07-28. Server capabilities come from discover.
• ProtocolEra.legacy: the initialize handshake (2025-11-25 and earlier).

statelessMode is the 2.x form: true is ProtocolEra.modern, false is ProtocolEra.legacy. Pass it or era, not both.

onConnect fires once the connection is ready.

Implementation

Future<void> connect(
  ClientTransport transport, {
  ProtocolEra era = ProtocolEra.auto,
  @Deprecated('Use era: ProtocolEra.modern (true) or ProtocolEra.legacy (false).')
  bool? statelessMode,
}) async {
  if (statelessMode != null) {
    if (era != ProtocolEra.auto) {
      throw ArgumentError('Pass era or statelessMode, not both');
    }
    era = statelessMode ? ProtocolEra.modern : ProtocolEra.legacy;
  }
  if (_transport != null) {
    throw McpError('Client is already connected to a transport');
  }

  if (_connecting) {
    throw McpError('Client is already connecting to a transport');
  }

  _connecting = true;
  _statelessMode = era != ProtocolEra.legacy;
  _transport = transport;
  // `onError` is load-bearing. A transport reports a wire failure by adding
  // an error to this stream (an HTTP POST to a closed port, a dropped
  // socket). Listening without a handler left that error unhandled and the
  // request completers untouched, so a request on a dead endpoint waited out
  // its 30-second timeout instead of failing when the wire did.
  _transport!.onMessage.listen(_handleMessage, onError: _handleTransportError);
  _transport!.onClose
      .then((_) {
        // Only send disconnect event if we're still connected
        if (_transport != null) {
          _disconnectStreamController.add(DisconnectReason.transportClosed);
          _onDisconnect();
        }
      })
      .catchError((error) {
        // Only handle error if we're still connected
        if (_transport != null) {
          _errorStreamController.add(McpError('Transport error: $error'));
          _disconnectStreamController.add(DisconnectReason.transportError);
          _onDisconnect();
        }
      });

  // Set up message handling
  _messageController.stream.listen((message) async {
    try {
      await _processMessage(message);
    } catch (e) {
      _logger.debug('Error processing message: $e');
      _errorStreamController.add(McpError('Error processing message: $e'));
    }
  });

  // Initialize the connection
  try {
    if (_statelessMode) {
      // Stateless core: no handshake. Record the version so the HTTP
      // transport stamps `MCP-Protocol-Version: 2026-07-28`, and mark the
      // client ready to send requests immediately. Server capabilities are
      // fetched on demand via [discover].
      _useModern();
      if (era == ProtocolEra.auto && !await _probeModern()) {
        _useLegacy();
        await initialize();
      }
    } else {
      await initialize();
    }
    _connecting = false;

    // Emit connection event after successful initialization
    if (_initialized && _serverInfo != null && _serverCapabilities != null) {
      _connectStreamController.add(
        Implementation.fromJson(Map<String, Object?>.from(_serverInfo!)),
      );
    }
  } catch (e) {
    _connecting = false;
    _errorStreamController.add(McpError('Initialization error: $e'));
    disconnect();
    rethrow;
  }
}