Fix broken workflow engine + add 14 working pipeline presets + persistence
This commit is contained in:
drjones
@@ -27,11 +27,29 @@ export const WorkflowsTab: React.FC = () => {
|
||||
const [workflows, setWorkflows] = useState<MCPWorkflow[]>([]);
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [executingId, setExecutingId] = useState<string | null>(null);
|
||||
const [inputsJson, setInputsJson] = useState<Record<string, string>>({
|
||||
'wf-preset-summarize-slack': '{"url": "https://en.wikipedia.org/wiki/Model_Context_Protocol"}',
|
||||
});
|
||||
const [inputsJson, setInputsJson] = useState<Record<string, string>>({});
|
||||
const [executionResult, setExecutionResult] = useState<any | null>(null);
|
||||
|
||||
// Derive example input keys from a workflow's {{input.X}} references.
|
||||
const extractInputKeys = (wf: MCPWorkflow): string[] => {
|
||||
const keys = new Set<string>();
|
||||
for (const step of wf.steps) {
|
||||
const raw = JSON.stringify(step.argsTemplate);
|
||||
const re = /\{\{input\.([a-zA-Z0-9_]+)\}\}/g;
|
||||
let m;
|
||||
while ((m = re.exec(raw)) !== null) keys.add(m[1]);
|
||||
}
|
||||
return Array.from(keys);
|
||||
};
|
||||
|
||||
const defaultInputFor = (wf: MCPWorkflow): string => {
|
||||
const keys = extractInputKeys(wf);
|
||||
if (keys.length === 0) return '{}';
|
||||
const obj: Record<string, string> = {};
|
||||
for (const k of keys) obj[k] = '';
|
||||
return JSON.stringify(obj, null, 2);
|
||||
};
|
||||
|
||||
const fetchWorkflows = async () => {
|
||||
setLoading(true);
|
||||
try {
|
||||
@@ -177,7 +195,7 @@ export const WorkflowsTab: React.FC = () => {
|
||||
<label className="block text-xs font-semibold text-slate-700">Pipeline Input JSON Arguments:</label>
|
||||
<input
|
||||
type="text"
|
||||
value={inputsJson[wf.id] || '{"url": "https://en.wikipedia.org/wiki/Model_Context_Protocol"}'}
|
||||
value={inputsJson[wf.id] ?? defaultInputFor(wf)}
|
||||
onChange={(e) => setInputsJson({ ...inputsJson, [wf.id]: e.target.value })}
|
||||
className="w-full rounded-lg border border-slate-300 p-2.5 font-mono text-xs bg-white text-slate-900 focus:outline-none focus:ring-2 focus:ring-indigo-500"
|
||||
/>
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { mcpRegistry } from './mcpRegistry.js';
|
||||
import { ToolExecutionResult } from '../types.js';
|
||||
|
||||
@@ -5,7 +7,7 @@ export interface WorkflowStep {
|
||||
stepId: string;
|
||||
name: string;
|
||||
toolName: string;
|
||||
// Arguments mapping template: e.g. { "url": "{{input.url}}", "text": "{{step1.result.markdown}}" }
|
||||
// Arguments mapping template: e.g. { "url": "{{input.url}}", "text": "{{step1.result}}" }
|
||||
argsTemplate: Record<string, any>;
|
||||
}
|
||||
|
||||
@@ -15,6 +17,7 @@ export interface MCPWorkflow {
|
||||
description: string;
|
||||
enabled: boolean;
|
||||
steps: WorkflowStep[];
|
||||
isPreset?: boolean;
|
||||
createdAt: string;
|
||||
updatedAt?: string;
|
||||
lastExecution?: {
|
||||
@@ -27,38 +30,218 @@ export interface MCPWorkflow {
|
||||
|
||||
class MCPWorkflowManager {
|
||||
private workflows: Map<string, MCPWorkflow> = new Map();
|
||||
private persistenceFilePath: string;
|
||||
|
||||
constructor() {
|
||||
this.persistenceFilePath = path.join(process.cwd(), 'data', 'workflows_persistence.json');
|
||||
this.initializePresets();
|
||||
this.loadPersistedWorkflows();
|
||||
}
|
||||
|
||||
private ensureDataDir() {
|
||||
const dir = path.dirname(this.persistenceFilePath);
|
||||
if (!fs.existsSync(dir)) {
|
||||
try {
|
||||
fs.mkdirSync(dir, { recursive: true });
|
||||
} catch (err) {
|
||||
console.error('[Workflows] Failed to create data dir:', err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private loadPersistedWorkflows() {
|
||||
try {
|
||||
if (fs.existsSync(this.persistenceFilePath)) {
|
||||
const raw = fs.readFileSync(this.persistenceFilePath, 'utf-8');
|
||||
const data = JSON.parse(raw);
|
||||
if (Array.isArray(data)) {
|
||||
for (const wf of data) {
|
||||
if (wf && wf.id) {
|
||||
// Presets: keep code definition (source of truth) but restore enabled + lastExecution
|
||||
const existing = this.workflows.get(wf.id);
|
||||
if (existing && wf.isPreset) {
|
||||
existing.enabled = wf.enabled ?? existing.enabled;
|
||||
existing.lastExecution = wf.lastExecution;
|
||||
} else {
|
||||
this.workflows.set(wf.id, wf);
|
||||
}
|
||||
}
|
||||
}
|
||||
console.log(`[Workflows] Restored ${data.length} persisted workflow(s).`);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.warn('[Workflows] Could not load persisted workflows, using presets:', err);
|
||||
}
|
||||
}
|
||||
|
||||
private savePersistedWorkflows() {
|
||||
try {
|
||||
this.ensureDataDir();
|
||||
const all = Array.from(this.workflows.values());
|
||||
fs.writeFileSync(this.persistenceFilePath, JSON.stringify(all, null, 2), 'utf-8');
|
||||
} catch (err) {
|
||||
console.error('[Workflows] Error saving workflows:', err);
|
||||
}
|
||||
}
|
||||
|
||||
private initializePresets() {
|
||||
const sampleWorkflow: MCPWorkflow = {
|
||||
id: 'wf-preset-summarize-slack',
|
||||
name: 'Web Scrape & Sentiment Pipeline',
|
||||
description: 'Scrapes HTML from target URL, converts to markdown, then extracts top keywords and text sentiment.',
|
||||
const now = () => new Date().toISOString();
|
||||
const presets: Omit<MCPWorkflow, 'id' | 'createdAt'>[] = [
|
||||
{
|
||||
name: 'Web Scrape → AI Summary',
|
||||
description: 'Fetches a page, extracts clean markdown, then summarizes it with the local LLM.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{
|
||||
stepId: 'step1',
|
||||
name: 'Fetch Web Content',
|
||||
toolName: 'web_scrape_markdown',
|
||||
argsTemplate: {
|
||||
url: '{{input.url}}',
|
||||
},
|
||||
},
|
||||
{
|
||||
stepId: 'step2',
|
||||
name: 'Analyze Sentiment & Keywords',
|
||||
toolName: 'extract_text_keywords_mcp',
|
||||
argsTemplate: {
|
||||
text: '{{step1.result.markdown}}',
|
||||
},
|
||||
},
|
||||
{ stepId: 'step1', name: 'Fetch Web Content', toolName: 'web_scrape_markdown', argsTemplate: { url: '{{input.url}}' } },
|
||||
{ stepId: 'step2', name: 'Summarize', toolName: 'text_summarize_classify', argsTemplate: { text: '{{step1.result}}', task: 'summarize' } },
|
||||
],
|
||||
createdAt: new Date().toISOString(),
|
||||
};
|
||||
this.workflows.set(sampleWorkflow.id, sampleWorkflow);
|
||||
},
|
||||
{
|
||||
name: 'Web Scrape → Sentiment & Keywords',
|
||||
description: 'Fetches a page then extracts key takeaways, sentiment, and entities.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Fetch Web Content', toolName: 'web_scrape_markdown', argsTemplate: { url: '{{input.url}}' } },
|
||||
{ stepId: 'step2', name: 'Analyze', toolName: 'text_summarize_classify', argsTemplate: { text: '{{step1.result}}', task: 'all' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Search → Summarize → Remember',
|
||||
description: 'Runs a web search, summarizes the synthesis, and stores it in the memory vault.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Search', toolName: 'web_search_gemini', argsTemplate: { query: '{{input.query}}' } },
|
||||
{ stepId: 'step2', name: 'Summarize', toolName: 'text_summarize_classify', argsTemplate: { text: '{{step1.result}}', task: 'all' } },
|
||||
{ stepId: 'step3', name: 'Store Memory', toolName: 'memory_store_set', argsTemplate: { key: '{{input.memoryKey}}', value: '{{step2.result}}', tags: ['research', 'workflow'] } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'API Fetch → Convert → Archive',
|
||||
description: 'GETs a JSON API, converts it to another format, and archives it to memory.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Fetch API', toolName: 'http_client', argsTemplate: { url: '{{input.url}}', method: 'GET' } },
|
||||
{ stepId: 'step2', name: 'Convert Format', toolName: 'data_converter', argsTemplate: { data: '{{step1.result}}', fromFormat: 'json', toFormat: '{{input.toFormat}}' } },
|
||||
{ stepId: 'step3', name: 'Archive', toolName: 'memory_store_set', argsTemplate: { key: '{{input.memoryKey}}', value: '{{step2.result}}' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Encode → Hash Fingerprint',
|
||||
description: 'Base64-encodes text then produces a SHA-256 hash of the encoded value.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Base64 Encode', toolName: 'encoder_decoder', argsTemplate: { input: '{{input.text}}', action: 'base64_encode' } },
|
||||
{ stepId: 'step2', name: 'SHA-256 Hash', toolName: 'crypto_hash_generator', argsTemplate: { input: '{{step1.result}}', algorithm: 'sha256' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'JSON → YAML → Stored Config',
|
||||
description: 'Converts a JSON payload to YAML and stores it as a config in memory.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Convert to YAML', toolName: 'data_converter', argsTemplate: { data: '{{input.json}}', fromFormat: 'json', toFormat: 'yaml' } },
|
||||
{ stepId: 'step2', name: 'Store Config', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step1.result}}', tags: ['config'] } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Regex Extract → Convert → Store',
|
||||
description: 'Extracts matches with a regex, converts the result set, and saves it.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Extract Matches', toolName: 'regex_tester', argsTemplate: { pattern: '{{input.pattern}}', text: '{{input.text}}', flags: 'g' } },
|
||||
{ stepId: 'step2', name: 'Convert to JSON', toolName: 'data_converter', argsTemplate: { data: '{{step1.result}}', fromFormat: 'json', toFormat: 'json' } },
|
||||
{ stepId: 'step3', name: 'Store', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step2.result}}' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Math Evaluate → Store Result',
|
||||
description: 'Evaluates a math expression or stats set and saves the answer to memory.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Evaluate', toolName: 'math_evaluator', argsTemplate: { operation: '{{input.operation}}', expression: '{{input.expression}}' } },
|
||||
{ stepId: 'step2', name: 'Store', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step1.result}}' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'URL Parse → QR Code',
|
||||
description: 'Parses and validates a URL, then renders it as a QR code.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Parse URL', toolName: 'network_utilities', argsTemplate: { action: 'parse_url', target: '{{input.url}}' } },
|
||||
{ stepId: 'step2', name: 'Generate QR', toolName: 'qr_barcode_generator', argsTemplate: { text: '{{input.url}}', type: 'qr' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Cron → Explain → Remember',
|
||||
description: 'Validates a cron expression, explains it in plain English, and saves the schedule.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Explain Cron', toolName: 'cron_calculator', argsTemplate: { expression: '{{input.expression}}' } },
|
||||
{ stepId: 'step2', name: 'Store Schedule', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step1.result}}' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Hash → URL Encode',
|
||||
description: 'Generates a SHA-256 hash then URL-encodes it for safe transport.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Hash', toolName: 'crypto_hash_generator', argsTemplate: { input: '{{input.text}}', algorithm: 'sha256' } },
|
||||
{ stepId: 'step2', name: 'URL Encode', toolName: 'encoder_decoder', argsTemplate: { input: '{{step1.result}}', action: 'url_encode' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Search → Store Memory',
|
||||
description: 'Runs a web search and stores the raw synthesis directly into the memory vault.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Search', toolName: 'web_search_gemini', argsTemplate: { query: '{{input.query}}' } },
|
||||
{ stepId: 'step2', name: 'Store', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step1.result}}' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Convert JSON → QR Code',
|
||||
description: 'Converts a JSON payload to a query string and encodes it as a QR code.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'Convert to Query String', toolName: 'data_converter', argsTemplate: { data: '{{input.json}}', fromFormat: 'json', toFormat: 'query_string' } },
|
||||
{ stepId: 'step2', name: 'Generate QR', toolName: 'qr_barcode_generator', argsTemplate: { text: '{{step1.result}}', type: 'qr' } },
|
||||
],
|
||||
},
|
||||
{
|
||||
name: 'Encode → Hash → Remember',
|
||||
description: 'URL-encodes input, hashes it, and persists the final fingerprint to memory.',
|
||||
enabled: true,
|
||||
isPreset: true,
|
||||
steps: [
|
||||
{ stepId: 'step1', name: 'URL Encode', toolName: 'encoder_decoder', argsTemplate: { input: '{{input.text}}', action: 'url_encode' } },
|
||||
{ stepId: 'step2', name: 'Hash', toolName: 'crypto_hash_generator', argsTemplate: { input: '{{step1.result}}', algorithm: 'sha256' } },
|
||||
{ stepId: 'step3', name: 'Store', toolName: 'memory_store_set', argsTemplate: { key: '{{input.key}}', value: '{{step2.result}}' } },
|
||||
],
|
||||
},
|
||||
];
|
||||
|
||||
for (const p of presets) {
|
||||
const id = `wf-preset-${p.name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/(^-|-$)/g, '').slice(0, 48)}`;
|
||||
this.workflows.set(id, {
|
||||
...p,
|
||||
id,
|
||||
createdAt: now(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
public getWorkflows(): MCPWorkflow[] {
|
||||
@@ -78,6 +261,7 @@ class MCPWorkflowManager {
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
this.workflows.set(id, newWf);
|
||||
this.savePersistedWorkflows();
|
||||
return newWf;
|
||||
}
|
||||
|
||||
@@ -91,11 +275,14 @@ class MCPWorkflowManager {
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
this.workflows.set(id, updated);
|
||||
this.savePersistedWorkflows();
|
||||
return updated;
|
||||
}
|
||||
|
||||
public deleteWorkflow(id: string): boolean {
|
||||
return this.workflows.delete(id);
|
||||
const deleted = this.workflows.delete(id);
|
||||
if (deleted) this.savePersistedWorkflows();
|
||||
return deleted;
|
||||
}
|
||||
|
||||
public async executeWorkflow(id: string, initialInputs: Record<string, any>): Promise<{
|
||||
@@ -121,10 +308,7 @@ class MCPWorkflowManager {
|
||||
const contextData: Record<string, any> = { input: initialInputs };
|
||||
|
||||
for (const step of wf.steps) {
|
||||
// Interpolate arguments template from contextData
|
||||
const resolvedArgs = this.resolveArgsTemplate(step.argsTemplate, contextData);
|
||||
|
||||
// Execute tool
|
||||
const execRes = await mcpRegistry.executeServerTool(step.toolName, resolvedArgs);
|
||||
stepResults[step.stepId] = execRes;
|
||||
contextData[step.stepId] = execRes;
|
||||
@@ -137,6 +321,7 @@ class MCPWorkflowManager {
|
||||
success: false,
|
||||
stepResults,
|
||||
};
|
||||
this.savePersistedWorkflows();
|
||||
return {
|
||||
success: false,
|
||||
durationMs,
|
||||
@@ -157,6 +342,7 @@ class MCPWorkflowManager {
|
||||
success: true,
|
||||
stepResults,
|
||||
};
|
||||
this.savePersistedWorkflows();
|
||||
|
||||
return {
|
||||
success: true,
|
||||
@@ -183,14 +369,14 @@ class MCPWorkflowManager {
|
||||
}
|
||||
|
||||
private interpolateString(str: string, context: Record<string, any>): any {
|
||||
// Exact match e.g. "{{step1.result.markdown}}"
|
||||
// Exact match e.g. "{{step1.result}}"
|
||||
const exactMatch = str.match(/^\{\{([a-zA-Z0-9_.]+)\}\}$/);
|
||||
if (exactMatch) {
|
||||
const path = exactMatch[1];
|
||||
return this.getValueByPath(context, path);
|
||||
}
|
||||
|
||||
// Partial string match e.g. "Result is {{step1.result.title}}"
|
||||
// Partial string match e.g. "Result is {{step1.result}}"
|
||||
return str.replace(/\{\{([a-zA-Z0-9_.]+)\}\}/g, (_, path) => {
|
||||
const v = this.getValueByPath(context, path);
|
||||
if (v === undefined || v === null) return '';
|
||||
|
||||
Reference in New Issue
Block a user