improvement(mcp): use SDK protocol versions + pool pinned undici agents + cover OAuth lock

- McpClient.SUPPORTED_VERSIONS removed; getVersionInfo() and the inbound
  serve route both import LATEST_PROTOCOL_VERSION / SUPPORTED_PROTOCOL_VERSIONS
  directly from @modelcontextprotocol/sdk/types.js. New protocol revisions
  ship automatically with SDK upgrades.
- pinned-fetch now caches undici Agents in a module-level LRU keyed by
  resolvedIP (max 64). Back-to-back MCP calls to the same server reuse the
  keep-alive connection pool instead of opening fresh TCP + TLS each time.
- New integration tests for withMcpOauthRefreshLock covering: in-process
  dedup via coalesceLocally, cross-process serialization via Redis mutex,
  fall-open on Redis unavailable, lock release on throw, release-failure
  swallow, per-row key isolation.

Author: waleedlatif1

apps/sim/app/api/mcp/serve/[serverId]/route.test.ts

@@ -232,14 +232,16 @@ describe('MCP Serve Route', () => {
       expect(body.result.protocolVersion).toBe('2024-11-05')
     })
 
-    it('falls back to latest when client requests unknown version', async () => {
+    it('falls back to SDK latest when client requests unknown version', async () => {
+      const { LATEST_PROTOCOL_VERSION } = await import('@modelcontextprotocol/sdk/types.js')
       const body = await callInitialize('2099-01-01')
-      expect(body.result.protocolVersion).toBe('2025-06-18')
+      expect(body.result.protocolVersion).toBe(LATEST_PROTOCOL_VERSION)
     })
 
-    it('falls back to latest when client omits protocolVersion', async () => {
+    it('falls back to SDK latest when client omits protocolVersion', async () => {
+      const { LATEST_PROTOCOL_VERSION } = await import('@modelcontextprotocol/sdk/types.js')
       const body = await callInitialize(undefined)
-      expect(body.result.protocolVersion).toBe('2025-06-18')
+      expect(body.result.protocolVersion).toBe(LATEST_PROTOCOL_VERSION)
     })
   })
 })

apps/sim/app/api/mcp/serve/[serverId]/route.ts

@@ -11,8 +11,10 @@ import {
   type JSONRPCError,
   type JSONRPCMessage,
   type JSONRPCResultResponse,
+  LATEST_PROTOCOL_VERSION,
   type ListToolsResult,
   type RequestId,
+  SUPPORTED_PROTOCOL_VERSIONS,
   type Tool,
 } from '@modelcontextprotocol/sdk/types.js'
 import { db } from '@sim/db'
@@ -36,20 +38,12 @@ import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('WorkflowMcpServeAPI')
 
