fix(provieders): support parallel agent tool calls, consolidate utils

Author: waleedlatif1

apps/sim/providers/anthropic/index.ts

@@ -303,6 +303,11 @@ export const azureOpenAIProvider: ProviderConfig = {
       usedForcedTools = firstCheckResult.usedForcedTools
 
       while (iterationCount < MAX_TOOL_ITERATIONS) {
+        // Extract text content FIRST, before checking for tool calls
+        if (currentResponse.choices[0]?.message?.content) {
+          content = currentResponse.choices[0].message.content
+        }
+
         // Check for tool calls
         const toolCallsInResponse = currentResponse.choices[0]?.message?.tool_calls
         if (!toolCallsInResponse || toolCallsInResponse.length === 0) {
@@ -316,85 +321,111 @@ export const azureOpenAIProvider: ProviderConfig = {
         // Track time for tool calls in this batch
         const toolsStartTime = Date.now()
 
-        // Process each tool call
-        for (const toolCall of toolCallsInResponse) {
+        // Execute all tool calls in parallel using Promise.allSettled for resilience
+        const toolExecutionPromises = toolCallsInResponse.map(async (toolCall) => {
+          const toolCallStartTime = Date.now()
+          const toolName = toolCall.function.name
+
           try {
-            const toolName = toolCall.function.name
             const toolArgs = JSON.parse(toolCall.function.arguments)
-
-            // Get the tool from the tools registry
             const tool = request.tools?.find((t) => t.id === toolName)
-            if (!tool) continue
 
-            // Execute the tool
-            const toolCallStartTime = Date.now()
+            if (!tool) return null
 
             const { toolParams, executionParams } = prepareToolExecution(tool, toolArgs, request)
-
             const result = await executeTool(toolName, executionParams, true)
             const toolCallEndTime = Date.now()
-            const toolCallDuration = toolCallEndTime - toolCallStartTime
 
-            // Add to time segments for both success and failure
-            timeSegments.push({
-              type: 'tool',
-              name: toolName,
+            return {
+              toolCall,
+              toolName,
+              toolParams,
+              result,
               startTime: toolCallStartTime,
               endTime: toolCallEndTime,
-              duration: toolCallDuration,
-            })
-
-            // Prepare result content for the LLM
-            let resultContent: any
-            if (result.success) {
-              toolResults.push(result.output)
-              resultContent = result.output
-            } else {
-              // Include error information so LLM can respond appropriately
-              resultContent = {
-                error: true,
-                message: result.error || 'Tool execution failed',
-                tool: toolName,
-              }
+              duration: toolCallEndTime - toolCallStartTime,
             }
-
-            toolCalls.push({
-              name: toolName,
-              arguments: toolParams,
-              startTime: new Date(toolCallStartTime).toISOString(),
-              endTime: new Date(toolCallEndTime).toISOString(),
-              duration: toolCallDuration,
-              result: resultContent,
-              success: result.success,
-            })
-
-            // Add the tool call and result to messages (both success and failure)
-            currentMessages.push({
-              role: 'assistant',
-              content: null,
-              tool_calls: [
-                {
-                  id: toolCall.id,
-                  type: 'function',
-                  function: {
-                    name: toolName,
-                    arguments: toolCall.function.arguments,
-                  },
-                },
-              ],
-            })
-
-            currentMessages.push({
-              role: 'tool',
-              tool_call_id: toolCall.id,
-              content: JSON.stringify(resultContent),
-            })
           } catch (error) {
-            logger.error('Error processing tool call:', {
-              error,
-              toolName: toolCall?.function?.name,
-            })
+            const toolCallEndTime = Date.now()
+            logger.error('Error processing tool call:', { error, toolName })
+
+            return {
+              toolCall,
+              toolName,
+              toolParams: {},
+              result: {
+                success: false,
+                output: undefined,
+                error: error instanceof Error ? error.message : 'Tool execution failed',
+              },
+              startTime: toolCallStartTime,
+              endTime: toolCallEndTime,
+              duration: toolCallEndTime - toolCallStartTime,
+            }
+          }
+        })
+
+        const executionResults = await Promise.allSettled(toolExecutionPromises)
+
+        // Add ONE assistant message with ALL tool calls BEFORE processing results
+        currentMessages.push({
+          role: 'assistant',
+          content: null,
+          tool_calls: toolCallsInResponse.map((tc) => ({
+            id: tc.id,
+            type: 'function',
+            function: {
+              name: tc.function.name,
+              arguments: tc.function.arguments,
+            },
+          })),
+        })
+
+        // Process results in order to maintain consistency
+        for (const settledResult of executionResults) {
+          if (settledResult.status === 'rejected' || !settledResult.value) continue
+
+          const { toolCall, toolName, toolParams, result, startTime, endTime, duration } =
+            settledResult.value
+
+          // Add to time segments
+          timeSegments.push({
+            type: 'tool',
+            name: toolName,
+            startTime: startTime,
+            endTime: endTime,
+            duration: duration,
+          })
+
+          // Prepare result content for the LLM
+          let resultContent: any
+          if (result.success) {
+            toolResults.push(result.output)
+            resultContent = result.output
+          } else {
+            resultContent = {
+              error: true,
+              message: result.error || 'Tool execution failed',
+              tool: toolName,
+            }
           }
+
+          toolCalls.push({
+            name: toolName,
+            arguments: toolParams,
+            startTime: new Date(startTime).toISOString(),
+            endTime: new Date(endTime).toISOString(),
+            duration: duration,
+            result: resultContent,
+            success: result.success,
+          })
+
+          // Add tool result message
+          currentMessages.push({
+            role: 'tool',
+            tool_call_id: toolCall.id,
+            content: JSON.stringify(resultContent),
+          })
         }
 
         // Calculate tool call time for this iteration

apps/sim/providers/azure-openai/index.ts

@@ -1,82 +1,37 @@
 import type { ChatCompletionChunk } from 'openai/resources/chat/completions'
 import type { CompletionUsage } from 'openai/resources/completions'
 import type { Stream } from 'openai/streaming'
-import { createLogger, type Logger } from '@/lib/logs/console/logger'
-import { trackForcedToolUsage } from '@/providers/utils'
-
-const logger = createLogger('AzureOpenAIUtils')
+import type { Logger } from '@/lib/logs/console/logger'
+import { checkForForcedToolUsageOpenAI, createOpenAICompatibleStream } from '@/providers/utils'
 
+/**
+ * Creates a ReadableStream from an Azure OpenAI streaming response.
+ * Uses the shared OpenAI-compatible streaming utility.
+ */
 export function createReadableStreamFromAzureOpenAIStream(
   azureOpenAIStream: Stream<ChatCompletionChunk>,
   onComplete?: (content: string, usage: CompletionUsage) => void
 ): ReadableStream {
-  let fullContent = ''
-  let promptTokens = 0
-  let completionTokens = 0
-  let totalTokens = 0
-
-  return new ReadableStream({
-    async start(controller) {
-      try {
-        for await (const chunk of azureOpenAIStream) {
-          if (chunk.usage) {
-            promptTokens = chunk.usage.prompt_tokens ?? 0
-            completionTokens = chunk.usage.completion_tokens ?? 0
-            totalTokens = chunk.usage.total_tokens ?? 0
-          }
-
-          const content = chunk.choices[0]?.delta?.content || ''
-          if (content) {
-            fullContent += content
-            controller.enqueue(new TextEncoder().encode(content))
-          }
-        }
-
-        if (onComplete) {
-          if (promptTokens === 0 && completionTokens === 0) {
-            logger.warn('Azure OpenAI stream completed without usage data')
-          }
-          onComplete(fullContent, {
-            prompt_tokens: promptTokens,
-            completion_tokens: completionTokens,
-            total_tokens: totalTokens || promptTokens + completionTokens,
-          })
-        }
-
-        controller.close()
-      } catch (error) {
-        controller.error(error)
-      }
-    },
-  })
+  return createOpenAICompatibleStream(azureOpenAIStream, 'Azure OpenAI', onComplete)
 }
 
 /**
- * Helper function to check for forced tool usage in responses
+ * Checks if a forced tool was used in an Azure OpenAI response.
+ * Uses the shared OpenAI-compatible forced tool usage helper.
  */
 export function checkForForcedToolUsage(
   response: any,
   toolChoice: string | { type: string; function?: { name: string }; name?: string; any?: any },
-  logger: Logger,
+  _logger: Logger,
   forcedTools: string[],
   usedForcedTools: string[]
 ): { hasUsedForcedTool: boolean; usedForcedTools: string[] } {
-  let hasUsedForcedTool = false
-  let updatedUsedForcedTools = [...usedForcedTools]
-
-  if (typeof toolChoice === 'object' && response.choices[0]?.message?.tool_calls) {
-    const toolCallsResponse = response.choices[0].message.tool_calls
-    const result = trackForcedToolUsage(
-      toolCallsResponse,
-      toolChoice,
-      logger,
-      'azure-openai',
-      forcedTools,
-      updatedUsedForcedTools
-    )
-    hasUsedForcedTool = result.hasUsedForcedTool
-    updatedUsedForcedTools = result.usedForcedTools
-  }
-
-  return { hasUsedForcedTool, usedForcedTools: updatedUsedForcedTools }
+  return checkForForcedToolUsageOpenAI(
+    response,
+    toolChoice,
+    'Azure OpenAI',
+    forcedTools,
+    usedForcedTools,
+    _logger
+  )
 }