improvement(mcp): bound MCP memory and lifecycle concurrency

apps/sim/app/api/mcp/servers/[id]/route.ts

@@ -3,7 +3,11 @@ import { toError } from '@sim/utils/errors'
 import type { NextRequest } from 'next/server'
 import { updateMcpServerBodySchema } from '@/lib/api/contracts/mcp'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { performUpdateMcpServer } from '@/lib/mcp/orchestration'
 import {
   createMcpErrorResponse,
@@ -28,7 +32,7 @@ export const PATCH = withRouteHandler(
       try {
         const { id: serverId } = await params
 
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = updateMcpServerBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -82,6 +86,8 @@ export const PATCH = withRouteHandler(
           server: { ...rest, hasOauthClientSecret: !!_secret },
         })
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error updating MCP server:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to update MCP server', 500)
       }

apps/sim/app/api/mcp/servers/route.ts

@@ -7,7 +7,11 @@ import type { NextRequest } from 'next/server'
 import { createMcpServerBodySchema, deleteMcpServerByQuerySchema } from '@/lib/api/contracts/mcp'
 import { validationErrorResponse } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { performCreateMcpServer, performDeleteMcpServer } from '@/lib/mcp/orchestration'
 import {
   createMcpErrorResponse,
@@ -55,7 +59,7 @@ export const POST = withRouteHandler(
   withMcpAuth('write')(
     async (request: NextRequest, { userId, userName, userEmail, workspaceId, requestId }) => {
       try {
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = createMcpServerBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -120,6 +124,8 @@ export const POST = withRouteHandler(
           result.updated ? 200 : 201
         )
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error registering MCP server:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to register MCP server', 500)
       }

apps/sim/app/api/mcp/servers/test-connection/route.ts

@@ -11,7 +11,11 @@ import {
   validateMcpDomain,
   validateMcpServerSsrf,
 } from '@/lib/mcp/domain-check'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { detectMcpAuthType } from '@/lib/mcp/oauth'
 import { resolveMcpConfigEnvVars } from '@/lib/mcp/resolve-config'
 import type { McpAuthType, McpTransport } from '@/lib/mcp/types'
@@ -64,7 +68,7 @@ function sanitizeConnectionError(error: unknown): string {
 export const POST = withRouteHandler(
   withMcpAuth('write')(async (request: NextRequest, { userId, workspaceId, requestId }) => {
     try {
-      const rawBody = getParsedBody(request) ?? (await request.json())
+      const rawBody = await readMcpJsonBodyWithLimit(request)
       const parsedBody = mcpServerTestBodySchema.safeParse(rawBody)
 
       if (!parsedBody.success) {
@@ -235,6 +239,8 @@ export const POST = withRouteHandler(
 
       return createMcpSuccessResponse(result, result.success ? 200 : 400)
     } catch (error) {
+      const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+      if (bodyErrorResponse) return bodyErrorResponse
       logger.error(`[${requestId}] Error testing MCP server connection:`, error)
       return createMcpErrorResponse(toError(error), 'Failed to test server connection', 500)
     }

apps/sim/app/api/mcp/tools/discover/route.ts

@@ -1,18 +1,55 @@
 import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js'
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import type { NextRequest } from 'next/server'
 import { mcpToolDiscoveryQuerySchema, refreshMcpToolsBodySchema } from '@/lib/api/contracts/mcp'
 import { validationErrorResponse } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { mcpService } from '@/lib/mcp/service'
 import { McpOauthAuthorizationRequiredError, type McpToolDiscoveryResponse } from '@/lib/mcp/types'
 import { categorizeError, createMcpErrorResponse, createMcpSuccessResponse } from '@/lib/mcp/utils'
 
 const logger = createLogger('McpToolDiscoveryAPI')
+const MCP_REFRESH_DISCOVERY_CONCURRENCY = 5
 
 export const dynamic = 'force-dynamic'
 
+async function settleWithConcurrency<T, R>(
+  items: T[],
+  concurrency: number,
+  task: (item: T) => Promise<R>
+): Promise<Array<PromiseSettledResult<R>>> {
+  const results: Array<PromiseSettledResult<R> | undefined> = new Array(items.length)
+  let nextIndex = 0
+
+  const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
+    while (nextIndex < items.length) {
+      const index = nextIndex
+      nextIndex += 1
+      try {
+        results[index] = { status: 'fulfilled', value: await task(items[index]) }
+      } catch (reason) {
+        results[index] = { status: 'rejected', reason }
+      }
+    }
+  })
+
+  await Promise.all(workers)
+
+  return results.map(
+    (result) =>
+      result ?? {
+        status: 'rejected',
+        reason: new Error('MCP refresh discovery task did not run'),
+      }
+  )
+}
+
 export const GET = withRouteHandler(
   withMcpAuth('read')(async (request: NextRequest, { userId, workspaceId, requestId }) => {
     try {
@@ -63,7 +100,7 @@ export const GET = withRouteHandler(
 export const POST = withRouteHandler(
   withMcpAuth('read')(async (request: NextRequest, { userId, workspaceId, requestId }) => {
     try {
-      const rawBody = getParsedBody(request) ?? (await request.json())
+      const rawBody = await readMcpJsonBodyWithLimit(request)
       const parsedBody = refreshMcpToolsBodySchema.safeParse(rawBody)
 
       if (!parsedBody.success) {
@@ -74,11 +111,13 @@ export const POST = withRouteHandler(
 
       logger.info(`[${requestId}] Refreshing tools for ${serverIds.length} servers`)
 
-      const results = await Promise.allSettled(
-        serverIds.map(async (serverId: string) => {
+      const results = await settleWithConcurrency(
+        serverIds,
+        MCP_REFRESH_DISCOVERY_CONCURRENCY,
+        async (serverId: string) => {
           const tools = await mcpService.discoverServerTools(userId, serverId, workspaceId, true)
           return { serverId, toolCount: tools.length }
-        })
+        }
       )
 
       const successes: Array<{ serverId: string; toolCount: number }> = []
@@ -91,7 +130,7 @@ export const POST = withRouteHandler(
         } else {
           failures.push({
             serverId,
-            error: result.reason instanceof Error ? result.reason.message : 'Unknown error',
+            error: getErrorMessage(result.reason, 'Unknown error'),
           })
         }
       })
@@ -107,6 +146,8 @@ export const POST = withRouteHandler(
         },
       })
     } catch (error) {
+      const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+      if (bodyErrorResponse) return bodyErrorResponse
       if (
         error instanceof McpOauthAuthorizationRequiredError ||
         error instanceof UnauthorizedError

apps/sim/app/api/mcp/tools/execute/route.ts

@@ -8,7 +8,11 @@ import { getExecutionTimeout } from '@/lib/core/execution-limits'
 import type { SubscriptionPlan } from '@/lib/core/rate-limiter/types'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { SIM_VIA_HEADER } from '@/lib/execution/call-chain'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { McpOauthRedirectRequired } from '@/lib/mcp/oauth'
 import { mcpService } from '@/lib/mcp/service'
 import {
@@ -53,7 +57,7 @@ export const POST = withRouteHandler(
   withMcpAuth('read')(async (request: NextRequest, { userId, workspaceId, requestId }) => {
     let serverId: string | undefined
     try {
-      const rawBody = getParsedBody(request) ?? (await request.json())
+      const rawBody = await readMcpJsonBodyWithLimit(request)
       const parsedBody = mcpToolExecutionBodySchema.safeParse(rawBody)
 
       if (!parsedBody.success) {
@@ -235,6 +239,8 @@ export const POST = withRouteHandler(
 
       return createMcpSuccessResponse(transformedResult)
     } catch (error) {
+      const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+      if (bodyErrorResponse) return bodyErrorResponse
       if (
         error instanceof McpOauthAuthorizationRequiredError ||
         error instanceof McpOauthRedirectRequired ||

apps/sim/app/api/mcp/workflow-servers/[id]/route.ts

@@ -9,7 +9,11 @@ import {
   workflowMcpServerParamsSchema,
 } from '@/lib/api/contracts/workflow-mcp-servers'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import {
   performDeleteWorkflowMcpServer,
   performUpdateWorkflowMcpServer,
@@ -94,7 +98,7 @@ export const PATCH = withRouteHandler(
     ) => {
       try {
         const { id: serverId } = workflowMcpServerParamsSchema.parse(await params)
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = updateWorkflowMcpServerBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -130,6 +134,8 @@ export const PATCH = withRouteHandler(
 
         return createMcpSuccessResponse({ server: updatedServer })
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error updating workflow MCP server:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to update workflow MCP server', 500)
       }

apps/sim/app/api/mcp/workflow-servers/[id]/tools/[toolId]/route.ts

@@ -9,9 +9,17 @@ import {
   workflowMcpToolParamsSchema,
 } from '@/lib/api/contracts/workflow-mcp-servers'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { performDeleteWorkflowMcpTool, performUpdateWorkflowMcpTool } from '@/lib/mcp/orchestration'
-import { createMcpErrorResponse, createMcpSuccessResponse } from '@/lib/mcp/utils'
+import {
+  createMcpErrorResponse,
+  createMcpSuccessResponse,
+  mcpOrchestrationStatus,
+} from '@/lib/mcp/utils'
 
 const logger = createLogger('WorkflowMcpToolAPI')
 
@@ -86,7 +94,7 @@ export const PATCH = withRouteHandler(
     ) => {
       try {
         const { id: serverId, toolId } = workflowMcpToolParamsSchema.parse(await params)
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = updateWorkflowMcpToolBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -109,12 +117,10 @@ export const PATCH = withRouteHandler(
           parameterSchema: body.parameterSchema,
         })
         if (!result.success || !result.tool) {
-          const status =
-            result.errorCode === 'not_found' ? 404 : result.errorCode === 'validation' ? 400 : 500
           return createMcpErrorResponse(
             new Error(result.error || 'Failed to update tool'),
             result.error || 'Failed to update tool',
-            status
+            mcpOrchestrationStatus(result.errorCode)
           )
         }
 
@@ -124,6 +130,8 @@ export const PATCH = withRouteHandler(
 
         return createMcpSuccessResponse({ tool: updatedTool })
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error updating tool:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to update tool', 500)
       }
@@ -158,7 +166,7 @@ export const DELETE = withRouteHandler(
           return createMcpErrorResponse(
             new Error(result.error || 'Tool not found'),
             result.error || 'Tool not found',
-            result.errorCode === 'not_found' ? 404 : 500
+            mcpOrchestrationStatus(result.errorCode)
           )
         }
         const deletedTool = result.tool

apps/sim/app/api/mcp/workflow-servers/[id]/tools/route.ts

@@ -9,7 +9,11 @@ import {
   workflowMcpServerParamsSchema,
 } from '@/lib/api/contracts/workflow-mcp-servers'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { performCreateWorkflowMcpTool } from '@/lib/mcp/orchestration'
 import {
   createMcpErrorResponse,
@@ -96,7 +100,7 @@ export const POST = withRouteHandler(
     ) => {
       try {
         const { id: serverId } = workflowMcpServerParamsSchema.parse(await params)
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = createWorkflowMcpToolBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -136,6 +140,8 @@ export const POST = withRouteHandler(
 
         return createMcpSuccessResponse({ tool }, 201)
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error adding tool:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to add tool', 500)
       }

apps/sim/app/api/mcp/workflow-servers/route.ts

@@ -6,7 +6,11 @@ import { and, eq, inArray, isNull, sql } from 'drizzle-orm'
 import type { NextRequest } from 'next/server'
 import { createWorkflowMcpServerBodySchema } from '@/lib/api/contracts/workflow-mcp-servers'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getParsedBody, withMcpAuth } from '@/lib/mcp/middleware'
+import {
+  mcpBodyReadErrorResponse,
+  readMcpJsonBodyWithLimit,
+  withMcpAuth,
+} from '@/lib/mcp/middleware'
 import { performCreateWorkflowMcpServer } from '@/lib/mcp/orchestration'
 import {
   createMcpErrorResponse,
@@ -96,7 +100,7 @@ export const POST = withRouteHandler(
   withMcpAuth('write')(
     async (request: NextRequest, { userId, userName, userEmail, workspaceId, requestId }) => {
       try {
-        const rawBody = getParsedBody(request) ?? (await request.json())
+        const rawBody = await readMcpJsonBodyWithLimit(request)
         const parsedBody = createWorkflowMcpServerBodySchema.safeParse(rawBody)
 
         if (!parsedBody.success) {
@@ -138,6 +142,8 @@ export const POST = withRouteHandler(
 
         return createMcpSuccessResponse({ server, addedTools }, 201)
       } catch (error) {
+        const bodyErrorResponse = mcpBodyReadErrorResponse(error, request)
+        if (bodyErrorResponse) return bodyErrorResponse
         logger.error(`[${requestId}] Error creating workflow MCP server:`, error)
         return createMcpErrorResponse(toError(error), 'Failed to create workflow MCP server', 500)
       }