fix(agent-core): reconnect dropped streamable-HTTP MCP sessions on tool call

.changeset/fix-mcp-http-session-reconnect.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix dropped streamable-HTTP MCP sessions never reconnecting on the legacy engine: tool calls now reconnect and retry transparently, and a failed server's tools stay registered and fail with the server's own error while it is down. `tool.list.updated` with reason `mcp.disconnected` now fires only when a server is removed or disabled, not on transient reconnects.

packages/agent-core/src/agent/tool/index.ts

@@ -7,16 +7,24 @@ import {
   collectLoadedDynamicToolNames,
 } from '../context/dynamic-tools';
 import type { ContextMessage } from '../context/types';
-import { makeErrorPayload } from '../../errors';
-import type { ExecutableTool, ToolUpdate } from '../../loop';
+import { ErrorCodes, KimiError, makeErrorPayload } from '../../errors';
+import type { ExecutableTool, ExecutableToolContext, ToolUpdate } from '../../loop';
+import { errorMessage, isAbortError } from '../../loop/errors';
 import { createMcpAuthTool } from '../../mcp/auth-tool';
 import type { McpConnectionManager, McpServerEntry } from '../../mcp';
+import {
+  isMcpConnectionClosedError,
+  isMcpMalformedResultError,
+  isMcpTransportFailure,
+  probeMcpLiveness,
+} from '../../mcp/client-shared';
 import { mcpResultToExecutableOutput } from '../../mcp/output';
 import { isMcpToolName, qualifyMcpToolName } from '../../mcp/tool-naming';
-import type { MCPClient, MCPToolDefinition } from '../../mcp/types';
+import type { MCPClient, MCPToolDefinition, MCPToolResult } from '../../mcp/types';
 import { resolveSubagentTimeoutMs } from '../../session/subagent-host';
 import { buildSubagentModelDescriptions } from '../../session/subagent-binding';
 import { extendWorkspaceWithSkillRoots } from '../../skill';
+import { abortable } from '../../utils/abort';
 import { fingerprint } from '../llm-request-logger';
 import * as b from '../../tools/builtin';
 import type { ToolStore, ToolStoreData, ToolStoreKey } from '../../tools/store';
