fix(mcp): coordinate OAuth refresh without blocking tools (#7479)

Author: waleedlatif1

apps/sim/lib/credentials/application/discover-managed-mcp-tools.test.ts

@@ -121,7 +121,7 @@ describe('discoverManagedMcpToolsUseCase', () => {
     expect(mocks.discoverTools).toHaveBeenCalledWith(
       context.mcpServerId,
       context.workspaceId,
-      {},
+      { credentialId: context.credentialId, loadProvider: expect.any(Function) },
       signal,
       { requireComplete: true }
     )

apps/sim/lib/credentials/application/discover-managed-mcp-tools.ts

@@ -10,7 +10,6 @@ import {
   saveManagedMcpToolSnapshot,
 } from '@/lib/credentials/managed-mcp'
 import { loadManagedMcpAuthProvider } from '@/lib/mcp/application/managed-auth-provider'
-import { withMcpOauthRefreshLock } from '@/lib/mcp/oauth'
 import { mcpService } from '@/lib/mcp/service'
 
 export interface DiscoverManagedMcpToolsInput {
@@ -35,14 +34,15 @@ export const discoverManagedMcpToolsUseCase = defineAuthorizedWorkspaceUseCase({
   async execute({ input, context }) {
     input.signal?.throwIfAborted()
     const runtime = await loadManagedMcpRuntimeCredential(context.credentialId, context.workspaceId)
-    const tools = await withMcpOauthRefreshLock(runtime.credentialId, async () =>
-      mcpService.discoverManagedMcpTools(
-        runtime.mcpServerId,
-        runtime.workspaceId,
-        await loadManagedMcpAuthProvider(runtime.credentialId, runtime.workspaceId),
-        input.signal,
-        { requireComplete: true }
-      )
+    const tools = await mcpService.discoverManagedMcpTools(
+      runtime.mcpServerId,
+      runtime.workspaceId,
+      {
+        credentialId: runtime.credentialId,
+        loadProvider: () => loadManagedMcpAuthProvider(runtime.credentialId, runtime.workspaceId),
+      },
+      input.signal,
+      { requireComplete: true }
     )
     await saveManagedMcpToolSnapshot(
       runtime.credentialId,

apps/sim/lib/mcp/application/execute-managed-tool.test.ts

@@ -186,7 +186,7 @@ describe('executeManagedMcpToolUseCase', () => {
     expect(mocks.discoverTools).toHaveBeenCalledWith(
       context.mcpServerId,
       context.workspaceId,
-      {},
+      { credentialId: context.credentialId, loadProvider: expect.any(Function) },
       signal,
       { requireComplete: true }
     )

apps/sim/lib/mcp/application/execute-managed-tool.ts

@@ -17,7 +17,6 @@ import {
   validateToolArguments,
 } from '@/lib/mcp/application/execute-tool'
 import { loadManagedMcpAuthProvider } from '@/lib/mcp/application/managed-auth-provider'
-import { withMcpOauthRefreshLock } from '@/lib/mcp/oauth'
 import { mcpService } from '@/lib/mcp/service'
 import type { McpTool, McpToolCall, McpToolSchema } from '@/lib/mcp/types'
 
@@ -55,14 +54,15 @@ export const executeManagedMcpToolUseCase = defineAuthorizedWorkspaceUseCase({
   async execute({ input, context }): Promise<ExecuteMcpToolResult> {
     input.signal?.throwIfAborted()
     const runtime = await loadManagedMcpRuntimeCredential(context.credentialId, context.workspaceId)
-    const tools = await withMcpOauthRefreshLock(runtime.credentialId, async () =>
-      mcpService.discoverManagedMcpTools(
-        runtime.mcpServerId,
-        runtime.workspaceId,
-        await loadManagedMcpAuthProvider(runtime.credentialId, runtime.workspaceId),
-        input.signal,
-        { requireComplete: true }
-      )
+    const tools = await mcpService.discoverManagedMcpTools(
+      runtime.mcpServerId,
+      runtime.workspaceId,
+      {
+        credentialId: runtime.credentialId,
+        loadProvider: () => loadManagedMcpAuthProvider(runtime.credentialId, runtime.workspaceId),
+      },
+      input.signal,
+      { requireComplete: true }
     )
     await saveManagedMcpToolSnapshot(
       runtime.credentialId,

apps/sim/lib/mcp/client.test.ts

@@ -71,8 +71,12 @@ vi.mock('@/lib/core/execution-limits', () => ({
 
 import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
-import { McpClient } from './client'
-import type { McpClientOptions, McpServerConfig } from './types'
+import { McpClient } from '@/lib/mcp/client'
+import {
+  type McpClientOptions,
+  McpOauthAuthorizationRequiredError,
+  type McpServerConfig,
+} from '@/lib/mcp/types'
 
 function createConfig(): McpServerConfig {
   return {
@@ -94,6 +98,14 @@ describe('McpClient notification handler', () => {
     vi.mocked(getMaxExecutionTimeout).mockReturnValue(30_000)
   })
 
+  it('preserves authorization-required errors raised by a locked credential reload', async () => {
+    const error = new McpOauthAuthorizationRequiredError('server-1', 'Test Server')
+    mockSdkConnect.mockRejectedValueOnce(error)
+    const client = new McpClient({ config: createConfig() })
+    await expect(client.connect()).rejects.toBe(error)
+    expect(client.getStatus().lastError).toBeUndefined()
+  })
+
   it('fires onToolsChanged when a notification arrives while connected', async () => {
     const onToolsChanged = vi.fn()
 

apps/sim/lib/mcp/client.ts

@@ -13,6 +13,7 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { getMcpSafeErrorDiagnostics } from '@/lib/mcp/error-diagnostics'
 import { McpOauthRedirectRequired } from '@/lib/mcp/oauth'
+import { createCoordinatedMcpOauthFetch } from '@/lib/mcp/oauth/coordinated-fetch'
 import { createGuardedMcpFetch, createPinnedPrivateMcpFetch } from '@/lib/mcp/pinned-fetch'
 import {
   type McpClientOptions,
@@ -21,6 +22,7 @@ import {
   type McpConsentRequest,
   type McpConsentResponse,
   McpError,
+  McpOauthAuthorizationRequiredError,
   type McpSecurityPolicy,
   type McpServerConfig,
   type McpTool,
@@ -47,7 +49,10 @@ function classifyConnectionOutcome(
   error: unknown,
   authType: McpServerConfig['authType']
 ): ConnectionOutcome {
-  if (error instanceof McpOauthRedirectRequired) {
+  if (
+    error instanceof McpOauthRedirectRequired ||
+    error instanceof McpOauthAuthorizationRequiredError
+  ) {
     return 'authorization_required'
   }
   if (error instanceof UnauthorizedError) {
@@ -100,8 +105,15 @@ export class McpClient {
       throw new McpError('URL required for Streamable HTTP transport')
     }
 
-    if (this.config.authType === 'oauth' && this.authProvider == null) {
-      throw new McpError('OAuth MCP server requires an authProvider')
+    if (
+      this.config.authType === 'oauth' &&
+      this.authProvider == null &&
+      !options.oauthCredentials
+    ) {
+      throw new McpError('OAuth MCP server requires OAuth credentials')
+    }
+    if (options.oauthCredentials && this.authProvider) {
+      throw new McpError('OAuth MCP server must use one authentication strategy')
     }
     const useOauth = this.config.authType === 'oauth'
     // `resolvedIP` is null only when the hostname still carries an unresolved env-var
@@ -115,10 +127,18 @@ export class McpClient {
         : createGuardedMcpFetch(this.config.url)
       : undefined
     this.closeGuardedTransport = guarded?.close
+    const transportFetch =
+      useOauth && options.oauthCredentials
+        ? createCoordinatedMcpOauthFetch(options.oauthCredentials, {
+            serverUrl: this.config.url,
+            fetch: guarded?.fetch ?? fetch,
+            requestInit: { headers: this.config.headers },
+          })
+        : guarded?.fetch
     this.transport = new StreamableHTTPClientTransport(new URL(this.config.url), {
       authProvider: useOauth ? this.authProvider : undefined,
       requestInit: { headers: this.config.headers },
-      ...(guarded ? { fetch: guarded.fetch } : {}),
+      ...(transportFetch ? { fetch: transportFetch } : {}),
     })
 
     this.client = new Client(

apps/sim/lib/mcp/connection-manager.test.ts

@@ -63,6 +63,7 @@ vi.mock('@/lib/mcp/oauth', () => ({
 }))
 
 import { McpConnectionManager } from '@/lib/mcp/connection-manager'
+import type { McpClientOptions } from '@/lib/mcp/types'
 
 beforeAll(() => {
   setEnvFlags({ isTest: false })
@@ -168,6 +169,11 @@ describe('McpConnectionManager', () => {
         userId: 'user-1',
         workspaceId: 'ws-1',
       })
+      const options: McpClientOptions = MockMcpClientConstructor.mock.calls[0][0]
+      expect(options.authProvider).toBeUndefined()
+      expect(options.oauthCredentials?.credentialId).toBe('server-oauth')
+      await options.oauthCredentials?.loadProvider()
+      expect(mockGetOrCreateOauthRow).toHaveBeenCalledTimes(2)
     })
 
     it('allows a new connect() after a previous one completes', async () => {

apps/sim/lib/mcp/connection-manager.ts

@@ -16,12 +16,13 @@ import { isTest } from '@/lib/core/config/env-flags'
 import { McpClient } from '@/lib/mcp/client'
 import { getOrCreateOauthRow, loadPreregisteredClient, SimMcpOauthProvider } from '@/lib/mcp/oauth'
 import { mcpPubSub } from '@/lib/mcp/pubsub'
-import type {
-  ManagedConnectionState,
-  McpClientOptions,
-  McpServerConfig,
-  McpToolsChangedCallback,
-  ToolsChangedEvent,
+import {
+  type ManagedConnectionState,
+  type McpClientOptions,
+  McpOauthAuthorizationRequiredError,
+  type McpServerConfig,
+  type McpToolsChangedCallback,
+  type ToolsChangedEvent,
 } from '@/lib/mcp/types'
 
 const logger = createLogger('McpConnectionManager')
@@ -137,7 +138,7 @@ export class McpConnectionManager {
         this.handleToolsChanged(key)
       }
 
-      let authProvider: McpClientOptions['authProvider']
+      let oauthCredentials: McpClientOptions['oauthCredentials']
       if (config.authType === 'oauth') {
         const row = await getOrCreateOauthRow({
           mcpServerId: config.id,
@@ -150,8 +151,25 @@ export class McpConnectionManager {
           )
           return { supportsListChanged: false }
         }
-        const preregistered = await loadPreregisteredClient(config.id)
-        authProvider = new SimMcpOauthProvider({ row, preregistered })
+        oauthCredentials = {
+          credentialId: config.id,
+          initialProvider: new SimMcpOauthProvider({
+            row,
+            preregistered: await loadPreregisteredClient(config.id),
+          }),
+          loadProvider: async () => {
+            const current = await getOrCreateOauthRow({
+              mcpServerId: config.id,
+              userId,
+              workspaceId,
+            })
+            if (!current.tokens) {
+              throw new McpOauthAuthorizationRequiredError(config.id, config.name)
+            }
+            const preregistered = await loadPreregisteredClient(config.id)
+            return new SimMcpOauthProvider({ row: current, preregistered })
+          },
+        }
       }
 
       const client = new McpClient({
@@ -163,7 +181,7 @@ export class McpConnectionManager {
         },
         onToolsChanged,
         resolvedIP: resolvedIP ?? undefined,
-        authProvider,
+        oauthCredentials,
       })
 
       try {