wehub-resource-sync
d25d482dc2
Publish CLI Package / publish-npm (push) Waiting to run
Publish Python SDK / publish-pypi (push) Waiting to run
Publish TypeScript SDK / publish-npm (push) Waiting to run
CI / Migrate Dev DB (push) Has been skipped
CI / Detect Version (push) Has been cancelled
CI / Migrate DB (push) Has been cancelled
CI / Build Dev ECR (./docker/app.Dockerfile, ECR_APP) (push) Has been cancelled
CI / Build Dev ECR (./docker/db.Dockerfile, ECR_MIGRATIONS) (push) Has been cancelled
CI / Build Dev ECR (./docker/pii.Dockerfile, ECR_PII) (push) Has been cancelled
CI / Build Dev ECR (./docker/realtime.Dockerfile, ECR_REALTIME) (push) Has been cancelled
CI / Deploy Trigger.dev (Dev) (push) Has been cancelled
CI / Build AMD64 (./docker/app.Dockerfile, ECR_APP, ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Build AMD64 (./docker/db.Dockerfile, ECR_MIGRATIONS, ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Build AMD64 (./docker/pii.Dockerfile, ECR_PII, ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Build AMD64 (./docker/realtime.Dockerfile, ECR_REALTIME, ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/app.Dockerfile, ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/db.Dockerfile, ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/pii.Dockerfile, ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/realtime.Dockerfile, ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Check Docs Changes (push) Has been cancelled
CI / Process Docs (push) Has been cancelled
CI / Create GitHub Release (push) Has been cancelled
CI / Test and Build (push) Has been cancelled
2192 lines
69 KiB
TypeScript
2192 lines
69 KiB
TypeScript
import { createLogger } from '@sim/logger'
|
|
import { getErrorMessage, toError } from '@sim/utils/errors'
|
|
import { sleep } from '@sim/utils/helpers'
|
|
import { backoffWithJitter, parseRetryAfter } from '@sim/utils/retry'
|
|
import { getBYOKKey } from '@/lib/api-key/byok'
|
|
import { generateInternalToken } from '@/lib/auth/internal'
|
|
import { isHosted } from '@/lib/core/config/env-flags'
|
|
import { DEFAULT_EXECUTION_TIMEOUT_MS, getMaxExecutionTimeout } from '@/lib/core/execution-limits'
|
|
import { getHostedKeyRateLimiter } from '@/lib/core/rate-limiter'
|
|
import {
|
|
secureFetchWithPinnedIP,
|
|
validateUrlWithDNS,
|
|
} from '@/lib/core/security/input-validation.server'
|
|
import { PlatformEvents } from '@/lib/core/telemetry'
|
|
import { generateRequestId } from '@/lib/core/utils/request'
|
|
import {
|
|
isPayloadSizeLimitError,
|
|
readResponseToBufferWithLimit,
|
|
} from '@/lib/core/utils/stream-limits'
|
|
import { getBaseUrl, getInternalApiBaseUrl } from '@/lib/core/utils/urls'
|
|
import { isUserFile } from '@/lib/core/utils/user-file'
|
|
import { isSameOrigin } from '@/lib/core/utils/validation'
|
|
import { SIM_VIA_HEADER, serializeCallChain } from '@/lib/execution/call-chain'
|
|
import { parseMcpToolId } from '@/lib/mcp/utils'
|
|
import { hostedKeyMetrics } from '@/lib/monitoring/metrics'
|
|
import { resolveWorkspaceFileReference } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
|
import { assertPermissionsAllowed } from '@/ee/access-control/utils/permission-check'
|
|
import { isCustomTool, isMcpTool } from '@/executor/constants'
|
|
import { resolveSkillContent } from '@/executor/handlers/agent/skills-resolver'
|
|
import type { ExecutionContext, UserFile } from '@/executor/types'
|
|
import type { ErrorInfo } from '@/tools/error-extractors'
|
|
import { extractErrorMessage } from '@/tools/error-extractors'
|
|
import type {
|
|
BYOKProviderId,
|
|
OAuthTokenPayload,
|
|
ToolConfig,
|
|
ToolHostingPricing,
|
|
ToolResponse,
|
|
ToolRetryConfig,
|
|
} from '@/tools/types'
|
|
import { formatRequestParams, getTool, validateRequiredParametersAfterMerge } from '@/tools/utils'
|
|
import * as toolsUtilsServer from '@/tools/utils.server'
|
|
|
|
const logger = createLogger('Tools')
|
|
|
|
interface ToolExecutionScope {
|
|
workspaceId?: string
|
|
workflowId?: string
|
|
userId?: string
|
|
executionId?: string
|
|
callChain?: string[]
|
|
isDeployedContext?: boolean
|
|
enforceCredentialAccess?: boolean
|
|
copilotToolExecution?: boolean
|
|
}
|
|
|
|
function resolveToolScope(
|
|
params: Record<string, unknown>,
|
|
executionContext?: ExecutionContext
|
|
): ToolExecutionScope {
|
|
const ctx = params._context as Record<string, unknown> | undefined
|
|
return {
|
|
workspaceId: (executionContext?.workspaceId ?? ctx?.workspaceId) as string | undefined,
|
|
workflowId: (executionContext?.workflowId ?? ctx?.workflowId) as string | undefined,
|
|
userId: (executionContext?.userId ?? ctx?.userId) as string | undefined,
|
|
executionId: (executionContext?.executionId ?? ctx?.executionId) as string | undefined,
|
|
callChain: (executionContext?.callChain ?? ctx?.callChain) as string[] | undefined,
|
|
isDeployedContext: (executionContext?.isDeployedContext ?? ctx?.isDeployedContext) as
|
|
| boolean
|
|
| undefined,
|
|
enforceCredentialAccess: (executionContext?.enforceCredentialAccess ??
|
|
ctx?.enforceCredentialAccess) as boolean | undefined,
|
|
copilotToolExecution: (executionContext?.copilotToolExecution ?? ctx?.copilotToolExecution) as
|
|
| boolean
|
|
| undefined,
|
|
}
|
|
}
|
|
|
|
function toUserFileFromWorkspaceRecord(record: {
|
|
id: string
|
|
name: string
|
|
path: string
|
|
url?: string
|
|
size: number
|
|
type: string
|
|
key: string
|
|
}): UserFile {
|
|
return {
|
|
id: record.id,
|
|
name: record.name,
|
|
url: record.url ?? record.path,
|
|
size: record.size,
|
|
type: record.type,
|
|
key: record.key,
|
|
context: 'workspace',
|
|
}
|
|
}
|
|
|
|
async function resolveCopilotFileReference(
|
|
value: unknown,
|
|
workspaceId: string,
|
|
paramId: string
|
|
): Promise<UserFile | unknown> {
|
|
if (isUserFile(value)) {
|
|
return value
|
|
}
|
|
|
|
const referenceId =
|
|
typeof value === 'string'
|
|
? value
|
|
: value &&
|
|
typeof value === 'object' &&
|
|
typeof (value as Record<string, unknown>).id === 'string'
|
|
? ((value as Record<string, unknown>).id as string)
|
|
: null
|
|
|
|
if (!referenceId) {
|
|
return value
|
|
}
|
|
|
|
const fileRecord = await resolveWorkspaceFileReference(workspaceId, referenceId)
|
|
if (!fileRecord) {
|
|
throw new Error(
|
|
`Could not resolve workspace file reference "${referenceId}" for parameter "${paramId}"`
|
|
)
|
|
}
|
|
|
|
const resolvedFile = toUserFileFromWorkspaceRecord(fileRecord)
|
|
if (!value || typeof value !== 'object') {
|
|
return resolvedFile
|
|
}
|
|
|
|
const candidate = value as Record<string, unknown>
|
|
return {
|
|
...resolvedFile,
|
|
context: typeof candidate.context === 'string' ? candidate.context : resolvedFile.context,
|
|
base64: typeof candidate.base64 === 'string' ? candidate.base64 : undefined,
|
|
}
|
|
}
|
|
|
|
async function normalizeCopilotFileParams(
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
scope: ToolExecutionScope
|
|
): Promise<void> {
|
|
if (!scope.copilotToolExecution) {
|
|
return
|
|
}
|
|
|
|
for (const [paramId, paramDef] of Object.entries(tool.params || {})) {
|
|
const paramType = paramDef?.type
|
|
const currentValue = params[paramId]
|
|
if (currentValue === undefined || currentValue === null) {
|
|
continue
|
|
}
|
|
|
|
if (paramType === 'file') {
|
|
if (!scope.workspaceId) {
|
|
throw new Error(`Missing workspaceId while resolving file parameter "${paramId}"`)
|
|
}
|
|
params[paramId] = await resolveCopilotFileReference(currentValue, scope.workspaceId, paramId)
|
|
continue
|
|
}
|
|
|
|
if (paramType === 'file[]') {
|
|
if (!scope.workspaceId) {
|
|
throw new Error(`Missing workspaceId while resolving file parameter "${paramId}"`)
|
|
}
|
|
|
|
const values = Array.isArray(currentValue) ? currentValue : [currentValue]
|
|
params[paramId] = await Promise.all(
|
|
values.map((item) => resolveCopilotFileReference(item, scope.workspaceId!, paramId))
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
function readExplicitCredentialSelector(params: Record<string, unknown>): string | undefined {
|
|
for (const key of ['credentialId', 'oauthCredential', 'credential'] as const) {
|
|
const value = params[key]
|
|
if (typeof value === 'string' && value.trim().length > 0) {
|
|
return value.trim()
|
|
}
|
|
}
|
|
return undefined
|
|
}
|
|
|
|
function normalizeCopilotCredentialParams(params: Record<string, unknown>): void {
|
|
const credentialId = typeof params.credentialId === 'string' ? params.credentialId.trim() : ''
|
|
if (credentialId && !params.credential && !params.oauthCredential) {
|
|
params.credential = credentialId
|
|
}
|
|
}
|
|
|
|
function enforceCopilotCredentialSelection(
|
|
toolId: string,
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
scope: ToolExecutionScope
|
|
): void {
|
|
if (!scope.copilotToolExecution || !tool.oauth?.required) {
|
|
return
|
|
}
|
|
|
|
if (readExplicitCredentialSelector(params)) {
|
|
return
|
|
}
|
|
|
|
const toolLabel = tool.name || toolId
|
|
throw new Error(
|
|
`Copilot must pass credentialId for ${toolLabel}. Read environment/credentials.json and pass the exact credentialId for provider "${tool.oauth.provider}".`
|
|
)
|
|
}
|
|
|
|
/** Result from hosted key injection */
|
|
interface HostedKeyInjectionResult {
|
|
isUsingHostedKey: boolean
|
|
envVarName?: string
|
|
}
|
|
|
|
/**
|
|
* Inject hosted API key if tool supports it and user didn't provide one.
|
|
* Checks BYOK workspace keys first, then uses the HostedKeyRateLimiter for round-robin key selection.
|
|
* Returns whether a hosted (billable) key was injected and which env var it came from.
|
|
*/
|
|
async function injectHostedKeyIfNeeded(
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
executionContext: ExecutionContext | undefined,
|
|
requestId: string
|
|
): Promise<HostedKeyInjectionResult> {
|
|
if (!tool.hosting) return { isUsingHostedKey: false }
|
|
if (!isHosted) return { isUsingHostedKey: false }
|
|
if (tool.hosting.enabled && !tool.hosting.enabled(params)) {
|
|
return { isUsingHostedKey: false }
|
|
}
|
|
|
|
const { envKeyPrefix, apiKeyParam, byokProviderId, rateLimit } = tool.hosting
|
|
const userProvidedKey = params[apiKeyParam]
|
|
if (typeof userProvidedKey === 'string' && userProvidedKey.trim().length > 0) {
|
|
return { isUsingHostedKey: false }
|
|
}
|
|
|
|
const { workspaceId, userId, workflowId } = resolveToolScope(params, executionContext)
|
|
|
|
// Check BYOK workspace key first
|
|
if (byokProviderId && workspaceId) {
|
|
try {
|
|
const byokResult = await getBYOKKey(workspaceId, byokProviderId as BYOKProviderId)
|
|
if (byokResult) {
|
|
params[apiKeyParam] = byokResult.apiKey
|
|
logger.info(`[${requestId}] Using BYOK key for ${tool.id}`)
|
|
return { isUsingHostedKey: false } // Don't bill - user's own key
|
|
}
|
|
} catch (error) {
|
|
logger.error(`[${requestId}] Failed to get BYOK key for ${tool.id}:`, error)
|
|
// Fall through to hosted key
|
|
}
|
|
}
|
|
|
|
const rateLimiter = getHostedKeyRateLimiter()
|
|
const provider = byokProviderId || tool.id
|
|
const billingActorId = workspaceId
|
|
|
|
if (!billingActorId) {
|
|
logger.error(`[${requestId}] No workspace ID available for hosted key rate limiting`)
|
|
return { isUsingHostedKey: false }
|
|
}
|
|
|
|
const acquireResult = await rateLimiter.acquireKey(
|
|
provider,
|
|
envKeyPrefix,
|
|
rateLimit,
|
|
billingActorId,
|
|
executionContext?.abortSignal
|
|
)
|
|
|
|
if (!acquireResult.success && acquireResult.billingActorRateLimited) {
|
|
logger.warn(`[${requestId}] Billing actor ${billingActorId} rate limited for ${tool.id}`, {
|
|
provider,
|
|
retryAfterMs: acquireResult.retryAfterMs,
|
|
})
|
|
|
|
PlatformEvents.hostedKeyUserThrottled({
|
|
toolId: tool.id,
|
|
reason: 'billing_actor_limit',
|
|
provider,
|
|
retryAfterMs: acquireResult.retryAfterMs ?? 0,
|
|
userId,
|
|
workspaceId,
|
|
workflowId,
|
|
})
|
|
|
|
const error = new Error(acquireResult.error || `Rate limit exceeded for ${tool.id}`)
|
|
;(error as any).status = 429
|
|
;(error as any).retryAfterMs = acquireResult.retryAfterMs
|
|
throw error
|
|
}
|
|
|
|
// Handle no keys configured (503)
|
|
if (!acquireResult.success) {
|
|
logger.error(`[${requestId}] No hosted keys configured for ${tool.id}: ${acquireResult.error}`)
|
|
const error = new Error(acquireResult.error || `No hosted keys configured for ${tool.id}`)
|
|
;(error as any).status = 503
|
|
throw error
|
|
}
|
|
|
|
params[apiKeyParam] = acquireResult.key
|
|
params.__usingHostedKey = true
|
|
logger.info(`[${requestId}] Using hosted key for ${tool.id} (${acquireResult.envVarName})`, {
|
|
keyIndex: acquireResult.keyIndex,
|
|
provider,
|
|
})
|
|
|
|
return {
|
|
isUsingHostedKey: true,
|
|
envVarName: acquireResult.envVarName,
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Re-acquire a hosted key after upstream-429 retries have been exhausted. Calls
|
|
* `acquireKey` (which now blocks on the per-workspace bucket) and re-injects the
|
|
* fresh key into `params`. Returns false if no key could be obtained — caller
|
|
* should re-throw the original upstream 429.
|
|
*
|
|
* Does not consult BYOK. We only enter this path from inside the hosted-key
|
|
* branch of `executeTool`, so BYOK has already been ruled out for this call.
|
|
*/
|
|
async function reacquireHostedKey(
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
executionContext: ExecutionContext | undefined,
|
|
requestId: string
|
|
): Promise<string | null> {
|
|
if (!tool.hosting) return null
|
|
const { envKeyPrefix, apiKeyParam, byokProviderId, rateLimit } = tool.hosting
|
|
const { workspaceId } = resolveToolScope(params, executionContext)
|
|
if (!workspaceId) return null
|
|
|
|
const provider = byokProviderId || tool.id
|
|
const acquireResult = await getHostedKeyRateLimiter().acquireKey(
|
|
provider,
|
|
envKeyPrefix,
|
|
rateLimit,
|
|
workspaceId,
|
|
executionContext?.abortSignal
|
|
)
|
|
|
|
if (!acquireResult.success || !acquireResult.key) {
|
|
logger.warn(
|
|
`[${requestId}] Re-acquire of hosted key for ${tool.id} failed: ${acquireResult.error ?? 'unknown'}`
|
|
)
|
|
return null
|
|
}
|
|
|
|
params[apiKeyParam] = acquireResult.key
|
|
logger.info(
|
|
`[${requestId}] Re-acquired hosted key for ${tool.id} (${acquireResult.envVarName}) after upstream throttling`
|
|
)
|
|
return acquireResult.envVarName ?? 'unknown'
|
|
}
|
|
|
|
/**
|
|
* Check if an error is a rate limit (throttling) or quota exhaustion error.
|
|
* Some providers (e.g. Perplexity) return 401/403 with "insufficient_quota"
|
|
* instead of the standard 429, so we also inspect the error message.
|
|
*/
|
|
function isRateLimitError(error: unknown): boolean {
|
|
if (error && typeof error === 'object') {
|
|
const status = (error as { status?: number }).status
|
|
if (status === 429 || status === 503) return true
|
|
|
|
if (status === 401 || status === 403) {
|
|
const message = ((error as { message?: string }).message || '').toLowerCase()
|
|
if (message.includes('quota') || message.includes('rate limit')) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
/**
|
|
* Map a thrown tool error to a hosted-key failure reason for metrics. Mirrors
|
|
* `isRateLimitError`: some providers signal quota/rate-limit via 401/403 with a
|
|
* descriptive message, so those count as `rate_limited`, not `auth`.
|
|
*/
|
|
function classifyHostedKeyFailure(error: unknown): 'rate_limited' | 'auth' | 'other' {
|
|
const status = (error as { status?: number } | null)?.status
|
|
if (status === 429 || status === 503) return 'rate_limited'
|
|
if (status === 401 || status === 403) {
|
|
const message = ((error as { message?: string } | null)?.message ?? '').toLowerCase()
|
|
if (message.includes('quota') || message.includes('rate limit')) return 'rate_limited'
|
|
return 'auth'
|
|
}
|
|
return 'other'
|
|
}
|
|
|
|
/** Context for retry with rate limit tracking */
|
|
interface RetryContext {
|
|
requestId: string
|
|
toolId: string
|
|
provider: string
|
|
envVarName: string
|
|
executionContext?: ExecutionContext
|
|
/**
|
|
* Optional callback invoked after the local exponential backoff has been exhausted by
|
|
* upstream 429s. Should re-enter the per-workspace hosted-key queue (which now blocks
|
|
* on the bucket) and return a fresh execution thunk bound to the newly acquired key.
|
|
* If the callback returns null, we give up and re-throw the last error.
|
|
*/
|
|
reacquireAfterRetriesExhausted?: () => Promise<(() => Promise<unknown>) | null>
|
|
}
|
|
|
|
/**
|
|
* Execute a function with exponential backoff retry for rate limiting errors.
|
|
* Only used for hosted key requests. Tracks rate limit events via telemetry.
|
|
*
|
|
* On terminal upstream 429, optionally re-enters the hosted-key queue (which waits for
|
|
* the per-workspace bucket to refill) and retries once with a freshly acquired key.
|
|
* This handles the case where the upstream provider's limit is tighter than ours — we
|
|
* re-queue the call instead of surfacing the error.
|
|
*/
|
|
async function executeWithRetry<T>(
|
|
fn: () => Promise<T>,
|
|
context: RetryContext,
|
|
maxRetries = 3,
|
|
baseDelayMs = 1000
|
|
): Promise<T> {
|
|
const {
|
|
requestId,
|
|
toolId,
|
|
provider,
|
|
envVarName,
|
|
executionContext,
|
|
reacquireAfterRetriesExhausted,
|
|
} = context
|
|
let lastError: unknown
|
|
|
|
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
|
try {
|
|
return await fn()
|
|
} catch (error) {
|
|
lastError = error
|
|
|
|
if (!isRateLimitError(error) || attempt === maxRetries) {
|
|
if (isRateLimitError(error) && attempt === maxRetries) {
|
|
if (reacquireAfterRetriesExhausted) {
|
|
try {
|
|
const requeued = await reacquireAfterRetriesExhausted()
|
|
if (requeued) {
|
|
logger.warn(
|
|
`[${requestId}] Upstream retries exhausted for ${toolId} (${envVarName}); re-queued and retrying once with fresh key`
|
|
)
|
|
return (await requeued()) as T
|
|
}
|
|
} catch (requeueError) {
|
|
logger.error(
|
|
`[${requestId}] Re-queue after exhausted upstream retries failed for ${toolId}`,
|
|
{ error: toError(requeueError).message }
|
|
)
|
|
}
|
|
}
|
|
|
|
PlatformEvents.hostedKeyUserThrottled({
|
|
toolId,
|
|
reason: 'upstream_retries_exhausted',
|
|
provider,
|
|
userId: executionContext?.userId,
|
|
workspaceId: executionContext?.workspaceId,
|
|
workflowId: executionContext?.workflowId,
|
|
})
|
|
}
|
|
throw error
|
|
}
|
|
|
|
const delayMs = backoffWithJitter(attempt + 1, null, { baseMs: baseDelayMs })
|
|
|
|
// Track throttling event via telemetry
|
|
PlatformEvents.hostedKeyRateLimited({
|
|
toolId,
|
|
envVarName,
|
|
attempt: attempt + 1,
|
|
maxRetries,
|
|
delayMs,
|
|
userId: executionContext?.userId,
|
|
workspaceId: executionContext?.workspaceId,
|
|
workflowId: executionContext?.workflowId,
|
|
})
|
|
|
|
logger.warn(
|
|
`[${requestId}] Rate limited for ${toolId} (${envVarName}), retrying in ${delayMs}ms (attempt ${attempt + 1}/${maxRetries})`
|
|
)
|
|
await sleep(delayMs)
|
|
}
|
|
}
|
|
|
|
throw lastError
|
|
}
|
|
|
|
/** Result from cost calculation */
|
|
interface ToolCostResult {
|
|
cost: number
|
|
metadata?: Record<string, unknown>
|
|
}
|
|
|
|
/**
|
|
* Calculate cost based on pricing model
|
|
*/
|
|
function calculateToolCost(
|
|
pricing: ToolHostingPricing,
|
|
params: Record<string, unknown>,
|
|
response: Record<string, unknown>
|
|
): ToolCostResult {
|
|
switch (pricing.type) {
|
|
case 'per_request':
|
|
return { cost: pricing.cost }
|
|
|
|
case 'custom': {
|
|
const result = pricing.getCost(params, response)
|
|
if (typeof result === 'number') {
|
|
return { cost: result }
|
|
}
|
|
return result
|
|
}
|
|
|
|
default: {
|
|
const exhaustiveCheck: never = pricing
|
|
throw new Error(`Unknown pricing type: ${(exhaustiveCheck as ToolHostingPricing).type}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
interface HostedKeyCostResult {
|
|
cost: number
|
|
metadata?: Record<string, unknown>
|
|
}
|
|
|
|
/**
|
|
* Calculate and log hosted key cost for a tool execution.
|
|
* Logs to usageLog for audit trail and returns cost + metadata for output.
|
|
*/
|
|
async function processHostedKeyCost(
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
response: Record<string, unknown>,
|
|
executionContext: ExecutionContext | undefined,
|
|
requestId: string
|
|
): Promise<HostedKeyCostResult> {
|
|
if (!tool.hosting?.pricing) {
|
|
return { cost: 0 }
|
|
}
|
|
|
|
const { cost, metadata } = calculateToolCost(tool.hosting.pricing, params, response)
|
|
|
|
if (cost <= 0) return { cost: 0 }
|
|
|
|
const { userId } = resolveToolScope(params, executionContext)
|
|
|
|
if (!userId) return { cost, metadata }
|
|
|
|
logger.debug(
|
|
`[${requestId}] Hosted key cost for ${tool.id}: $${cost}`,
|
|
metadata ? { metadata } : {}
|
|
)
|
|
|
|
return { cost, metadata }
|
|
}
|
|
|
|
/**
|
|
* Report custom dimension usage after successful hosted-key tool execution.
|
|
* Only applies to tools with `custom` rate limit mode. Fires and logs;
|
|
* failures here do not block the response since execution already succeeded.
|
|
*/
|
|
async function reportCustomDimensionUsage(
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
response: Record<string, unknown>,
|
|
executionContext: ExecutionContext | undefined,
|
|
requestId: string
|
|
): Promise<void> {
|
|
if (tool.hosting?.rateLimit.mode !== 'custom') return
|
|
const { workspaceId: billingActorId } = resolveToolScope(params, executionContext)
|
|
if (!billingActorId) return
|
|
|
|
const rateLimiter = getHostedKeyRateLimiter()
|
|
const provider = tool.hosting.byokProviderId || tool.id
|
|
|
|
try {
|
|
const result = await rateLimiter.reportUsage(
|
|
provider,
|
|
billingActorId,
|
|
tool.hosting.rateLimit,
|
|
params,
|
|
response
|
|
)
|
|
|
|
for (const dim of result.dimensions) {
|
|
if (!dim.allowed) {
|
|
logger.warn(`[${requestId}] Dimension ${dim.name} overdrawn after ${tool.id} execution`, {
|
|
consumed: dim.consumed,
|
|
tokensRemaining: dim.tokensRemaining,
|
|
})
|
|
}
|
|
}
|
|
} catch (error) {
|
|
logger.error(`[${requestId}] Failed to report custom dimension usage for ${tool.id}:`, error)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Strips internal fields (keys starting with `__`) from tool output before
|
|
* returning to users. The double-underscore prefix is reserved for transient
|
|
* data (e.g. `__costDollars`) and will never collide with legitimate API
|
|
* fields like `_id`.
|
|
*/
|
|
function stripInternalFields(output: Record<string, unknown>): Record<string, unknown> {
|
|
if (typeof output !== 'object' || output === null || Array.isArray(output)) {
|
|
return output
|
|
}
|
|
const result: Record<string, unknown> = {}
|
|
for (const [key, value] of Object.entries(output)) {
|
|
if (!key.startsWith('__')) {
|
|
result[key] = value
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
export function postProcessToolOutput(toolId: string, output: Record<string, unknown>) {
|
|
return isCustomTool(toolId) ? output : stripInternalFields(output)
|
|
}
|
|
|
|
/**
|
|
* Apply post-execution hosted-key cost tracking to a successful tool result.
|
|
* Reports custom dimension usage, calculates cost, and merges it into the output.
|
|
*/
|
|
async function applyHostedKeyCostToResult(
|
|
finalResult: ToolResponse,
|
|
tool: ToolConfig,
|
|
params: Record<string, unknown>,
|
|
executionContext: ExecutionContext | undefined,
|
|
requestId: string,
|
|
envVarName: string | undefined
|
|
): Promise<void> {
|
|
await reportCustomDimensionUsage(tool, params, finalResult.output, executionContext, requestId)
|
|
|
|
const { cost: hostedKeyCost, metadata } = await processHostedKeyCost(
|
|
tool,
|
|
params,
|
|
finalResult.output,
|
|
executionContext,
|
|
requestId
|
|
)
|
|
|
|
const provider = tool.hosting?.byokProviderId || tool.id
|
|
const key = envVarName ?? 'unknown'
|
|
hostedKeyMetrics.recordUsed({ provider, tool: tool.id, key })
|
|
hostedKeyMetrics.recordCostCharged(hostedKeyCost, { provider, tool: tool.id })
|
|
|
|
if (hostedKeyCost > 0) {
|
|
finalResult.output = {
|
|
...finalResult.output,
|
|
cost: {
|
|
...metadata,
|
|
total: hostedKeyCost,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
import { normalizeToolId } from '@/tools/normalize'
|
|
|
|
/**
|
|
* Maximum request body size in bytes before we warn/error about size limits.
|
|
* Next.js 16 has a default middleware/proxy body limit of 10MB.
|
|
*/
|
|
const MAX_REQUEST_BODY_SIZE_BYTES = 10 * 1024 * 1024 // 10MB
|
|
const MAX_TOOL_RESPONSE_BODY_BYTES = 10 * 1024 * 1024 // 10MB
|
|
|
|
/**
|
|
* User-friendly error message for body size limit exceeded
|
|
*/
|
|
const BODY_SIZE_LIMIT_ERROR_MESSAGE =
|
|
'Request body size limit exceeded (10MB). The workflow data is too large to process. Try reducing the size of variables, inputs, or data being passed between blocks.'
|
|
|
|
const RESPONSE_SIZE_LIMIT_ERROR_MESSAGE =
|
|
'Tool response size limit exceeded (10MB). The response is too large to keep in workflow data. Reduce the response size or return a file reference instead.'
|
|
|
|
/**
|
|
* Validates request body size and throws a user-friendly error if exceeded
|
|
* @param body - The request body string to check
|
|
* @param requestId - Request ID for logging
|
|
* @param context - Context string for logging (e.g., toolId)
|
|
* @throws Error if body size exceeds the limit
|
|
*/
|
|
function validateRequestBodySize(
|
|
body: string | undefined,
|
|
requestId: string,
|
|
context: string
|
|
): void {
|
|
if (!body) return
|
|
|
|
const bodySize = Buffer.byteLength(body, 'utf8')
|
|
if (bodySize > MAX_REQUEST_BODY_SIZE_BYTES) {
|
|
const bodySizeMB = (bodySize / (1024 * 1024)).toFixed(2)
|
|
const maxSizeMB = (MAX_REQUEST_BODY_SIZE_BYTES / (1024 * 1024)).toFixed(0)
|
|
logger.error(`[${requestId}] Request body size exceeds limit for ${context}:`, {
|
|
bodySize,
|
|
bodySizeMB: `${bodySizeMB}MB`,
|
|
maxSize: MAX_REQUEST_BODY_SIZE_BYTES,
|
|
maxSizeMB: `${maxSizeMB}MB`,
|
|
})
|
|
throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Checks if an error message indicates a body size limit issue
|
|
* @param errorMessage - The error message to check
|
|
* @returns true if the error is related to body size limits
|
|
*/
|
|
function isBodySizeLimitError(errorMessage: string): boolean {
|
|
const lowerMessage = errorMessage.toLowerCase()
|
|
return (
|
|
lowerMessage.includes('body size') ||
|
|
lowerMessage.includes('payload too large') ||
|
|
lowerMessage.includes('entity too large') ||
|
|
lowerMessage.includes('request entity too large') ||
|
|
lowerMessage.includes('body_not_allowed') ||
|
|
lowerMessage.includes('request body larger than')
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Handles body size limit errors by logging and throwing a user-friendly error
|
|
* @param error - The original error
|
|
* @param requestId - Request ID for logging
|
|
* @param context - Context string for logging (e.g., toolId)
|
|
* @throws Error with user-friendly message if it's a size limit error
|
|
* @returns false if not a size limit error (caller should continue handling)
|
|
*/
|
|
function handleBodySizeLimitError(error: unknown, requestId: string, context: string): boolean {
|
|
const errorMessage = toError(error).message
|
|
|
|
if (isBodySizeLimitError(errorMessage)) {
|
|
logger.error(`[${requestId}] Request body size limit exceeded for ${context}:`, {
|
|
originalError: errorMessage,
|
|
})
|
|
throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
function handleResponseSizeLimitError(error: unknown, requestId: string, context: string): boolean {
|
|
if (!isPayloadSizeLimitError(error)) return false
|
|
|
|
logger.error(`[${requestId}] Response body size limit exceeded for ${context}:`, {
|
|
label: error.label,
|
|
maxBytes: error.maxBytes,
|
|
observedBytes: error.observedBytes,
|
|
})
|
|
throw new Error(RESPONSE_SIZE_LIMIT_ERROR_MESSAGE)
|
|
}
|
|
|
|
function cloneResponseHeaders(headers: Headers | HeadersInit | undefined): Headers {
|
|
const clonedHeaders = new Headers()
|
|
if (!headers) return clonedHeaders
|
|
|
|
if (typeof (headers as Headers).forEach === 'function') {
|
|
;(headers as Headers).forEach((value, key) => {
|
|
clonedHeaders.set(key, value)
|
|
})
|
|
return clonedHeaders
|
|
}
|
|
|
|
return new Headers(headers)
|
|
}
|
|
|
|
async function readToolResponseBody(
|
|
response: {
|
|
ok?: boolean
|
|
headers?: { get(name: string): string | null }
|
|
body?: ReadableStream<Uint8Array> | null
|
|
arrayBuffer?: () => Promise<ArrayBuffer>
|
|
text?: () => Promise<string>
|
|
},
|
|
options: {
|
|
requestId: string
|
|
toolId: string
|
|
signal?: AbortSignal
|
|
}
|
|
): Promise<Buffer> {
|
|
try {
|
|
return await readResponseToBufferWithLimit(response, {
|
|
maxBytes: MAX_TOOL_RESPONSE_BODY_BYTES,
|
|
label: `${options.toolId} response body`,
|
|
signal: options.signal,
|
|
allowNoBodyFallback: true,
|
|
})
|
|
} catch (error) {
|
|
if (isPayloadSizeLimitError(error) || response.ok !== false) {
|
|
throw error
|
|
}
|
|
|
|
logger.warn(
|
|
`[${options.requestId}] Failed to read non-OK response body for ${options.toolId}`,
|
|
{
|
|
error: toError(error).message,
|
|
}
|
|
)
|
|
return Buffer.alloc(0)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* System parameters that should be filtered out when extracting tool arguments
|
|
* These are internal parameters used by the execution framework, not tool inputs
|
|
*/
|
|
const MCP_SYSTEM_PARAMETERS = new Set([
|
|
'serverId',
|
|
'serverUrl',
|
|
'toolName',
|
|
'serverName',
|
|
'_context',
|
|
'envVars',
|
|
'workflowVariables',
|
|
'blockData',
|
|
'blockNameMapping',
|
|
'_toolSchema',
|
|
])
|
|
|
|
/**
|
|
* Create an Error instance from errorInfo and attach useful context
|
|
* Uses the error extractor registry to find the best error message
|
|
*/
|
|
function createTransformedErrorFromErrorInfo(errorInfo?: ErrorInfo, extractorId?: string): Error {
|
|
const message = extractErrorMessage(errorInfo, extractorId)
|
|
const transformed = new Error(message)
|
|
Object.assign(transformed, {
|
|
status: errorInfo?.status,
|
|
statusText: errorInfo?.statusText,
|
|
data: errorInfo?.data,
|
|
})
|
|
return transformed
|
|
}
|
|
|
|
/**
|
|
* Process file outputs for a tool result if execution context is available
|
|
* Uses dynamic imports to avoid client-side bundling issues
|
|
*/
|
|
async function processFileOutputs(
|
|
result: ToolResponse,
|
|
tool: ToolConfig,
|
|
executionContext?: ExecutionContext
|
|
): Promise<ToolResponse> {
|
|
// Skip file processing if no execution context or not successful
|
|
if (!executionContext || !result.success) {
|
|
return result
|
|
}
|
|
|
|
// Skip file processing on client-side (no Node.js modules available)
|
|
if (typeof window !== 'undefined') {
|
|
return result
|
|
}
|
|
|
|
try {
|
|
// Dynamic import to avoid client-side bundling issues
|
|
const { FileToolProcessor } = await import('@/executor/utils/file-tool-processor')
|
|
|
|
// Check if tool has file outputs
|
|
if (!FileToolProcessor.hasFileOutputs(tool)) {
|
|
return result
|
|
}
|
|
|
|
const processedOutput = await FileToolProcessor.processToolOutputs(
|
|
result.output,
|
|
tool,
|
|
executionContext
|
|
)
|
|
|
|
return {
|
|
...result,
|
|
output: processedOutput,
|
|
}
|
|
} catch (error) {
|
|
logger.error(`Error processing file outputs for tool ${tool.id}:`, error)
|
|
// Return original result if file processing fails
|
|
return result
|
|
}
|
|
}
|
|
|
|
export interface ExecuteToolOptions {
|
|
skipPostProcess?: boolean
|
|
executionContext?: ExecutionContext
|
|
signal?: AbortSignal
|
|
}
|
|
|
|
/**
|
|
* Execute a tool by making the appropriate HTTP request
|
|
* All requests go directly - internal routes use regular fetch, external use SSRF-protected fetch
|
|
*/
|
|
export async function executeTool(
|
|
toolId: string,
|
|
params: Record<string, any>,
|
|
options: ExecuteToolOptions = {}
|
|
): Promise<ToolResponse> {
|
|
const { skipPostProcess = false, executionContext, signal } = options
|
|
// Fall back to the workflow execution's abort signal so plan-based execution timeouts
|
|
// and cancellation propagate to tool fetches when the caller passes no explicit signal.
|
|
const effectiveSignal = signal ?? executionContext?.abortSignal
|
|
// Capture start time for precise timing
|
|
const startTime = new Date()
|
|
const startTimeISO = startTime.toISOString()
|
|
const requestId = generateRequestId()
|
|
|
|
// Hoisted so the outer catch can attribute a thrown failure to the chosen key.
|
|
let hostedKeyForMetrics: { provider: string; tool: string; key: string } | undefined
|
|
|
|
try {
|
|
let tool: ToolConfig | undefined
|
|
|
|
// Normalize tool ID to strip resource suffixes (e.g., workflow_executor_<uuid> -> workflow_executor)
|
|
const normalizedToolId = normalizeToolId(toolId)
|
|
|
|
const scope = resolveToolScope(params, executionContext)
|
|
|
|
const toolKind: 'skill' | 'custom' | 'mcp' | undefined =
|
|
normalizedToolId === 'load_skill' || normalizedToolId === 'load_user_skill'
|
|
? 'skill'
|
|
: isCustomTool(normalizedToolId)
|
|
? 'custom'
|
|
: isMcpTool(normalizedToolId)
|
|
? 'mcp'
|
|
: undefined
|
|
|
|
// Runs for ALL tools (not just kinded ones) so the per-tool `deniedTools`
|
|
// denylist is enforced alongside the existing mcp/custom/skill gates.
|
|
if (scope.userId && scope.workspaceId) {
|
|
await assertPermissionsAllowed({
|
|
userId: scope.userId,
|
|
workspaceId: scope.workspaceId,
|
|
toolId: normalizedToolId,
|
|
toolKind,
|
|
ctx: executionContext,
|
|
})
|
|
}
|
|
|
|
if (normalizedToolId === 'load_skill' || normalizedToolId === 'load_user_skill') {
|
|
const skillName = params.skill_name
|
|
if (!skillName || !scope.workspaceId) {
|
|
return {
|
|
success: false,
|
|
output: { error: 'Missing skill_name or workspace context' },
|
|
error: 'Missing skill_name or workspace context',
|
|
}
|
|
}
|
|
const content = await resolveSkillContent(skillName, scope.workspaceId)
|
|
if (!content) {
|
|
return {
|
|
success: false,
|
|
output: { error: `Skill "${skillName}" not found` },
|
|
error: `Skill "${skillName}" not found`,
|
|
}
|
|
}
|
|
return {
|
|
success: true,
|
|
output: { content },
|
|
}
|
|
}
|
|
|
|
if (isCustomTool(normalizedToolId)) {
|
|
tool = await toolsUtilsServer.getToolAsync(normalizedToolId, {
|
|
workflowId: scope.workflowId,
|
|
userId: scope.userId,
|
|
workspaceId: scope.workspaceId,
|
|
})
|
|
if (!tool) {
|
|
logger.error(`[${requestId}] Custom tool not found: ${normalizedToolId}`)
|
|
}
|
|
} else if (isMcpTool(normalizedToolId)) {
|
|
return await executeMcpTool(
|
|
normalizedToolId,
|
|
params,
|
|
executionContext,
|
|
requestId,
|
|
startTimeISO,
|
|
effectiveSignal
|
|
)
|
|
} else {
|
|
// For built-in tools, use the synchronous version
|
|
tool = getTool(normalizedToolId)
|
|
if (!tool) {
|
|
logger.error(`[${requestId}] Built-in tool not found: ${normalizedToolId}`)
|
|
}
|
|
}
|
|
|
|
// Ensure context is preserved if it exists
|
|
const contextParams = { ...params }
|
|
|
|
// Validate the tool and its parameters
|
|
validateRequiredParametersAfterMerge(toolId, tool, contextParams)
|
|
|
|
// After validation, we know tool exists
|
|
if (!tool) {
|
|
throw new Error(`Tool not found: ${toolId}`)
|
|
}
|
|
|
|
await normalizeCopilotFileParams(tool, contextParams, scope)
|
|
normalizeCopilotCredentialParams(contextParams)
|
|
enforceCopilotCredentialSelection(toolId, tool, contextParams, scope)
|
|
|
|
// Inject hosted API key if tool supports it and user didn't provide one
|
|
const hostedKeyInfo = await injectHostedKeyIfNeeded(
|
|
tool,
|
|
contextParams,
|
|
executionContext,
|
|
requestId
|
|
)
|
|
|
|
if (hostedKeyInfo.isUsingHostedKey) {
|
|
hostedKeyForMetrics = {
|
|
provider: tool.hosting?.byokProviderId || tool.id,
|
|
tool: tool.id,
|
|
key: hostedKeyInfo.envVarName ?? 'unknown',
|
|
}
|
|
}
|
|
|
|
// If we have a credential parameter, fetch the access token
|
|
if (contextParams.oauthCredential) {
|
|
contextParams.credential = contextParams.oauthCredential
|
|
}
|
|
if (contextParams.credential) {
|
|
logger.info(
|
|
`[${requestId}] Tool ${toolId} needs access token for credential: ${contextParams.credential}`
|
|
)
|
|
try {
|
|
const baseUrl = getInternalApiBaseUrl()
|
|
|
|
const workflowId = contextParams._context?.workflowId
|
|
const userId = contextParams._context?.userId
|
|
|
|
const tokenPayload: OAuthTokenPayload = {
|
|
credentialId: contextParams.credential as string,
|
|
}
|
|
if (workflowId) {
|
|
tokenPayload.workflowId = workflowId
|
|
}
|
|
if (contextParams.impersonateUserEmail) {
|
|
tokenPayload.impersonateEmail = contextParams.impersonateUserEmail as string
|
|
}
|
|
if (tool?.oauth?.provider) {
|
|
const { getCanonicalScopesForProvider } = await import('@/lib/oauth/utils')
|
|
const providerScopes = getCanonicalScopesForProvider(tool.oauth.provider)
|
|
if (providerScopes.length > 0) {
|
|
tokenPayload.scopes = providerScopes
|
|
}
|
|
}
|
|
|
|
logger.info(`[${requestId}] Fetching access token from ${baseUrl}/api/auth/oauth/token`)
|
|
|
|
const tokenUrlObj = new URL('/api/auth/oauth/token', baseUrl)
|
|
if (workflowId) {
|
|
tokenUrlObj.searchParams.set('workflowId', workflowId)
|
|
}
|
|
if (userId && contextParams._context?.enforceCredentialAccess) {
|
|
tokenUrlObj.searchParams.set('userId', userId)
|
|
}
|
|
|
|
// Always send Content-Type; add internal auth on server-side runs
|
|
const tokenHeaders: Record<string, string> = { 'Content-Type': 'application/json' }
|
|
if (typeof window === 'undefined') {
|
|
try {
|
|
const internalToken = await generateInternalToken(userId)
|
|
tokenHeaders.Authorization = `Bearer ${internalToken}`
|
|
} catch (_e) {
|
|
// Swallow token generation errors; the request will fail and be reported upstream
|
|
}
|
|
}
|
|
|
|
const response = await fetch(tokenUrlObj.toString(), {
|
|
method: 'POST',
|
|
headers: tokenHeaders,
|
|
body: JSON.stringify(tokenPayload),
|
|
})
|
|
|
|
if (!response.ok) {
|
|
const errorText = await response.text()
|
|
logger.error(`[${requestId}] Token fetch failed for ${toolId}:`, {
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
let parsedError = errorText
|
|
try {
|
|
const parsed = JSON.parse(errorText)
|
|
if (parsed.error) parsedError = parsed.error
|
|
} catch {
|
|
// Use raw text
|
|
}
|
|
const toolLabel = tool?.name || toolId
|
|
throw new Error(`Failed to obtain credential for ${toolLabel}: ${parsedError}`)
|
|
}
|
|
|
|
const data = await response.json()
|
|
contextParams.accessToken = data.accessToken
|
|
if (data.idToken) {
|
|
contextParams.idToken = data.idToken
|
|
}
|
|
if (data.instanceUrl) {
|
|
contextParams.instanceUrl = data.instanceUrl
|
|
}
|
|
if (data.cloudId && !contextParams.cloudId) {
|
|
contextParams.cloudId = data.cloudId
|
|
}
|
|
if (data.domain && !contextParams.domain) {
|
|
contextParams.domain = data.domain
|
|
}
|
|
|
|
logger.info(`[${requestId}] Successfully got access token for ${toolId}`)
|
|
|
|
// Preserve credential for downstream transforms while removing it from request payload
|
|
// so we don't leak it to external services.
|
|
if (contextParams.credential) {
|
|
;(contextParams as any)._credentialId = contextParams.credential
|
|
}
|
|
if (workflowId) {
|
|
;(contextParams as any)._workflowId = workflowId
|
|
}
|
|
// Clean up params we don't need to pass to the actual tool
|
|
contextParams.credential = undefined
|
|
contextParams.impersonateUserEmail = undefined
|
|
if (contextParams.workflowId) contextParams.workflowId = undefined
|
|
} catch (error: any) {
|
|
logger.error(`[${requestId}] Error fetching access token for ${toolId}:`, {
|
|
error: toError(error).message,
|
|
})
|
|
throw error
|
|
}
|
|
}
|
|
|
|
// Check for direct execution (no HTTP request needed)
|
|
if (tool.directExecution) {
|
|
logger.info(`[${requestId}] Using directExecution for ${toolId}`)
|
|
const result = await tool.directExecution(contextParams)
|
|
|
|
// Apply post-processing if available and not skipped
|
|
let finalResult = result
|
|
if (tool.postProcess && result.success && !skipPostProcess) {
|
|
try {
|
|
finalResult = await tool.postProcess(result, contextParams, executeTool)
|
|
} catch (error) {
|
|
logger.error(`[${requestId}] Post-processing error for ${toolId}:`, {
|
|
error: toError(error).message,
|
|
})
|
|
finalResult = result
|
|
}
|
|
}
|
|
|
|
// Process file outputs if execution context is available
|
|
finalResult = await processFileOutputs(finalResult, tool, executionContext)
|
|
|
|
// Add timing data to the result
|
|
const endTime = new Date()
|
|
const endTimeISO = endTime.toISOString()
|
|
const duration = endTime.getTime() - startTime.getTime()
|
|
|
|
if (hostedKeyInfo.isUsingHostedKey && finalResult.success) {
|
|
await applyHostedKeyCostToResult(
|
|
finalResult,
|
|
tool,
|
|
contextParams,
|
|
executionContext,
|
|
requestId,
|
|
hostedKeyInfo.envVarName
|
|
)
|
|
} else if (hostedKeyForMetrics) {
|
|
hostedKeyMetrics.recordFailed({ ...hostedKeyForMetrics, reason: 'other' })
|
|
}
|
|
|
|
const strippedOutput = postProcessToolOutput(normalizedToolId, finalResult.output ?? {})
|
|
|
|
return {
|
|
...finalResult,
|
|
output: strippedOutput,
|
|
timing: {
|
|
startTime: startTimeISO,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
|
|
// Execute the tool request directly (internal routes use regular fetch, external use SSRF-protected fetch)
|
|
// Wrap with retry logic for hosted keys to handle rate limiting due to higher usage
|
|
const result = hostedKeyInfo.isUsingHostedKey
|
|
? await executeWithRetry(
|
|
() => executeToolRequest(toolId, tool, contextParams, effectiveSignal),
|
|
{
|
|
requestId,
|
|
toolId,
|
|
provider: tool.hosting?.byokProviderId || tool.id,
|
|
envVarName: hostedKeyInfo.envVarName!,
|
|
executionContext,
|
|
reacquireAfterRetriesExhausted: async () => {
|
|
const reacquiredEnvVar = await reacquireHostedKey(
|
|
tool,
|
|
contextParams,
|
|
executionContext,
|
|
requestId
|
|
)
|
|
if (!reacquiredEnvVar) return null
|
|
// Re-point metric labels at the freshly acquired key.
|
|
hostedKeyInfo.envVarName = reacquiredEnvVar
|
|
if (hostedKeyForMetrics) hostedKeyForMetrics.key = reacquiredEnvVar
|
|
return () => executeToolRequest(toolId, tool, contextParams, effectiveSignal)
|
|
},
|
|
}
|
|
)
|
|
: await executeToolRequest(toolId, tool, contextParams, effectiveSignal)
|
|
|
|
// Apply post-processing if available and not skipped
|
|
let finalResult = result
|
|
if (tool.postProcess && result.success && !skipPostProcess) {
|
|
try {
|
|
finalResult = await tool.postProcess(result, contextParams, executeTool)
|
|
} catch (error) {
|
|
logger.error(`[${requestId}] Post-processing error for ${toolId}:`, {
|
|
error: toError(error).message,
|
|
})
|
|
finalResult = result
|
|
}
|
|
}
|
|
|
|
// Process file outputs if execution context is available
|
|
finalResult = await processFileOutputs(finalResult, tool, executionContext)
|
|
|
|
// Add timing data to the result
|
|
const endTime = new Date()
|
|
const endTimeISO = endTime.toISOString()
|
|
const duration = endTime.getTime() - startTime.getTime()
|
|
|
|
if (hostedKeyInfo.isUsingHostedKey && finalResult.success) {
|
|
await applyHostedKeyCostToResult(
|
|
finalResult,
|
|
tool,
|
|
contextParams,
|
|
executionContext,
|
|
requestId,
|
|
hostedKeyInfo.envVarName
|
|
)
|
|
} else if (hostedKeyForMetrics) {
|
|
hostedKeyMetrics.recordFailed({ ...hostedKeyForMetrics, reason: 'other' })
|
|
}
|
|
|
|
const strippedOutput = postProcessToolOutput(normalizedToolId, finalResult.output ?? {})
|
|
|
|
return {
|
|
...finalResult,
|
|
output: strippedOutput,
|
|
timing: {
|
|
startTime: startTimeISO,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
} catch (error: any) {
|
|
logger.error(`[${requestId}] Error executing tool ${toolId}:`, {
|
|
error: toError(error).message,
|
|
stack: error instanceof Error ? error.stack : undefined,
|
|
})
|
|
|
|
if (hostedKeyForMetrics) {
|
|
hostedKeyMetrics.recordFailed({
|
|
...hostedKeyForMetrics,
|
|
reason: classifyHostedKeyFailure(error),
|
|
})
|
|
}
|
|
|
|
// Default error handling
|
|
let errorMessage = 'Unknown error occurred'
|
|
let errorDetails = {}
|
|
|
|
if (error instanceof Error) {
|
|
errorMessage = error.message || `Error executing tool ${toolId}`
|
|
// HTTP errors are thrown as Error instances carrying `status`/`statusText`/
|
|
// `data` (see createTransformedErrorFromErrorInfo). Surface them on the
|
|
// output so callers can branch on the status (e.g. treat 404 as a clean
|
|
// no-match) — the object branch below only ran for non-Error throws.
|
|
const httpStatus = (error as { status?: unknown }).status
|
|
if (typeof httpStatus === 'number') {
|
|
errorDetails = {
|
|
status: httpStatus,
|
|
statusText: (error as { statusText?: string }).statusText,
|
|
data: (error as { data?: unknown }).data,
|
|
}
|
|
}
|
|
} else if (typeof error === 'string') {
|
|
errorMessage = error
|
|
} else if (error && typeof error === 'object') {
|
|
// Handle HTTP response errors
|
|
if (error.status) {
|
|
errorMessage = `HTTP ${error.status}: ${error.statusText || 'Request failed'}`
|
|
|
|
if (error.data) {
|
|
if (typeof error.data === 'string') {
|
|
errorMessage = `${errorMessage} - ${error.data}`
|
|
} else if (error.data.message) {
|
|
errorMessage = `${errorMessage} - ${error.data.message}`
|
|
} else if (error.data.error) {
|
|
errorMessage = `${errorMessage} - ${
|
|
typeof error.data.error === 'string'
|
|
? error.data.error
|
|
: JSON.stringify(error.data.error)
|
|
}`
|
|
}
|
|
}
|
|
|
|
errorDetails = {
|
|
status: error.status,
|
|
statusText: error.statusText,
|
|
data: error.data,
|
|
}
|
|
}
|
|
// Handle other errors with messages
|
|
else if (error.message) {
|
|
// Don't pass along "undefined (undefined)" messages
|
|
if (error.message === 'undefined (undefined)') {
|
|
errorMessage = `Error executing tool ${toolId}`
|
|
// Add status if available
|
|
if (error.status) {
|
|
errorMessage += ` (Status: ${error.status})`
|
|
}
|
|
} else {
|
|
errorMessage = error.message
|
|
}
|
|
|
|
if ((error as any).cause) {
|
|
errorMessage = `${errorMessage} (${(error as any).cause})`
|
|
}
|
|
}
|
|
}
|
|
|
|
// Add timing data even for errors
|
|
const endTime = new Date()
|
|
const endTimeISO = endTime.toISOString()
|
|
const duration = endTime.getTime() - startTime.getTime()
|
|
return {
|
|
success: false,
|
|
output: errorDetails,
|
|
error: errorMessage,
|
|
timing: {
|
|
startTime: startTimeISO,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Determines if a response or result represents an error condition
|
|
*/
|
|
function isErrorResponse(
|
|
response: Response | any,
|
|
data?: any
|
|
): { isError: boolean; errorInfo?: { status?: number; statusText?: string; data?: any } } {
|
|
// HTTP Response object
|
|
if (response && typeof response === 'object' && 'ok' in response) {
|
|
if (!response.ok) {
|
|
return {
|
|
isError: true,
|
|
errorInfo: {
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
data: data,
|
|
},
|
|
}
|
|
}
|
|
return { isError: false }
|
|
}
|
|
|
|
// ToolResponse object
|
|
if (response && typeof response === 'object' && 'success' in response) {
|
|
return {
|
|
isError: !response.success,
|
|
errorInfo: response.success ? undefined : { data: response },
|
|
}
|
|
}
|
|
|
|
// Check for error indicators in data
|
|
if (data && typeof data === 'object') {
|
|
if (data.error || data.success === false) {
|
|
return {
|
|
isError: true,
|
|
errorInfo: { data: data },
|
|
}
|
|
}
|
|
}
|
|
|
|
return { isError: false }
|
|
}
|
|
|
|
/**
|
|
* Checks whether a fully resolved URL points back to this Sim instance.
|
|
* Used to propagate cycle-detection headers on API blocks that target
|
|
* the platform's own workflow execution endpoints via absolute URL.
|
|
*/
|
|
function isSelfOriginUrl(url: string): boolean {
|
|
return isSameOrigin(url, getBaseUrl()) || isSameOrigin(url, getInternalApiBaseUrl())
|
|
}
|
|
|
|
/**
|
|
* Add internal authentication token to headers if running on server
|
|
* @param headers - Headers object to modify
|
|
* @param isInternalRoute - Whether the target URL is an internal route
|
|
* @param requestId - Request ID for logging
|
|
* @param context - Context string for logging (e.g., toolId or 'proxy')
|
|
*/
|
|
async function addInternalAuthIfNeeded(
|
|
headers: Headers | Record<string, string>,
|
|
isInternalRoute: boolean,
|
|
requestId: string,
|
|
context: string,
|
|
userId?: string
|
|
): Promise<void> {
|
|
if (typeof window === 'undefined') {
|
|
if (isInternalRoute) {
|
|
try {
|
|
const internalToken = await generateInternalToken(userId)
|
|
if (headers instanceof Headers) {
|
|
headers.set('Authorization', `Bearer ${internalToken}`)
|
|
} else {
|
|
headers.Authorization = `Bearer ${internalToken}`
|
|
}
|
|
logger.info(`[${requestId}] Added internal auth token for ${context}`)
|
|
} catch (error) {
|
|
logger.error(`[${requestId}] Failed to generate internal token for ${context}:`, error)
|
|
}
|
|
} else {
|
|
logger.info(`[${requestId}] Skipping internal auth token for external URL: ${context}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
interface ResolvedRetryConfig {
|
|
maxRetries: number
|
|
initialDelayMs: number
|
|
maxDelayMs: number
|
|
}
|
|
|
|
function getRetryConfig(
|
|
retry: ToolRetryConfig | undefined,
|
|
params: Record<string, any>,
|
|
method: string
|
|
): ResolvedRetryConfig | null {
|
|
if (!retry?.enabled) return null
|
|
|
|
const isIdempotent = ['GET', 'HEAD', 'PUT', 'DELETE'].includes(method.toUpperCase())
|
|
if (retry.retryIdempotentOnly && !isIdempotent && !params.retryNonIdempotent) {
|
|
return null
|
|
}
|
|
|
|
const maxRetries = Math.min(10, Math.max(0, Number(params.retries) || retry.maxRetries || 0))
|
|
if (maxRetries === 0) return null
|
|
|
|
return {
|
|
maxRetries,
|
|
initialDelayMs: Number(params.retryDelayMs) || retry.initialDelayMs || 500,
|
|
maxDelayMs: Number(params.retryMaxDelayMs) || retry.maxDelayMs || 30000,
|
|
}
|
|
}
|
|
|
|
function isRetryableFailure(error: unknown, status?: number): boolean {
|
|
if (status === 429 || (status && status >= 500 && status <= 599)) return true
|
|
if (error instanceof Error) {
|
|
const code = (error as NodeJS.ErrnoException).code
|
|
if (code === 'ETIMEDOUT' || code === 'ECONNRESET' || code === 'ECONNABORTED') {
|
|
return true
|
|
}
|
|
const msg = error.message.toLowerCase()
|
|
if (isBodySizeLimitError(msg)) return false
|
|
return msg.includes('timeout') || msg.includes('timed out')
|
|
}
|
|
return false
|
|
}
|
|
|
|
function shouldRetryWithoutReadingBody(
|
|
status: number,
|
|
headers: { get(name: string): string | null },
|
|
retryConfig: ResolvedRetryConfig | null | undefined,
|
|
isLastAttempt: boolean
|
|
): boolean {
|
|
if (!retryConfig || isLastAttempt || !isRetryableFailure(null, status)) {
|
|
return false
|
|
}
|
|
return (
|
|
(parseRetryAfter(headers.get('retry-after'), Number.POSITIVE_INFINITY) ?? 0) <=
|
|
retryConfig.maxDelayMs
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Execute a tool request directly
|
|
* Internal routes (/api/...) use regular fetch
|
|
* External URLs use SSRF-protected fetch with DNS validation and IP pinning
|
|
*/
|
|
async function executeToolRequest(
|
|
toolId: string,
|
|
tool: ToolConfig,
|
|
params: Record<string, any>,
|
|
signal?: AbortSignal
|
|
): Promise<ToolResponse> {
|
|
const requestId = generateRequestId()
|
|
|
|
const requestParams = formatRequestParams(tool, params)
|
|
|
|
try {
|
|
const endpointUrl =
|
|
typeof tool.request.url === 'function' ? tool.request.url(params) : tool.request.url
|
|
const isInternalRoute = endpointUrl.startsWith('/api/')
|
|
const baseUrl = isInternalRoute ? getInternalApiBaseUrl() : getBaseUrl()
|
|
|
|
const fullUrlObj = new URL(endpointUrl, baseUrl)
|
|
|
|
if (isInternalRoute) {
|
|
const workflowId = params._context?.workflowId
|
|
if (workflowId) {
|
|
fullUrlObj.searchParams.set('workflowId', workflowId)
|
|
}
|
|
const userId = params._context?.userId
|
|
if (userId) {
|
|
fullUrlObj.searchParams.set('userId', userId)
|
|
}
|
|
}
|
|
|
|
const fullUrl = fullUrlObj.toString()
|
|
|
|
if (isCustomTool(toolId) && tool.request.body) {
|
|
const requestBody = tool.request.body(params)
|
|
if (
|
|
typeof requestBody === 'object' &&
|
|
requestBody !== null &&
|
|
'schema' in requestBody &&
|
|
'params' in requestBody
|
|
) {
|
|
try {
|
|
validateClientSideParams(
|
|
requestBody.params as Record<string, any>,
|
|
requestBody.schema as {
|
|
type: string
|
|
properties: Record<string, any>
|
|
required?: string[]
|
|
}
|
|
)
|
|
} catch (validationError) {
|
|
logger.error(`[${requestId}] Custom tool validation failed for ${toolId}:`, {
|
|
error: toError(validationError).message,
|
|
})
|
|
throw validationError
|
|
}
|
|
}
|
|
}
|
|
|
|
const headers = new Headers(requestParams.headers)
|
|
if (!headers.has('User-Agent')) {
|
|
headers.set('User-Agent', 'Sim')
|
|
}
|
|
await addInternalAuthIfNeeded(
|
|
headers,
|
|
isInternalRoute,
|
|
requestId,
|
|
toolId,
|
|
params._context?.userId
|
|
)
|
|
|
|
const shouldPropagateCallChain = isInternalRoute || isSelfOriginUrl(fullUrl)
|
|
if (shouldPropagateCallChain) {
|
|
const callChain = params._context?.callChain as string[] | undefined
|
|
if (callChain && callChain.length > 0) {
|
|
headers.set(SIM_VIA_HEADER, serializeCallChain(callChain))
|
|
}
|
|
}
|
|
|
|
// Check request body size before sending to detect potential size limit issues
|
|
validateRequestBodySize(requestParams.body, requestId, toolId)
|
|
|
|
// Convert Headers to plain object for secureFetchWithPinnedIP
|
|
const headersRecord: Record<string, string> = {}
|
|
headers.forEach((value, key) => {
|
|
headersRecord[key] = value
|
|
})
|
|
|
|
const retryConfig = getRetryConfig(tool.request.retry, params, requestParams.method)
|
|
const maxAttempts = retryConfig ? 1 + retryConfig.maxRetries : 1
|
|
|
|
let response: Response | undefined
|
|
let lastError: unknown
|
|
const nullBodyStatuses = new Set([101, 204, 205, 304])
|
|
|
|
for (let attempt = 0; attempt < maxAttempts; attempt++) {
|
|
const isLastAttempt = attempt === maxAttempts - 1
|
|
|
|
try {
|
|
if (isInternalRoute) {
|
|
const controller = new AbortController()
|
|
// With a caller/execution abort signal present, the plan-based timeout bounds the call and
|
|
// this only acts as a ceiling; without one, keep the tighter default as the hang safety net.
|
|
const timeout =
|
|
requestParams.timeout ||
|
|
(signal ? getMaxExecutionTimeout() : DEFAULT_EXECUTION_TIMEOUT_MS)
|
|
const timeoutId = setTimeout(
|
|
() => controller.abort(`timeout:internal_tool_fetch:${timeout}ms`),
|
|
timeout
|
|
)
|
|
|
|
let abortListener: (() => void) | null = null
|
|
if (signal) {
|
|
if (signal.aborted) {
|
|
controller.abort('caller_aborted')
|
|
} else {
|
|
abortListener = () => controller.abort('caller_aborted')
|
|
signal.addEventListener('abort', abortListener, { once: true })
|
|
}
|
|
}
|
|
|
|
try {
|
|
const internalResponse = await fetch(fullUrl, {
|
|
method: requestParams.method,
|
|
headers: headers,
|
|
body: requestParams.body,
|
|
signal: controller.signal,
|
|
})
|
|
if (
|
|
nullBodyStatuses.has(internalResponse.status) ||
|
|
shouldRetryWithoutReadingBody(
|
|
internalResponse.status,
|
|
internalResponse.headers,
|
|
retryConfig,
|
|
isLastAttempt
|
|
)
|
|
) {
|
|
internalResponse.body?.cancel().catch(() => {})
|
|
response = new Response(null, {
|
|
status: internalResponse.status,
|
|
statusText: internalResponse.statusText,
|
|
headers: cloneResponseHeaders(internalResponse.headers),
|
|
})
|
|
} else {
|
|
const bodyBuffer = await readToolResponseBody(internalResponse, {
|
|
requestId,
|
|
toolId,
|
|
signal: controller.signal,
|
|
})
|
|
response = new Response(new Uint8Array(bodyBuffer), {
|
|
status: internalResponse.status,
|
|
statusText: internalResponse.statusText,
|
|
headers: cloneResponseHeaders(internalResponse.headers),
|
|
})
|
|
}
|
|
} catch (error) {
|
|
if (error instanceof Error && error.name === 'AbortError') {
|
|
// Distinguish caller cancellation from local timeout: rethrow the AbortError
|
|
// when the caller's signal triggered the abort so cancellation propagates as-is.
|
|
if (signal?.aborted) {
|
|
throw error
|
|
}
|
|
throw new Error(`Request timed out after ${timeout}ms`)
|
|
}
|
|
throw error
|
|
} finally {
|
|
clearTimeout(timeoutId)
|
|
if (abortListener) {
|
|
signal?.removeEventListener('abort', abortListener)
|
|
}
|
|
}
|
|
} else {
|
|
const urlValidation = await validateUrlWithDNS(fullUrl, 'toolUrl')
|
|
if (!urlValidation.isValid) {
|
|
throw new Error(`Invalid tool URL: ${urlValidation.error}`)
|
|
}
|
|
|
|
const secureResponse = await secureFetchWithPinnedIP(fullUrl, urlValidation.resolvedIP!, {
|
|
method: requestParams.method,
|
|
headers: headersRecord,
|
|
body: requestParams.body ?? undefined,
|
|
timeout: requestParams.timeout,
|
|
maxResponseBytes: MAX_TOOL_RESPONSE_BODY_BYTES,
|
|
signal,
|
|
})
|
|
|
|
const responseHeaders = new Headers(secureResponse.headers.toRecord())
|
|
|
|
if (
|
|
nullBodyStatuses.has(secureResponse.status) ||
|
|
shouldRetryWithoutReadingBody(
|
|
secureResponse.status,
|
|
responseHeaders,
|
|
retryConfig,
|
|
isLastAttempt
|
|
)
|
|
) {
|
|
secureResponse.body?.cancel().catch(() => {})
|
|
response = new Response(null, {
|
|
status: secureResponse.status,
|
|
statusText: secureResponse.statusText,
|
|
headers: responseHeaders,
|
|
})
|
|
} else {
|
|
const bodyBuffer = await readToolResponseBody(secureResponse, {
|
|
requestId,
|
|
toolId,
|
|
signal,
|
|
})
|
|
response = new Response(new Uint8Array(bodyBuffer), {
|
|
status: secureResponse.status,
|
|
statusText: secureResponse.statusText,
|
|
headers: responseHeaders,
|
|
})
|
|
}
|
|
}
|
|
} catch (error) {
|
|
lastError = error
|
|
if (!retryConfig || isLastAttempt || !isRetryableFailure(error)) {
|
|
throw error
|
|
}
|
|
const delayMs = backoffWithJitter(attempt + 1, null, {
|
|
baseMs: retryConfig.initialDelayMs,
|
|
maxMs: retryConfig.maxDelayMs,
|
|
})
|
|
logger.warn(
|
|
`[${requestId}] Retrying ${toolId} after error (attempt ${attempt + 1}/${maxAttempts})`,
|
|
{ delayMs }
|
|
)
|
|
await sleep(delayMs)
|
|
continue
|
|
}
|
|
|
|
if (
|
|
retryConfig &&
|
|
!isLastAttempt &&
|
|
response &&
|
|
!response.ok &&
|
|
isRetryableFailure(null, response.status)
|
|
) {
|
|
const retryAfterMs = parseRetryAfter(
|
|
response.headers.get('retry-after'),
|
|
Number.POSITIVE_INFINITY
|
|
)
|
|
if (retryAfterMs !== null && retryAfterMs > retryConfig.maxDelayMs) {
|
|
logger.warn(
|
|
`[${requestId}] Retry-After (${retryAfterMs}ms) exceeds maxDelayMs (${retryConfig.maxDelayMs}ms), skipping retry`
|
|
)
|
|
break
|
|
}
|
|
try {
|
|
await response.arrayBuffer()
|
|
} catch {
|
|
// Ignore errors when consuming body
|
|
}
|
|
const delayMs = backoffWithJitter(attempt + 1, retryAfterMs, {
|
|
baseMs: retryConfig.initialDelayMs,
|
|
maxMs: retryConfig.maxDelayMs,
|
|
})
|
|
logger.warn(
|
|
`[${requestId}] Retrying ${toolId} after HTTP ${response.status} (attempt ${attempt + 1}/${maxAttempts})`,
|
|
{ delayMs }
|
|
)
|
|
await sleep(delayMs)
|
|
continue
|
|
}
|
|
|
|
break
|
|
}
|
|
|
|
if (!response) {
|
|
throw lastError ?? new Error(`Request failed for ${toolId}`)
|
|
}
|
|
|
|
if (!response.ok) {
|
|
let errorData: any
|
|
try {
|
|
const errorText = await response.text()
|
|
try {
|
|
errorData = JSON.parse(errorText)
|
|
} catch {
|
|
errorData = errorText
|
|
}
|
|
} catch {
|
|
logger.error(`[${requestId}] Failed to read response body for ${toolId}`)
|
|
errorData = null
|
|
}
|
|
|
|
const errorInfo: ErrorInfo = {
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
data: errorData,
|
|
}
|
|
|
|
const errorToTransform = createTransformedErrorFromErrorInfo(errorInfo, tool.errorExtractor)
|
|
const hasStructuredErrorPayload =
|
|
errorData !== null &&
|
|
typeof errorData === 'object' &&
|
|
!Array.isArray(errorData) &&
|
|
('error' in errorData || 'message' in errorData)
|
|
|
|
if (response.status === 413 && !hasStructuredErrorPayload) {
|
|
logger.error(`[${requestId}] Request body too large for ${toolId} (HTTP 413):`, {
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
errorData,
|
|
})
|
|
throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
|
|
}
|
|
|
|
logger.error(`[${requestId}] Internal API error for ${toolId}:`, {
|
|
status: errorInfo.status,
|
|
errorData: errorInfo.data,
|
|
})
|
|
|
|
throw errorToTransform
|
|
}
|
|
|
|
let responseData
|
|
const status = response.status
|
|
if (status === 202 || status === 204 || status === 205) {
|
|
responseData = { status }
|
|
} else {
|
|
if (tool.transformResponse) {
|
|
responseData = null
|
|
} else {
|
|
try {
|
|
responseData = await response.json()
|
|
} catch (jsonError) {
|
|
logger.error(`[${requestId}] JSON parse error for ${toolId}:`, {
|
|
error: toError(jsonError).message,
|
|
})
|
|
throw new Error(`Failed to parse response from ${toolId}: ${jsonError}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check for error conditions
|
|
const { isError, errorInfo } = isErrorResponse(response, responseData)
|
|
|
|
if (isError) {
|
|
// Handle error case
|
|
const errorToTransform = createTransformedErrorFromErrorInfo(errorInfo, tool.errorExtractor)
|
|
|
|
logger.error(`[${requestId}] Internal API error for ${toolId}:`, {
|
|
status: errorInfo?.status,
|
|
errorData: errorInfo?.data,
|
|
})
|
|
|
|
throw errorToTransform
|
|
}
|
|
|
|
// Success case: use transformResponse if available
|
|
if (tool.transformResponse) {
|
|
try {
|
|
// Create a mock response object that provides the methods transformResponse needs
|
|
const mockResponse = {
|
|
ok: response.ok,
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
headers: response.headers,
|
|
url: fullUrl,
|
|
json: () => response.json(),
|
|
text: () => response.text(),
|
|
arrayBuffer: () => response.arrayBuffer(),
|
|
blob: () => response.blob(),
|
|
} as Response
|
|
|
|
const data = await tool.transformResponse(mockResponse, params)
|
|
return data
|
|
} catch (transformError) {
|
|
logger.error(`[${requestId}] Transform response error for ${toolId}:`, {
|
|
error: toError(transformError).message,
|
|
})
|
|
throw transformError
|
|
}
|
|
}
|
|
|
|
// Default success response handling
|
|
return {
|
|
success: true,
|
|
output: responseData.output || responseData,
|
|
error: undefined,
|
|
}
|
|
} catch (error: any) {
|
|
handleResponseSizeLimitError(error, requestId, toolId)
|
|
|
|
// Check if this is a body size limit error and throw user-friendly message
|
|
handleBodySizeLimitError(error, requestId, toolId)
|
|
|
|
logger.error(`[${requestId}] Internal request error for ${toolId}:`, {
|
|
error: toError(error).message,
|
|
})
|
|
|
|
// Let the error bubble up to be handled in the main executeTool function
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validates parameters on the client side before sending to the execute endpoint
|
|
*/
|
|
function validateClientSideParams(
|
|
params: Record<string, any>,
|
|
schema: {
|
|
type: string
|
|
properties: Record<string, any>
|
|
required?: string[]
|
|
}
|
|
) {
|
|
if (!schema || schema.type !== 'object') {
|
|
throw new Error('Invalid schema format')
|
|
}
|
|
|
|
// Internal parameters that should be excluded from validation
|
|
const internalParamSet = new Set([
|
|
'_context',
|
|
'_toolSchema',
|
|
'workflowId',
|
|
'envVars',
|
|
'workflowVariables',
|
|
'blockData',
|
|
'blockNameMapping',
|
|
])
|
|
|
|
// Check required parameters
|
|
if (schema.required) {
|
|
for (const requiredParam of schema.required) {
|
|
if (!(requiredParam in params)) {
|
|
throw new Error(`Required parameter missing: ${requiredParam}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check parameter types (basic validation)
|
|
for (const [paramName, paramValue] of Object.entries(params)) {
|
|
// Skip validation for internal parameters
|
|
if (internalParamSet.has(paramName)) {
|
|
continue
|
|
}
|
|
|
|
const paramSchema = schema.properties[paramName]
|
|
if (!paramSchema) {
|
|
throw new Error(`Unknown parameter: ${paramName}`)
|
|
}
|
|
|
|
// Basic type checking
|
|
const type = paramSchema.type
|
|
if (type === 'string' && typeof paramValue !== 'string') {
|
|
throw new Error(`Parameter ${paramName} should be a string`)
|
|
}
|
|
if (type === 'number' && typeof paramValue !== 'number') {
|
|
throw new Error(`Parameter ${paramName} should be a number`)
|
|
}
|
|
if (type === 'boolean' && typeof paramValue !== 'boolean') {
|
|
throw new Error(`Parameter ${paramName} should be a boolean`)
|
|
}
|
|
if (type === 'array' && !Array.isArray(paramValue)) {
|
|
throw new Error(`Parameter ${paramName} should be an array`)
|
|
}
|
|
if (type === 'object' && (typeof paramValue !== 'object' || paramValue === null)) {
|
|
throw new Error(`Parameter ${paramName} should be an object`)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Execute an MCP tool via the server-side MCP endpoint
|
|
*
|
|
* @param toolId - MCP tool ID in format "mcp-serverId-toolName"
|
|
* @param params - Tool parameters
|
|
* @param executionContext - Execution context
|
|
* @param requestId - Request ID for logging
|
|
* @param startTimeISO - Start time for timing
|
|
*/
|
|
async function executeMcpTool(
|
|
toolId: string,
|
|
params: Record<string, any>,
|
|
executionContext?: ExecutionContext,
|
|
requestId?: string,
|
|
startTimeISO?: string,
|
|
signal?: AbortSignal
|
|
): Promise<ToolResponse> {
|
|
const actualRequestId = requestId || generateRequestId()
|
|
const actualStartTime = startTimeISO || new Date().toISOString()
|
|
|
|
try {
|
|
logger.info(`[${actualRequestId}] Executing MCP tool: ${toolId}`)
|
|
|
|
const { serverId, toolName } = parseMcpToolId(toolId)
|
|
|
|
const baseUrl = getInternalApiBaseUrl()
|
|
|
|
const mcpScope = resolveToolScope(params, executionContext)
|
|
|
|
const headers: Record<string, string> = { 'Content-Type': 'application/json' }
|
|
|
|
if (typeof window === 'undefined') {
|
|
try {
|
|
const internalToken = await generateInternalToken(mcpScope.userId)
|
|
headers.Authorization = `Bearer ${internalToken}`
|
|
} catch (error) {
|
|
logger.error(`[${actualRequestId}] Failed to generate internal token:`, error)
|
|
}
|
|
}
|
|
|
|
// Handle two different parameter structures:
|
|
// 1. Direct MCP blocks: arguments are stored as JSON string in 'arguments' field
|
|
// 2. Agent blocks: arguments are passed directly as top-level parameters
|
|
let toolArguments = {}
|
|
|
|
// First check if we have the 'arguments' field (direct MCP block usage)
|
|
if (params.arguments) {
|
|
if (typeof params.arguments === 'string') {
|
|
try {
|
|
toolArguments = JSON.parse(params.arguments)
|
|
} catch (error) {
|
|
logger.warn(`[${actualRequestId}] Failed to parse MCP arguments JSON:`, params.arguments)
|
|
toolArguments = {}
|
|
}
|
|
} else {
|
|
toolArguments = params.arguments
|
|
}
|
|
} else {
|
|
// Agent block usage: extract MCP-specific arguments by filtering out system parameters
|
|
toolArguments = Object.fromEntries(
|
|
Object.entries(params).filter(([key]) => !MCP_SYSTEM_PARAMETERS.has(key))
|
|
)
|
|
}
|
|
|
|
if (mcpScope.callChain && mcpScope.callChain.length > 0) {
|
|
headers[SIM_VIA_HEADER] = serializeCallChain(mcpScope.callChain)
|
|
}
|
|
|
|
if (!mcpScope.workspaceId) {
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: `Missing workspaceId in execution context for MCP tool ${toolName}`,
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: new Date().toISOString(),
|
|
duration: Date.now() - new Date(actualStartTime).getTime(),
|
|
},
|
|
}
|
|
}
|
|
|
|
const requestBody: Record<string, any> = {
|
|
serverId,
|
|
toolName,
|
|
arguments: toolArguments,
|
|
workflowId: mcpScope.workflowId,
|
|
workspaceId: mcpScope.workspaceId,
|
|
}
|
|
|
|
const body = JSON.stringify(requestBody)
|
|
|
|
// Check request body size before sending
|
|
validateRequestBodySize(body, actualRequestId, `mcp:${toolId}`)
|
|
|
|
logger.info(`[${actualRequestId}] Making MCP tool request to ${toolName} on ${serverId}`, {
|
|
hasWorkspaceId: !!mcpScope.workspaceId,
|
|
hasWorkflowId: !!mcpScope.workflowId,
|
|
})
|
|
|
|
const mcpUrl = new URL('/api/mcp/tools/execute', baseUrl)
|
|
if (mcpScope.userId) {
|
|
mcpUrl.searchParams.set('userId', mcpScope.userId)
|
|
}
|
|
|
|
const response = await fetch(mcpUrl.toString(), {
|
|
method: 'POST',
|
|
headers,
|
|
body,
|
|
signal,
|
|
})
|
|
|
|
const endTime = new Date()
|
|
const endTimeISO = endTime.toISOString()
|
|
const duration = endTime.getTime() - new Date(actualStartTime).getTime()
|
|
|
|
if (!response.ok) {
|
|
// Check for 413 (Entity Too Large) - body size limit exceeded
|
|
if (response.status === 413) {
|
|
logger.error(`[${actualRequestId}] Request body too large for mcp:${toolId} (HTTP 413)`)
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: BODY_SIZE_LIMIT_ERROR_MESSAGE,
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
|
|
let errorMessage = `MCP tool execution failed: ${response.status} ${response.statusText}`
|
|
|
|
try {
|
|
const errorData = await response.json()
|
|
if (errorData.error) {
|
|
errorMessage = errorData.error
|
|
}
|
|
} catch {
|
|
// Failed to parse error response, use default message
|
|
}
|
|
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: errorMessage,
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
|
|
const result = await response.json()
|
|
|
|
if (!result.success) {
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: result.error || 'MCP tool execution failed',
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
|
|
logger.info(`[${actualRequestId}] MCP tool ${toolId} executed successfully`)
|
|
|
|
return {
|
|
success: true,
|
|
output: result.data?.output || result.output || result.data || {},
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
} catch (error) {
|
|
const endTime = new Date()
|
|
const endTimeISO = endTime.toISOString()
|
|
const duration = endTime.getTime() - new Date(actualStartTime).getTime()
|
|
|
|
// Check if this is a body size limit error
|
|
const errorMsg = toError(error).message
|
|
if (isBodySizeLimitError(errorMsg)) {
|
|
logger.error(`[${actualRequestId}] Request body size limit exceeded for mcp:${toolId}:`, {
|
|
originalError: errorMsg,
|
|
})
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: BODY_SIZE_LIMIT_ERROR_MESSAGE,
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
|
|
logger.error(`[${actualRequestId}] Error executing MCP tool ${toolId}:`, error)
|
|
|
|
const errorMessage = getErrorMessage(error, `Failed to execute MCP tool ${toolId}`)
|
|
|
|
return {
|
|
success: false,
|
|
output: {},
|
|
error: errorMessage,
|
|
timing: {
|
|
startTime: actualStartTime,
|
|
endTime: endTimeISO,
|
|
duration,
|
|
},
|
|
}
|
|
}
|
|
}
|