@@ -327,11 +335,19 @@ export class ToolManager {
               // `args` has already been JSON-parsed and schema-validated by
               // the loop's preflight (`loop/tool-call.ts`), so the MCP
               // client gets a plain object directly.
-              const result = await client.callTool(
-                tool.name,
-                (args ?? {}) as Record<string, unknown>,
-                context.signal,
-              );
+              const mcpArgs = (args ?? {}) as Record<string, unknown>;
+              let result: MCPToolResult;
+              try {
+                result = await client.callTool(tool.name, mcpArgs, context.signal);
+              } catch (error) {
+                result = await retryMcpCallAfterReconnect(
+                  error,
+                  client,
+                  (activeClient) => activeClient.callTool(tool.name, mcpArgs, context.signal),
+                  context,
+                  this.mcpToolCallReconnect(serverName, client, context.signal),
+                );
+              }
               return mcpResultToExecutableOutput(result, qualified, {
                 originalsDir: this.agent.mediaOriginalsDir,
                 telemetry: this.agent.telemetry,
@@ -359,6 +375,22 @@ export class ToolManager {
     return true;
   }
 
+  /**
+   * Builds the `reconnect` callback for the wrapped-tool recovery path (see
+   * {@link retryMcpCallAfterReconnect}), or returns `undefined` when this
+   * manager has no connection manager behind it (tests wiring a bare fake
+   * client) — in that case calls keep their old fail-fast behavior.
+   */
+  private mcpToolCallReconnect(
+    serverName: string,
+    staleClient: MCPClient,
+    signal: AbortSignal,
+  ): (() => Promise<MCPClient | undefined>) | undefined {
+    const mcp = this.agent.mcp;
+    if (mcp === undefined) return undefined;
+    return () => abortable(joinHealedMcpClientOrReconnect(mcp, serverName, staleClient), signal);
+  }
+
   private handleMcpServerStatusChange(mcp: McpConnectionManager, entry: McpServerEntry): void {
     if (entry.status === 'connected') {
       this.registerConnectedMcpServer(mcp, entry);
@@ -369,15 +401,15 @@ export class ToolManager {
       return;
     }
     if (entry.status === 'failed') {
-      this.unregisterMcpServer(entry.name);
-      this.agent.emitEvent({
-        type: 'tool.list.updated',
-        reason: 'mcp.failed',
-        serverName: entry.name,
-      });
+      // Keep the tools registered: a dropped connection is recovered through
+      // the wrapped call's reconnect-and-retry path, and until then calls
+      // fail loudly with the server's error instead of vanishing from the
+      // tool list (which made the model retry through *other* servers'
+      // tools — see #2742). The tool list itself did not change, so no
+      // `tool.list.updated` event is emitted.
       return;
     }
-    if (entry.status === 'disabled' || entry.status === 'pending') {
+    if (entry.status === 'disabled') {
       const removed = this.unregisterMcpServer(entry.name);
       if (removed) {
         this.agent.emitEvent({
@@ -387,6 +419,14 @@ export class ToolManager {
         });
       }
     }
+    // `pending` is deliberately NOT handled: it precedes every (re)connect
+    // attempt, so unregistering here would drop the tools mid-reconnect —
+    // and after a failed recovery they would stay gone for the rest of the
+    // session, leaving later calls unable to drive another reconnect (#2742).
+    // Keeping them is safe because both re-registration paths above start by
+    // unregistering the server's tools (`registerMcpServer` and
+    // `registerNeedsAuthMcpServer`), so a `connected` or `needs-auth`
+    // transition fully replaces the tool set rather than merging into it.
   }
 
   private registerNeedsAuthMcpServer(mcp: McpConnectionManager, entry: McpServerEntry): void {
@@ -1025,3 +1065,95 @@ export class ToolManager {
       .filter((tool) => !!tool);
   }
 }
+
+/**
+ * Recovery for a failed MCP tool call, mirroring agent-core-v2's
+ * `retryAfterReconnect` (`agent-core-v2/src/agent/mcp/tools/mcp.ts`):
+ *
+ * - The server answered (a JSON-RPC error, or a response that failed
+ *   client-side schema validation) → the error is rethrown; reconnecting
+ *   would not change the answer.
+ * - The failure is ambiguous (a raw fetch/socket error) → the client is
+ *   probed with a ping: alive means a transient blip and the call is
+ *   retried once in place; dead means the transport is gone.
+ * - The transport is provably dead (the SDK reported the connection closed,
+ *   or the probe failed) → the server is reconnected once through
+ *   `reconnect` and the call retried on the fresh client, so a dropped
+ *   streamable-HTTP session (e.g. an MCP server restart) surfaces as a slow
+ *   call instead of failing every call for the rest of the session (#2742).
+ *
+ * Retries are at-least-once: if the transport died after the server
+ * processed the call but before the response arrived, the retry may
+ * duplicate side effects. There is no protocol-level dedup across
+ * reconnects, so this trade-off is accepted deliberately.
+ */
+async function retryMcpCallAfterReconnect(
+  error: unknown,
+  client: MCPClient,
+  callTool: (activeClient: MCPClient) => Promise<MCPToolResult>,
+  context: ExecutableToolContext,
+  reconnect: (() => Promise<MCPClient | undefined>) | undefined,
+): Promise<MCPToolResult> {
+  const isUnrecoverable = (e: unknown): boolean =>
+    context.signal.aborted ||
+    isAbortError(e) ||
+    !isMcpTransportFailure(e) ||
+    isMcpMalformedResultError(e);
+  if (reconnect === undefined || isUnrecoverable(error)) {
+    throw error;
+  }
+
+  let failure = error;
+  if (!isMcpConnectionClosedError(failure)) {
+    const alive = await probeMcpLiveness(client, context.signal);
+    context.signal.throwIfAborted();
+    if (alive) {
+      try {
+        return await callTool(client);
+      } catch (retryError) {
+        if (isUnrecoverable(retryError)) {
+          throw retryError;
+        }
+        failure = retryError;
+      }
+    }
+  }
+
+  context.onUpdate?.({ kind: 'status', text: 'MCP connection lost — reconnecting…' });
+  let freshClient: MCPClient | undefined;
+  try {
+    freshClient = await reconnect();
+  } catch (reconnectError) {
+    if (context.signal.aborted || isAbortError(reconnectError)) {
+      throw reconnectError;
+    }
+    throw new KimiError(
+      ErrorCodes.MCP_STARTUP_FAILED,
+      `${errorMessage(failure)} (reconnecting the MCP server also failed: ${errorMessage(reconnectError)})`,
+      { cause: reconnectError },
+    );
+  }
+  if (freshClient === undefined) {
+    throw failure;
+  }
+  return callTool(freshClient);
+}
+
+/**
+ * Return the current client when the server already healed (a concurrent
+ * call finished the reconnect first), otherwise drive one shared reconnect
+ * through the manager — `reconnectAndJoin` dedupes parallel attempts — and
+ * return the client it produced. `undefined` means there is no fresh client
+ * to retry on, so the original failure should surface.
+ */
+async function joinHealedMcpClientOrReconnect(
+  mcp: McpConnectionManager,
+  serverName: string,
+  staleClient: MCPClient,
+): Promise<MCPClient | undefined> {
+  const healed = mcp.resolved(serverName)?.client;
+  if (healed !== undefined && healed !== staleClient) return healed;
+  await mcp.reconnectAndJoin(serverName);
+  const current = mcp.resolved(serverName)?.client;
+  return current !== undefined && current !== staleClient ? current : undefined;
+}

packages/agent-core/src/mcp/client-http.ts

@@ -7,6 +7,7 @@ import {
   buildRequestOptions,
   KIMI_MCP_CLIENT_NAME,
   KIMI_MCP_CLIENT_VERSION,
+  MCP_LIVENESS_PROBE_TIMEOUT_MS,
   toMcpToolDefinition,
   toMcpToolResult,
   type UnexpectedCloseListener,
@@ -145,6 +146,10 @@ export class HttpMcpClient implements MCPClient {
     return toMcpToolResult(result);
   }
 
+  async ping(signal?: AbortSignal): Promise<void> {
+    await this.client.ping(buildRequestOptions(MCP_LIVENESS_PROBE_TIMEOUT_MS, signal));
+  }
+
   private async closeStartedClient(): Promise<void> {
     if (!this.started) return;
     this.started = false;

packages/agent-core/src/mcp/client-shared.ts

@@ -1,6 +1,7 @@
 import { getCoreVersion } from '#/version';
+import { ErrorCode, McpError } from '@modelcontextprotocol/sdk/types.js';
 
-import type { MCPToolDefinition, MCPToolResult } from './types';
+import type { MCPClient, MCPToolDefinition, MCPToolResult } from './types';
 
 export const KIMI_MCP_CLIENT_NAME = 'kimi-code';
 // Resolved from agent-core's package.json so MCP servers see the real version
@@ -31,6 +32,56 @@ export interface McpRequestOptions {
   readonly signal?: AbortSignal;
 }
 
+/**
+ * True when the SDK reports the connection itself as gone (the transport was
+ * closed, so no in-flight request can ever complete).
+ */
+export function isMcpConnectionClosedError(error: unknown): boolean {
+  return (
+    error instanceof Error &&
+    (error as Error & { readonly code?: unknown }).code === ErrorCode.ConnectionClosed
+  );
+}
+
+/**
+ * True when a failed tool call might recover after a reconnect: either the
+ * connection is closed, or the error is a raw transport/fetch failure rather
+ * than a JSON-RPC answer from the server ({@link McpError}) — reconnecting
+ * would not change a server-side answer.
+ */
+export function isMcpTransportFailure(error: unknown): boolean {
+  if (!(error instanceof Error)) return false;
+  if (isMcpConnectionClosedError(error)) return true;
+  return !(error instanceof McpError);
+}
+
+/** Bounded so a wedged server cannot stall the reconnect decision indefinitely. */
+export const MCP_LIVENESS_PROBE_TIMEOUT_MS = 5_000;
+
+/** Response failed client-side schema validation: the server answered, so it is alive. */
+export function isMcpMalformedResultError(error: unknown): boolean {
+  return error instanceof Error && error.name === 'ZodError';
+}
+
+/**
+ * Ping the server to decide whether its transport is still usable after an
+ * ambiguous failure. A malformed answer still proves liveness; a timeout or
+ * any transport-level failure means dead.
+ */
+export async function probeMcpLiveness(client: MCPClient, signal: AbortSignal): Promise<boolean> {
+  try {
+    await client.ping(signal);
+    return true;
+  } catch (error) {
+    if (isMcpConnectionClosedError(error)) return false;
+    if (isMcpMalformedResultError(error)) return true;
+    if (error instanceof McpError) {
+      return (error as Error & { readonly code?: unknown }).code !== ErrorCode.RequestTimeout;
+    }
+    return false;
+  }
+}
+
 /**
  * Build the `RequestOptions` object accepted by MCP SDK requests, including
  * either a configured timeout, an in-flight abort signal, both, or neither.

packages/agent-core/src/mcp/client-sse.ts

@@ -7,6 +7,7 @@ import {
   buildRequestOptions,
   KIMI_MCP_CLIENT_NAME,
   KIMI_MCP_CLIENT_VERSION,
+  MCP_LIVENESS_PROBE_TIMEOUT_MS,
   toMcpToolDefinition,
   toMcpToolResult,
   type UnexpectedCloseListener,
@@ -136,6 +137,10 @@ export class SseMcpClient implements MCPClient {
     return toMcpToolResult(result);
   }
 
+  async ping(signal?: AbortSignal): Promise<void> {
+    await this.client.ping(buildRequestOptions(MCP_LIVENESS_PROBE_TIMEOUT_MS, signal));
+  }
+
   private async closeStartedClient(): Promise<void> {
     if (!this.started) return;
     this.started = false;

packages/agent-core/src/mcp/client-stdio.ts

@@ -10,6 +10,7 @@ import {
   buildRequestOptions,
   KIMI_MCP_CLIENT_NAME,
   KIMI_MCP_CLIENT_VERSION,
+  MCP_LIVENESS_PROBE_TIMEOUT_MS,
   toMcpToolDefinition,
   toMcpToolResult,
   type UnexpectedCloseListener,
@@ -162,6 +163,10 @@ export class StdioMcpClient implements MCPClient {
     return toMcpToolResult(result);
   }
 
+  async ping(signal?: AbortSignal): Promise<void> {
+    await this.client.ping(buildRequestOptions(MCP_LIVENESS_PROBE_TIMEOUT_MS, signal));
+  }
+
   private async closeStartedClient(): Promise<void> {
     if (!this.started) return;
     this.started = false;

packages/agent-core/src/mcp/connection-manager.ts

@@ -129,6 +129,7 @@ export interface McpConnectionManagerOptions {
 export class McpConnectionManager {
   private readonly entries = new Map<string, InternalEntry>();
   private readonly listeners = new Set<McpStatusListener>();
+  private readonly inFlightReconnects = new Map<string, Promise<void>>();
   private initialLoad: Promise<void> = Promise.resolve();
   private initialLoadAttemptId = 0;
   private initialLoadStartedAt: number | undefined;
@@ -314,6 +315,24 @@ export class McpConnectionManager {
     await this.connectOne(entry, attemptId);
   }
 
+  /**
+   * Reconnect that joins an already in-flight reconnect for the same server
+   * instead of starting a second one. Used by the tool-call recovery path,
+   * where several parallel calls to a dropped server can all decide to
+   * reconnect at once.
+   */
+  reconnectAndJoin(name: string): Promise<void> {
+    const existing = this.inFlightReconnects.get(name);
+    if (existing !== undefined) return existing;
+    const work = this.reconnect(name).finally(() => {
+      if (this.inFlightReconnects.get(name) === work) {
+        this.inFlightReconnects.delete(name);
+      }
+    });
+    this.inFlightReconnects.set(name, work);
+    return work;
+  }
+
   async shutdown(): Promise<void> {
     const entries = Array.from(this.entries.values());
     this.entries.clear();

packages/agent-core/src/mcp/types.ts

@@ -88,6 +88,11 @@ export interface MCPClient {
     args: Record<string, unknown>,
     signal?: AbortSignal,
   ): Promise<MCPToolResult>;
+  /**
+   * Liveness probe used after a failed tool call to distinguish a transient
+   * blip from a dead transport before attempting a reconnect.
+   */
+  ping(signal?: AbortSignal): Promise<void>;
 }
 
 /**

packages/agent-core/test/agent/llm-request-recorder.test.ts

@@ -212,6 +212,7 @@ describe('mcp.tools_discovered records', () => {
       async callTool() {
         return { content: [{ type: 'text', text: 'ok' }], isError: false };
       },
+      async ping() {},
     };
     const entry: McpServerEntry = {
       name: input.serverName ?? 'grafana',
@@ -375,6 +376,7 @@ describe('mcp.tools_discovered records', () => {
       async callTool() {
         return { content: [], isError: false };
       },
+      async ping() {},
     };
     ctx.agent.tools.registerMcpServer('graf.ana', occupant, [
       { name: 'query_range', description: 'occupies the qualified name', parameters: {} },