-// Newest first. We echo the client's version when we support it (per
-// MCP 2025-06-18 lifecycle spec); otherwise fall back to our latest.
-const SUPPORTED_PROTOCOL_VERSIONS = ['2025-06-18', '2025-03-26', '2024-11-05'] as const
-const LATEST_PROTOCOL_VERSION = SUPPORTED_PROTOCOL_VERSIONS[0]
-
 function negotiateProtocolVersion(rpcParams: unknown): string {
   const requested =
     rpcParams && typeof rpcParams === 'object' && 'protocolVersion' in rpcParams
       ? (rpcParams as { protocolVersion?: unknown }).protocolVersion
       : undefined
-  if (
-    typeof requested === 'string' &&
-    (SUPPORTED_PROTOCOL_VERSIONS as readonly string[]).includes(requested)
-  ) {
+  if (typeof requested === 'string' && SUPPORTED_PROTOCOL_VERSIONS.includes(requested)) {
     return requested
   }
   return LATEST_PROTOCOL_VERSION

apps/sim/lib/mcp/client.ts

@@ -2,7 +2,9 @@ import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js'
 import { Client } from '@modelcontextprotocol/sdk/client/index.js'
 import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
 import {
+  LATEST_PROTOCOL_VERSION,
   type ListToolsResult,
+  SUPPORTED_PROTOCOL_VERSIONS,
   type Tool,
   ToolListChangedNotificationSchema,
 } from '@modelcontextprotocol/sdk/types.js'
@@ -40,12 +42,6 @@ export class McpClient {
   private authProvider?: McpClientOptions['authProvider']
   private isConnected = false
 
-  private static readonly SUPPORTED_VERSIONS = [
-    '2025-06-18', // Latest stable with elicitation and OAuth 2.1
-    '2025-03-26', // Streamable HTTP support
-    '2024-11-05', // Initial stable release
-  ]
-
   constructor(options: McpClientOptions) {
     this.config = options.config
     this.securityPolicy = options.securityPolicy ?? {
@@ -245,8 +241,8 @@ export class McpClient {
 
   static getVersionInfo(): McpVersionInfo {
     return {
-      supported: [...McpClient.SUPPORTED_VERSIONS],
-      preferred: McpClient.SUPPORTED_VERSIONS[0],
+      supported: [...SUPPORTED_PROTOCOL_VERSIONS],
+      preferred: LATEST_PROTOCOL_VERSION,
     }
   }
 

apps/sim/lib/mcp/oauth/storage.test.ts

@@ -11,11 +11,25 @@ import {
 } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
+const { mockAcquireLock, mockReleaseLock } = vi.hoisted(() => ({
+  mockAcquireLock: vi.fn(),
+  mockReleaseLock: vi.fn(),
+}))
+
 vi.mock('@sim/db', () => dbChainMock)
 vi.mock('@sim/db/schema', () => schemaMock)
 vi.mock('@/lib/core/security/encryption', () => encryptionMock)
+vi.mock('@/lib/core/config/redis', () => ({
+  acquireLock: mockAcquireLock,
+  releaseLock: mockReleaseLock,
+}))
 
-import { getOrCreateOauthRow, loadOauthRow, setOauthRowUser } from './storage'
+import {
+  getOrCreateOauthRow,
+  loadOauthRow,
+  setOauthRowUser,
+  withMcpOauthRefreshLock,
+} from './storage'
 
 describe('MCP OAuth storage', () => {
   beforeEach(() => {
@@ -92,3 +106,83 @@ describe('MCP OAuth storage', () => {
     )
   })
 })
+
+describe('withMcpOauthRefreshLock', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockAcquireLock.mockReset()
+    mockReleaseLock.mockReset()
+    mockReleaseLock.mockResolvedValue(true)
+  })
+
+  it('coalesces concurrent in-process callers onto a single fn execution', async () => {
+    mockAcquireLock.mockResolvedValue(true)
+    const fn = vi.fn(async () => 'tokens')
+
+    const results = await Promise.all([
+      withMcpOauthRefreshLock('row-coalesce', fn),
+      withMcpOauthRefreshLock('row-coalesce', fn),
+      withMcpOauthRefreshLock('row-coalesce', fn),
+    ])
+
+    expect(results).toEqual(['tokens', 'tokens', 'tokens'])
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(mockAcquireLock).toHaveBeenCalledTimes(1)
+    expect(mockReleaseLock).toHaveBeenCalledTimes(1)
+  })
+
+  it('serializes cross-process callers: follower polls until leader releases', async () => {
+    // First acquire fails (another process holds it), second succeeds.
+    mockAcquireLock.mockResolvedValueOnce(false).mockResolvedValueOnce(true)
+    const fn = vi.fn(async () => 'fresh')
+
+    const result = await withMcpOauthRefreshLock('row-mutex', fn)
+
+    expect(result).toBe('fresh')
+    expect(mockAcquireLock).toHaveBeenCalledTimes(2)
+    expect(fn).toHaveBeenCalledTimes(1)
+  })
+
+  it('falls open when Redis is unavailable on acquire', async () => {
+    mockAcquireLock.mockRejectedValueOnce(new Error('Redis connection refused'))
+    const fn = vi.fn(async () => 'uncoordinated')
+
+    const result = await withMcpOauthRefreshLock('row-redis-down', fn)
+
+    expect(result).toBe('uncoordinated')
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(mockReleaseLock).not.toHaveBeenCalled()
+  })
+
+  it('releases the lock even when fn throws', async () => {
+    mockAcquireLock.mockResolvedValue(true)
+    const fn = vi.fn(async () => {
+      throw new Error('refresh failed')
+    })
+
+    await expect(withMcpOauthRefreshLock('row-throws', fn)).rejects.toThrow('refresh failed')
+
+    expect(mockReleaseLock).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not surface releaseLock failures to the caller', async () => {
+    mockAcquireLock.mockResolvedValue(true)
+    mockReleaseLock.mockRejectedValueOnce(new Error('release failed'))
+    const fn = vi.fn(async () => 'value')
+
+    const result = await withMcpOauthRefreshLock('row-release-fail', fn)
+    expect(result).toBe('value')
+  })
+
+  it('uses per-row lock keys so different rows do not serialize', async () => {
+    mockAcquireLock.mockResolvedValue(true)
+    const fn = vi.fn(async () => 'ok')
+
+    await Promise.all([withMcpOauthRefreshLock('row-a', fn), withMcpOauthRefreshLock('row-b', fn)])
+
+    expect(mockAcquireLock).toHaveBeenCalledTimes(2)
+    const keys = mockAcquireLock.mock.calls.map((c) => c[0])
+    expect(keys).toContain('mcp:oauth:refresh:row-a')
+    expect(keys).toContain('mcp:oauth:refresh:row-b')
+  })
+})

apps/sim/lib/mcp/pinned-fetch.test.ts

@@ -25,12 +25,13 @@ vi.mock('@/lib/core/security/input-validation.server', () => ({
   createPinnedLookup: mockCreatePinnedLookup,
 }))
 
-import { createMcpPinnedFetch } from '@/lib/mcp/pinned-fetch'
+import { __resetPinnedAgentsForTests, createMcpPinnedFetch } from '@/lib/mcp/pinned-fetch'
 
 describe('createMcpPinnedFetch', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     capturedAgentOptions.length = 0
+    __resetPinnedAgentsForTests()
     mockCreatePinnedLookup.mockReturnValue('pinned-lookup-fn')
     mockUndiciFetch.mockResolvedValue(new Response('ok'))
   })
@@ -73,7 +74,7 @@ describe('createMcpPinnedFetch', () => {
     expect(init.dispatcher).toBeInstanceOf(mockAgent)
   })
 
-  it('reuses the same dispatcher across calls (one Agent per fetch instance)', async () => {
+  it('reuses the same dispatcher across calls within a fetch instance', async () => {
     const fetchLike = createMcpPinnedFetch('203.0.113.10')
     await fetchLike('https://example.com/a')
     await fetchLike('https://example.com/b')
@@ -82,4 +83,26 @@ describe('createMcpPinnedFetch', () => {
     const d2 = (mockUndiciFetch.mock.calls[1][1] as { dispatcher: unknown }).dispatcher
     expect(d1).toBe(d2)
   })
+
+  it('pools agents by resolvedIP across createMcpPinnedFetch calls', async () => {
+    const a = createMcpPinnedFetch('203.0.113.10')
+    const b = createMcpPinnedFetch('203.0.113.10')
+    await a('https://example.com/a')
+    await b('https://example.com/b')
+    expect(capturedAgentOptions).toHaveLength(1)
+    const d1 = (mockUndiciFetch.mock.calls[0][1] as { dispatcher: unknown }).dispatcher
+    const d2 = (mockUndiciFetch.mock.calls[1][1] as { dispatcher: unknown }).dispatcher
+    expect(d1).toBe(d2)
+  })
+
+  it('creates separate agents for different resolved IPs', async () => {
+    const a = createMcpPinnedFetch('203.0.113.10')
+    const b = createMcpPinnedFetch('198.51.100.20')
+    await a('https://example.com/a')
+    await b('https://example.com/b')
+    expect(capturedAgentOptions).toHaveLength(2)
+    const d1 = (mockUndiciFetch.mock.calls[0][1] as { dispatcher: unknown }).dispatcher
+    const d2 = (mockUndiciFetch.mock.calls[1][1] as { dispatcher: unknown }).dispatcher
+    expect(d1).not.toBe(d2)
+  })
 })

apps/sim/lib/mcp/pinned-fetch.ts

@@ -3,29 +3,51 @@ import { Agent, type RequestInit as UndiciRequestInit, fetch as undiciFetch } fr
 import { createPinnedLookup } from '@/lib/core/security/input-validation.server'
 
 /**
- * Creates a FetchLike that pins all outbound HTTP connections to a pre-resolved
- * IP address. Used by the MCP transport to prevent DNS-rebinding (TOCTOU)
- * attacks: validation performs DNS once and confirms the IP is allowed; this
- * fetch then forces every subsequent request (initial POST, SSE GET, redirects)
- * to use that same IP, regardless of what the hostname now resolves to.
+ * Pins outbound HTTP connections to a pre-resolved IP to prevent DNS-rebinding
+ * between URL validation and connection. Hostname is preserved so TLS SNI and
+ * the Host header still match the certificate.
  *
- * Uses undici's `fetch` directly so the `dispatcher` option is part of the
- * real type contract — not a cast that would silently break if a future
- * runtime swapped out the implementation.
- *
- * The original hostname is preserved on the request so TLS SNI and the Host
- * header continue to match the certificate.
+ * Agents are pooled by `resolvedIP` so back-to-back calls to the same server
+ * reuse the same keep-alive connection pool instead of opening a fresh TCP +
+ * TLS connection per McpClient instance.
  */
+const MAX_POOLED_AGENTS = 64
+const pinnedAgents = new Map<string, Agent>()
+
+function getPinnedAgent(resolvedIP: string): Agent {
+  const existing = pinnedAgents.get(resolvedIP)
+  if (existing) {
+    // LRU touch — re-insert to mark as most recently used.
+    pinnedAgents.delete(resolvedIP)
+    pinnedAgents.set(resolvedIP, existing)
+    return existing
+  }
+  if (pinnedAgents.size >= MAX_POOLED_AGENTS) {
+    const oldestKey = pinnedAgents.keys().next().value
+    if (oldestKey !== undefined) {
+      const oldest = pinnedAgents.get(oldestKey)
+      pinnedAgents.delete(oldestKey)
+      oldest?.close().catch(() => {})
+    }
+  }
+  const agent = new Agent({ connect: { lookup: createPinnedLookup(resolvedIP) } })
+  pinnedAgents.set(resolvedIP, agent)
+  return agent
+}
+
+export function __resetPinnedAgentsForTests(): void {
+  for (const agent of pinnedAgents.values()) {
+    try {
+      void agent.close?.()
+    } catch {}
+  }
+  pinnedAgents.clear()
+}
+
 export function createMcpPinnedFetch(resolvedIP: string): FetchLike {
-  const dispatcher = new Agent({
-    connect: { lookup: createPinnedLookup(resolvedIP) },
-  })
+  const dispatcher = getPinnedAgent(resolvedIP)
 
   return (async (url, init) => {
-    // DOM `RequestInit` and undici's `RequestInit` are structurally compatible
-    // at runtime (Node's global fetch IS undici) but differ in TS types.
-    // Cast the init through unknown to bridge the typing without losing the
-    // critical `dispatcher` typing on the call itself.
     const undiciInit: UndiciRequestInit = {
       // double-cast-allowed: DOM RequestInit and undici RequestInit are structurally compatible at runtime (Node's global fetch IS undici) but the TS types differ
       ...(init as unknown as UndiciRequestInit),