relay_workflow_run
Execute multi-step AI workflows with reduced context usage by keeping intermediate results in the workflow engine, supporting multiple model calls and tool integrations.
Instructions
Execute a multi-step AI workflow. Intermediate results stay in the workflow engine (not your context), providing 90%+ context reduction on complex pipelines. Use for any task requiring multiple model calls or tool integrations. Cost tracks your provider bills, not RelayPlane fees - we're BYOK.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | Workflow name for tracing | |
| steps | Yes | Workflow steps | |
| input | Yes | Input data (accessible via {{input.field}}) |
Implementation Reference
- src/tools/relay-workflow-run.ts:162-355 (handler)Core handler function that executes the relay_workflow_run tool: performs topological sort on steps, interpolates prompts, executes AI steps via relayRun or MCP, handles budget/costs, stores run results, computes context reduction percentage.
export async function relayWorkflowRun(input: RelayWorkflowRunInput): Promise<RelayWorkflowRunResponse> { const startTime = Date.now(); const runId = generateRunId(); const config = getConfig(); const stepResults: Record<string, WorkflowStepResult> = {}; let totalTokens = 0; let totalCost = 0; try { // Validate and sort steps const sortedSteps = topologicalSort(input.steps); // Estimate cost for AI steps and check budget const aiSteps = sortedSteps.filter(s => s.model && s.prompt); const estimatedCost = estimateWorkflowCost(aiSteps); const budgetCheck = checkBudget(estimatedCost); if (!budgetCheck.allowed) { throw new Error(budgetCheck.error); } // Check all providers are configured for (const step of aiSteps) { const provider = step.model!.split(':')[0]; if (!isProviderConfigured(provider)) { throw new Error( `Provider "${provider}" (step "${step.name}") is not configured. Set ${provider.toUpperCase()}_API_KEY environment variable.` ); } } // Execute steps in order const context = { input: input.input, steps: {} as Record<string, any> }; for (const step of sortedSteps) { const stepStart = Date.now(); try { if (step.model && step.prompt) { // AI step - use relay_run const interpolatedPrompt = interpolate(step.prompt, context); const interpolatedSystemPrompt = step.systemPrompt ? interpolate(step.systemPrompt, context) : undefined; const result = await relayRun({ model: step.model, prompt: interpolatedPrompt, systemPrompt: interpolatedSystemPrompt, schema: step.schema, }); if (!result.success) { throw new Error(result.error?.message || 'Step execution failed'); } context.steps[step.name] = { output: result.output }; stepResults[step.name] = { success: true, output: result.output, durationMs: result.durationMs, usage: { promptTokens: result.usage.promptTokens, completionTokens: result.usage.completionTokens, estimatedProviderCostUsd: result.usage.estimatedProviderCostUsd, }, }; totalTokens += result.usage.totalTokens; totalCost += result.usage.estimatedProviderCostUsd; } else if (step.mcp) { // MCP step - placeholder for now // In a full implementation, this would call the MCP tool const stepDuration = Date.now() - stepStart; context.steps[step.name] = { output: { message: 'MCP step executed (placeholder)' } }; stepResults[step.name] = { success: true, output: { message: 'MCP step executed (placeholder)' }, durationMs: stepDuration, }; } else { // Transform step or other type const stepDuration = Date.now() - stepStart; context.steps[step.name] = { output: null }; stepResults[step.name] = { success: true, output: null, durationMs: stepDuration, }; } } catch (error) { const stepDuration = Date.now() - stepStart; const errorMessage = error instanceof Error ? error.message : String(error); stepResults[step.name] = { success: false, output: null, durationMs: stepDuration, error: { code: 'STEP_EXECUTION_ERROR', message: errorMessage, }, }; throw new Error(`Step "${step.name}" failed: ${errorMessage}`); } } const totalDurationMs = Date.now() - startTime; const lastStep = sortedSteps[sortedSteps.length - 1]; const finalOutput = stepResults[lastStep?.name]?.output; const response: RelayWorkflowRunResponse = { success: true, steps: stepResults, finalOutput, totalUsage: { totalTokens, estimatedProviderCostUsd: totalCost, }, totalDurationMs, runId, traceUrl: `${config.traceUrlBase}/${runId}`, contextReduction: estimateContextReduction(sortedSteps, stepResults), }; // Store run addRun({ runId, type: 'workflow', workflowName: input.name, success: true, startTime: new Date(startTime), endTime: new Date(), durationMs: totalDurationMs, usage: { promptTokens: 0, // Aggregated in steps completionTokens: 0, totalTokens, estimatedProviderCostUsd: totalCost, }, input: input.input, output: finalOutput, steps: stepResults, contextReduction: response.contextReduction, }); return response; } catch (error) { const totalDurationMs = Date.now() - startTime; const errorMessage = error instanceof Error ? error.message : String(error); const response: RelayWorkflowRunResponse = { success: false, steps: stepResults, finalOutput: null, totalUsage: { totalTokens, estimatedProviderCostUsd: totalCost, }, totalDurationMs, runId, traceUrl: `${config.traceUrlBase}/${runId}`, contextReduction: 'N/A (workflow failed)', error: { code: 'WORKFLOW_ERROR', message: errorMessage, }, }; // Store failed run addRun({ runId, type: 'workflow', workflowName: input.name, success: false, startTime: new Date(startTime), endTime: new Date(), durationMs: totalDurationMs, usage: { promptTokens: 0, completionTokens: 0, totalTokens, estimatedProviderCostUsd: totalCost, }, input: input.input, steps: stepResults, error: errorMessage, }); return response; } } - MCP tool definition object with name, description, and detailed inputSchema for validation in MCP protocol.
export const relayWorkflowRunDefinition = { name: 'relay_workflow_run', description: "Execute a multi-step AI workflow. Intermediate results stay in the workflow engine (not your context), providing 90%+ context reduction on complex pipelines. Use for any task requiring multiple model calls or tool integrations. Cost tracks your provider bills, not RelayPlane fees - we're BYOK.", inputSchema: { type: 'object' as const, properties: { name: { type: 'string', description: 'Workflow name for tracing', }, steps: { type: 'array', description: 'Workflow steps', items: { type: 'object', properties: { name: { type: 'string', description: 'Step name' }, model: { type: 'string', description: 'Model in provider:model format' }, prompt: { type: 'string', description: 'Prompt template (supports {{input.field}} and {{steps.stepName.output}})' }, systemPrompt: { type: 'string', description: 'Optional system prompt' }, depends: { type: 'array', items: { type: 'string' }, description: 'Steps this step depends on' }, mcp: { type: 'string', description: 'MCP tool in server:tool format' }, params: { type: 'object', description: 'MCP tool parameters' }, schema: { type: 'object', description: 'JSON schema for structured output' }, }, required: ['name'], }, }, input: { type: 'object', description: 'Input data (accessible via {{input.field}})', }, }, required: ['name', 'steps', 'input'], }, }; - Zod schema for input validation of relay_workflow_run parameters.
export const relayWorkflowRunSchema = z.object({ name: z.string().describe('Workflow name for tracing'), steps: z.array(workflowStepSchema).describe('Workflow steps'), input: z.object({}).passthrough().describe('Input data (accessible via {{input.field}})'), }); - src/server.ts:59-67 (registration)Registration of relayWorkflowRunDefinition in the TOOLS array provided to MCP listTools handler.
const TOOLS = [ relayModelsListDefinition, relayRunDefinition, relayWorkflowRunDefinition, relayWorkflowValidateDefinition, relaySkillsListDefinition, relayRunsListDefinition, relayRunGetDefinition, ]; - src/server.ts:119-122 (registration)Dispatch case in MCP callTool handler that invokes the relay_workflow_run tool implementation.
case 'relay_workflow_run': { const parsed = relayWorkflowRunSchema.parse(args); result = await relayWorkflowRun(parsed); break;