fix(workflows): keep long-running calls active

Author: waleedlatif1

apps/docs/openapi-v2-workflows.json

@@ -2502,7 +2502,7 @@
       "post": {
         "operationId": "executeWorkflowV2",
         "summary": "Execute Workflow",
-        "description": "Execute the deployment; `run.source: \"manual\"` uses draft state. Manual runs require a personal key or OAuth write access; workspace keys, anonymous callers, and async are rejected. Start at a runnable trigger, or resume from `sourceRunId` using the same-workflow snapshot. Public deployments allow anonymous sync or streaming; async requires credentials. Sync timeouts return `200` with failed status and `TIMEOUT`. Each option carries the modes it requires and the modes that reject it; a violated combination is a 400.\n\nOAuth scope: `api:write`.",
+        "description": "Execute a deployment or use `run.source: \"manual\"` for draft state. Manual runs require personal or OAuth write access and reject workspace keys, anonymous callers, and async. Start at a trigger or resume from same-workflow `sourceRunId`. Public deployments allow anonymous sync or streaming. Request `application/x-ndjson` for 15-second heartbeats and final resource. Timeouts return `200` with failed status and `TIMEOUT`. Each option carries the modes it requires and the modes that reject it; a violated combination is a 400.\n\nOAuth scope: `api:write`.",
         "x-sim-operation": "workflows.execute",
         "x-oauth-scope": "api:write",
         "tags": ["Workflows"],
@@ -2566,7 +2566,7 @@
         },
         "responses": {
           "200": {
-            "description": "A synchronous run result or Server-Sent Event stream.",
+            "description": "A synchronous run result, heartbeat-delimited NDJSON result stream, or Server-Sent Event stream.",
             "headers": {
               "X-Run-Id": {
                 "$ref": "#/components/headers/X-Run-Id"
@@ -2587,6 +2587,11 @@
                   "$ref": "#/components/schemas/ExecuteWorkflowSyncResponse"
                 }
               },
+              "application/x-ndjson": {
+                "schema": {
+                  "type": "string"
+                }
+              },
               "text/event-stream": {
                 "schema": {
                   "type": "string"

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

@@ -95,7 +95,7 @@ vi.mock('@/lib/auth/internal', () => ({
 }))
 
 vi.mock('@/lib/core/execution-limits', () => ({
-  getMaxExecutionTimeout: () => 10_000,
+  getMaxExecutionTimeout: () => 60_000,
 }))
 
 vi.mock('@/lib/workflows/executor/execute-service', () => ({
@@ -338,6 +338,130 @@ describe('MCP Serve Route', () => {
     })
   })
 
+  it('keeps a Streamable HTTP tool call active and ends with its JSON-RPC response', async () => {
+    vi.useFakeTimers()
+    try {
+      dbChainMockFns.limit
+        .mockResolvedValueOnce([
+          {
+            id: 'server-1',
+            name: 'Public Server',
+            workspaceId: 'ws-1',
+            isPublic: true,
+            createdBy: 'owner-1',
+          },
+        ])
+        .mockResolvedValueOnce([{ toolName: 'tool_a', workflowId: 'wf-1' }])
+        .mockResolvedValueOnce([{ workspaceId: 'ws-1', deploymentVersionId: 'deployment-1' }])
+
+      let finishExecution!: (result: unknown) => void
+      mockExecuteWorkflowService.mockReturnValueOnce(
+        new Promise((resolve) => {
+          finishExecution = resolve
+        })
+      )
+
+      const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+        method: 'POST',
+        headers: { accept: 'application/json, text/event-stream' },
+        body: JSON.stringify({
+          jsonrpc: '2.0',
+          id: 1,
+          method: 'tools/call',
+          params: { name: 'tool_a', arguments: { q: 'test' } },
+        }),
+      })
+      const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
+
+      expect(response.status).toBe(200)
+      expect(response.headers.get('content-type')).toContain('text/event-stream')
+      if (!response.body) throw new Error('Expected MCP event stream')
+      const reader = response.body.getReader()
+      const decoder = new TextDecoder()
+      expect(decoder.decode((await reader.read()).value)).toBe(': keepalive\n\n')
+      await vi.advanceTimersByTimeAsync(15_000)
+      expect(decoder.decode((await reader.read()).value)).toBe(': keepalive\n\n')
+
+      finishExecution({
+        ok: true,
+        executionId: 'exec-1',
+        workflowId: 'wf-1',
+        status: 'completed',
+        aborted: null,
+        output: { ok: true },
+        error: null,
+        hasResponseBlock: false,
+        resolvedSecretTraceProvenance: createResolvedSecretTraceProvenance('owner-1'),
+      })
+
+      const event = decoder.decode((await reader.read()).value)
+      expect(JSON.parse(event.replace(/^data: /, '').trim())).toMatchObject({
+        jsonrpc: '2.0',
+        id: 1,
+        result: { content: [{ type: 'text' }], isError: false },
+      })
+      expect((await reader.read()).done).toBe(true)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('cancels the workflow when an MCP event-stream consumer disconnects', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([
+        {
+          id: 'server-1',
+          name: 'Public Server',
+          workspaceId: 'ws-1',
+          isPublic: true,
+          createdBy: 'owner-1',
+        },
+      ])
+      .mockResolvedValueOnce([{ toolName: 'tool_a', workflowId: 'wf-1' }])
+      .mockResolvedValueOnce([{ workspaceId: 'ws-1', deploymentVersionId: 'deployment-1' }])
+
+    let executionSignal: AbortSignal | undefined
+    mockExecuteWorkflowService.mockImplementationOnce(
+      ({ abortSignal }: { abortSignal: AbortSignal }) =>
+        new Promise((resolve) => {
+          executionSignal = abortSignal
+          const finish = () =>
+            resolve({
+              ok: true,
+              executionId: 'exec-1',
+              workflowId: 'wf-1',
+              status: 'cancelled',
+              aborted: 'client',
+              output: undefined,
+              error: { message: 'Client cancelled request', code: 'CANCELLED' },
+              hasResponseBlock: false,
+            })
+          if (abortSignal.aborted) finish()
+          else abortSignal.addEventListener('abort', finish, { once: true })
+        })
+    )
+
+    const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+      method: 'POST',
+      headers: { accept: 'application/json, text/event-stream' },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 1,
+        method: 'tools/call',
+        params: { name: 'tool_a', arguments: { q: 'test' } },
+      }),
+    })
+    const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
+    if (!response.body) throw new Error('Expected MCP event stream')
+    const reader = response.body.getReader()
+    await reader.read()
+    await vi.waitFor(() => expect(executionSignal).toBeDefined())
+
+    await reader.cancel('client disconnected')
+
+    expect(executionSignal?.aborted).toBe(true)
+  })
+
   it('rejects a personal api key when the workspace disallows personal api keys', async () => {
     dbChainMockFns.limit.mockResolvedValueOnce([
       {

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

@@ -27,6 +27,7 @@ import {
   workspace,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
 import { and, asc, eq, gt, isNull, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
@@ -45,6 +46,7 @@ import {
 } from '@/lib/billing/core/billing-attribution'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { generateRequestId } from '@/lib/core/utils/request'
+import { encodeSSE, encodeSSEComment, SSE_HEADERS } from '@/lib/core/utils/sse'
 import {
   assertContentLengthWithinLimit,
   assertKnownSizeWithinLimit,
@@ -75,6 +77,7 @@ const MAX_MCP_WORKFLOW_REQUEST_BYTES = 10 * 1024 * 1024
 const MAX_MCP_TOOL_RESULT_TEXT_BYTES = 10 * 1024 * 1024
 const MAX_MCP_TOOLS_LIST_COUNT = MAX_MCP_TOOLS_PER_SERVER
 const MAX_MCP_TOOLS_LIST_SCHEMA_BYTES = MAX_MCP_PARAMETER_SCHEMA_BYTES
+const MCP_STREAM_KEEPALIVE_INTERVAL_MS = 15_000
 const MB = 1024 * 1024
 
 function negotiateProtocolVersion(rpcParams: unknown): string {
@@ -137,6 +140,95 @@ function callerAbortedJsonRpcResponse(
   return abortSignal?.isCallerAborted() ? clientCancelledJsonRpcResponse(id) : null
 }
 
+function acceptsEventStream(request: NextRequest): boolean {
+  return request.headers.get('accept')?.includes('text/event-stream') === true
+}
+
+/**
+ * Sends a Streamable HTTP response as SSE so a long tool call can keep the
+ * connection active before its terminal JSON-RPC message is available.
+ */
+function streamJsonRpcResponse(
+  id: RequestId,
+  requestSignal: AbortSignal,
+  run: (signal: AbortSignal) => Promise<NextResponse>
+): Response {
+  const executionController = new AbortController()
+  let cancelled = false
+  let keepaliveId: ReturnType<typeof setInterval> | undefined
+
+  const stopKeepalive = () => {
+    if (keepaliveId) {
+      clearInterval(keepaliveId)
+      keepaliveId = undefined
+    }
+  }
+  const abortExecution = (reason?: unknown) => {
+    if (!executionController.signal.aborted) {
+      executionController.abort(reason ?? new Error('MCP client disconnected'))
+    }
+  }
+  const abortFromRequest = () => abortExecution(requestSignal.reason)
+
+  if (requestSignal.aborted) {
+    abortFromRequest()
+  } else {
+    requestSignal.addEventListener('abort', abortFromRequest, { once: true })
+  }
+
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      const send = (chunk: Uint8Array): boolean => {
+        if (cancelled) return false
+        try {
+          controller.enqueue(chunk)
+          return true
+        } catch {
+          cancelled = true
+          stopKeepalive()
+          abortExecution()
+          return false
+        }
+      }
+
+      if (send(encodeSSEComment('keepalive'))) {
+        keepaliveId = setInterval(() => {
+          send(encodeSSEComment('keepalive'))
+        }, MCP_STREAM_KEEPALIVE_INTERVAL_MS)
+      }
+
+      void run(executionController.signal)
+        .then(async (response) => {
+          const message: unknown = await response.json()
+          send(encodeSSE(message))
+        })
+        .catch((error) => {
+          logger.error('MCP response stream failed', { error: getErrorMessage(error) })
+          send(encodeSSE(createError(id, ErrorCode.InternalError, 'Internal error')))
+        })
+        .finally(() => {
+          stopKeepalive()
+          requestSignal.removeEventListener('abort', abortFromRequest)
+          if (!cancelled) controller.close()
+        })
+    },
+    cancel(reason) {
+      cancelled = true
+      stopKeepalive()
+      requestSignal.removeEventListener('abort', abortFromRequest)
+      abortExecution(reason)
+    },
+  })
+
+  return new Response(stream, {
+    headers: {
+      ...SSE_HEADERS,
+      'Cache-Control': 'no-cache, no-transform',
+      Vary: 'Accept',
+    },
+  })
+}
+
 function limitMessage(label: string, maxBytes: number): string {
   return `${label} exceeds maximum size of ${Math.round(maxBytes / MB)}MB`
 }
@@ -406,12 +498,12 @@ async function authorizeMcpServeRequest(
   }
 }
 
-function unsupportedSseTransportResponse(): NextResponse {
+function unsupportedSseGetResponse(): NextResponse {
   return NextResponse.json(
     {
       error: {
         code: 'unsupported_transport',
-        message: 'SSE transport is not supported for workflow MCP servers',
+        message: 'Standalone SSE GET transport is not supported for workflow MCP servers',
         supportedTransports: ['streamable-http'],
         allowedMethods: ['GET', 'POST', 'DELETE'],
       },
@@ -439,7 +531,7 @@ export const GET = withRouteHandler(
       if (authResult.response) return authResult.response
 
       if (request.headers.get('accept')?.includes('text/event-stream')) {
-        return unsupportedSseTransportResponse()
+        return unsupportedSseGetResponse()
       }
 
       return NextResponse.json({
@@ -557,16 +649,22 @@ export const POST = withRouteHandler(
             )
           }
 
-          return handleToolsCall(
-            id,
-            serverId,
-            server.workspaceId,
-            paramsValidation.data,
-            executeAuthContext,
-            server.isPublic ? server.createdBy : undefined,
-            request.headers.get(SIM_VIA_HEADER),
-            request.signal
-          )
+          const callTool = (signal: AbortSignal) =>
+            handleToolsCall(
+              id,
+              serverId,
+              server.workspaceId,
+              paramsValidation.data,
+              executeAuthContext,
+              server.isPublic ? server.createdBy : undefined,
+              request.headers.get(SIM_VIA_HEADER),
+              signal
+            )
+
+          if (acceptsEventStream(request)) {
+            return streamJsonRpcResponse(id, request.signal, callTool)
+          }
+          return callTool(request.signal)
         }
 
         default: