Skip to main content

Workflow Examples

Practical workflow examples demonstrating common patterns with correct node IDs, connection structures, and configuration approaches. All node types referenced here are from the actual workflow node catalog.

Example 1: MySQL to S3 Data Export​

Use Case: Export data from a MySQL database to S3 as a CSV file on a daily schedule.

Workflow Structure​

schedule → mysql → to-file → s3

Node Configuration​

schedule (trigger):

{
  "scheduleType": "cron",
  "cronExpression": "0 0 * * *",
  "timezone": "America/New_York"
}

mysql (source):

{
  "connectionType": "datasource",
  "dataSourceId": "<your-mysql-datasource-id>",
  "query": "SELECT id, name, email, created_at FROM customers WHERE created_at >= DATE_SUB(NOW(), INTERVAL 1 DAY)"
}

Input mapping: receives trigger output from the schedule node.

to-file (transform):

{
  "format": "csv",
  "filename": "customers-export"
}

Input mapping: the query results from the mysql node provide the data to convert. The node outputs the generated filename (with extension) and base64 content.

s3 (destination):

{
  "dataSourceId": "<your-s3-datasource-id>",
  "bucket": "data-exports",
  "prefix": "customers/"
}

Input mapping: receives the generated file from the to-file node and uploads it under the configured bucket and prefix.

Connections​

[
  { "source": "schedule", "target": "mysql" },
  { "source": "mysql", "target": "to-file" },
  { "source": "to-file", "target": "s3" }
]

Example 2: Webhook API with AI Processing​

Use Case: Receive a webhook request, process the data with an AI model, and return a structured response.

Workflow Structure​

webhook → llm → set-fields → webhook-response

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "POST"
}

The webhook URL is generated for you when the trigger is created.

llm (AI):

{
  "model": "<model-id-from-model-selector>",
  "defaultSystemPrompt": "Analyze the provided text and return a JSON object with: sentiment (positive/negative/neutral), summary (one sentence), and key_topics (array of strings).",
  "defaultTemperature": 0.3,
  "defaultResponseFormat": "json_object"
}

Input mapping: the userPrompt input is mapped from the webhook trigger output (the request body's text field).

set-fields (transform):

{
  "mode": "manual",
  "fields": [
    { "name": "analysis", "from": "response" },
    { "name": "model_used", "from": "model.name" }
  ]
}

Input mapping: the llm node's output provides the response and model fields referenced above.

webhook-response (destination):

{
  "statusCode": 200,
  "contentType": "application/json"
}

Input mapping: the set-fields output provides the response body.

Connections​

[
  { "source": "webhook", "target": "llm" },
  { "source": "llm", "target": "set-fields" },
  { "source": "set-fields", "target": "webhook-response" }
]

Example 3: PDF Document Processing Pipeline​

Use Case: Upload a PDF via webhook, extract text, generate embeddings, and store in a vector database.

Workflow Structure​

webhook → pdf-parser → text-chunker → embeddings → milvus

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "POST"
}

Receives the PDF file upload.

pdf-parser (transform):

{
  "outputFormat": "markdown",
  "extractTables": true
}

Input mapping: the file input comes from the webhook trigger's uploaded file. Outputs content (path to the extracted text).

text-chunker (transform):

{
  "chunkStrategy": "recursive",
  "chunkSize": 512,
  "chunkOverlap": 50
}

Input mapping: the textPath input is mapped from the pdf-parser content output. Outputs texts (chunk strings ready for the embeddings node) and chunkMetadata.

embeddings (AI):

{
  "model": "<embedding-model-id-from-model-selector>"
}

Input mapping: the texts input is mapped from the text-chunker texts output.

milvus (destination):

{
  "connectionType": "addon",
  "addonId": "<your-milvus-addon-id>",
  "collectionName": "documents",
  "embeddingField": "vector",
  "dimension": 1536,
  "metricType": "COSINE"
}

Input mapping: the vectors input is mapped from the embeddings node output; the optional metadata input is mapped from the text-chunker chunkMetadata output.

Connections​

[
  { "source": "webhook", "target": "pdf-parser" },
  { "source": "pdf-parser", "target": "text-chunker" },
  { "source": "text-chunker", "target": "embeddings" },
  { "source": "embeddings", "target": "milvus" }
]

Example 4: AI Agent with MCP Tools​

Use Case: An AI agent that can search the web and create GitHub issues based on user requests.

Workflow Structure​

webhook → react-agent → webhook-response
             |     |
           [llm] [mcp-tools-provider]
           ("ai" connector)  ("tools" connector)

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "POST"
}

react-agent (agent):

{
  "systemPrompt": "You are a helpful assistant that can search the web and create GitHub issues. When the user asks about a topic, search the web first. When they report a bug, create a GitHub issue.",
  "maxIterations": 5
}

Input mapping: the task input is mapped from the webhook request body's message field. The model itself is not configured on the agent; it comes from the LLM node connected to the agent's ai connector.

llm (AI):

{
  "model": "<model-id-from-model-selector>"
}

Connected to the react-agent via the ai dependency connector.

mcp-tools-provider (operator):

{
  "mcpServerId": "brave-search",
  "mcpServerIds": ["github"],
  "filterTools": [],
  "cacheTimeout": 300
}

Connected to the react-agent via the tools connector (not the regular data flow).

webhook-response (destination):

{
  "statusCode": 200,
  "contentType": "application/json"
}

Input mapping: the react-agent output (finalAnswer, success, stopReason) provides the response body.

Connections​

[
  { "source": "webhook", "target": "react-agent" },
  { "source": "llm", "target": "react-agent", "type": "ai" },
  { "source": "mcp-tools-provider", "target": "react-agent", "type": "tools" },
  { "source": "react-agent", "target": "webhook-response" }
]

Example 5: Conditional Routing with Switch-Case​

Use Case: Receive webhook events, screen them, route them based on event type, and process each type differently.

Workflow Structure​

webhook → guardrails → switch-case → [branch: api]
                                    → [branch: slack]
                                    → [branch: smtp]

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "POST"
}

guardrails (evaluation):

{
  "action": "block",
  "toxicityCheck": true,
  "maxLength": 10000
}

Screens the incoming payload for unsafe content and oversized input before routing. Guardrails performs content validation (PII detection, toxicity checking, custom regex rules), not schema validation.

Input mapping: the webhook request body.

switch-case (control-flow):

{
  "switchField": "event_type",
  "cases": [
    { "name": "api", "values": ["order.created"] },
    { "name": "slack", "values": ["alert.triggered"] },
    { "name": "email", "values": ["user.signup"] }
  ],
  "defaultCase": "api",
  "matchMode": "first"
}

Input mapping: the guardrails output.

api (for the order.created branch):

{
  "endpointSource": "config",
  "url": "https://api.internal.example.com/orders",
  "method": "POST"
}

Input mapping: the event payload from switch-case provides the request body.

slack (for the alert.triggered branch):

{
  "credentials": "<your-slack-credentials-id>",
  "operation": "sendMessage",
  "channel": "#alerts"
}

Input mapping: the event payload's message from switch-case provides the text input.

smtp (for the user.signup branch):

{
  "host": "smtp.example.com",
  "port": 587,
  "secure": "tls",
  "username": "<smtp-username>",
  "password": "<smtp-password>",
  "fromEmail": "noreply@example.com",
  "to": "welcome@example.com",
  "subject": "New User Signup",
  "bodyType": "html",
  "body": "A new user signed up."
}

Input mapping: the event payload from switch-case.

Connections​

[
  { "source": "webhook", "target": "guardrails" },
  { "source": "guardrails", "target": "switch-case" },
  { "source": "switch-case", "target": "api", "label": "order.created" },
  { "source": "switch-case", "target": "slack", "label": "alert.triggered" },
  { "source": "switch-case", "target": "smtp", "label": "user.signup" }
]

Example 6: Scheduled Report with Database and Email​

Use Case: Generate a weekly report from PostgreSQL data, archive it to S3, and email it to stakeholders.

Workflow Structure​

schedule → postgresql → report-builder → s3
                                       → smtp

Node Configuration​

schedule (trigger):

{
  "scheduleType": "cron",
  "cronExpression": "0 8 * * 1",
  "timezone": "America/New_York"
}

Runs every Monday at 8 AM.

postgresql (source):

{
  "connectionType": "datasource",
  "dataSourceId": "<your-postgresql-datasource-id>",
  "query": "SELECT date_trunc('day', created_at) as day, COUNT(*) as orders, SUM(amount) as revenue FROM orders WHERE created_at >= NOW() - INTERVAL '7 days' GROUP BY 1 ORDER BY 1"
}

report-builder (transform):

{
  "layoutMode": "simple",
  "outputFormat": "pdf",
  "title": "Weekly Sales Report",
  "includeTotals": true
}

Input mapping: the query rows from the postgresql node.

s3 (destination):

{
  "dataSourceId": "<your-s3-datasource-id>",
  "bucket": "reports",
  "prefix": "weekly/"
}

Input mapping: receives the generated report file from report-builder and archives it.

smtp (destination):

{
  "host": "smtp.example.com",
  "port": 587,
  "secure": "tls",
  "username": "<smtp-username>",
  "password": "<smtp-password>",
  "fromEmail": "reports@example.com",
  "to": "team@example.com",
  "subject": "Weekly Sales Report",
  "bodyType": "html",
  "body": "Please find attached the weekly sales report.",
  "attachInputFiles": true
}

Input mapping: receives the generated report file from report-builder; with attachInputFiles enabled (the default), incoming files are attached to the email.

Connections​

[
  { "source": "schedule", "target": "postgresql" },
  { "source": "postgresql", "target": "report-builder" },
  { "source": "report-builder", "target": "s3" },
  { "source": "report-builder", "target": "smtp" }
]

Example 7: Parallel API Aggregation​

Use Case: Call multiple APIs in parallel, aggregate the results, and return a combined response.

Workflow Structure​

webhook → set-fields → parallel-branch → webhook-response
                            |
                          [api]
                    (branch handler connector)

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "GET"
}

set-fields (transform):

{
  "mode": "json",
  "jsonData": "{ \"branches\": [ { \"name\": \"service-a\", \"url\": \"https://api.service-a.com/data\" }, { \"name\": \"service-b\", \"url\": \"https://api.service-b.com/data\" } ] }"
}

Builds the branches array that parallel-branch fans out.

parallel-branch (control-flow):

{
  "joinStrategy": "all",
  "aggregateResults": "array",
  "failOnError": true
}

Input mapping: the branches input is mapped from the set-fields output. Each branch entry is passed to the branch handler node; the join strategy (all, any, first-N, majority) controls when the node completes, and the aggregated results are exposed on its data output (with per-branch detail in branchResults).

api (branch handler):

{
  "endpointSource": "input",
  "method": "GET"
}

Connected to the parallel-branch node's branch (Branch Handler) dependency connector. With Endpoint Source set to Dynamic URL, each branch's url drives the request.

webhook-response (destination):

{
  "statusCode": 200,
  "contentType": "application/json"
}

Input mapping: the parallel-branch data output provides the combined results.

Connections​

[
  { "source": "webhook", "target": "set-fields" },
  { "source": "set-fields", "target": "parallel-branch" },
  { "source": "api", "target": "parallel-branch", "type": "branch" },
  { "source": "parallel-branch", "target": "webhook-response" }
]

Example 8: RAG Question-Answering​

Use Case: A RAG pipeline that retrieves relevant documents from a vector store and generates answers.

Workflow Structure​

webhook → semantic-memory (search) → rag → llm → webhook-response
               |
          [embeddings]
          ("ai" connector)

Node Configuration​

webhook (trigger):

{
  "provider": "generic",
  "method": "POST"
}

semantic-memory (memory):

{
  "operation": "search",
  "vectorStoreAddonId": "<your-milvus-addon-id>",
  "collection": "company-docs",
  "topK": 5
}

Input mapping: the query input is mapped from the webhook request body's question field. The node's ai dependency connector must be connected to an Embeddings node, which is used to embed the query.

rag (operator, RAG Prompt Builder):

{
  "query": "",
  "topK": 5
}

Input mapping: the retrievedDocs input is mapped from the semantic-memory results output; the query input is mapped from the webhook request body's question field (overriding the config value). Outputs rag_prompt, the formatted prompt combining context and question.

llm (AI):

{
  "model": "<model-id-from-model-selector>",
  "defaultSystemPrompt": "Answer the user's question based on the provided context. If the context does not contain the answer, say so."
}

Input mapping: the userPrompt input is mapped from the rag node's rag_prompt output.

webhook-response (destination):

{
  "statusCode": 200,
  "contentType": "application/json"
}

Input mapping: the llm response output provides the generated answer.

Connections​

[
  { "source": "webhook", "target": "semantic-memory" },
  { "source": "embeddings", "target": "semantic-memory", "type": "ai" },
  { "source": "semantic-memory", "target": "rag" },
  { "source": "rag", "target": "llm" },
  { "source": "llm", "target": "webhook-response" }
]

Common Workflow Patterns​

Pattern: Data Pipeline (ETL)​

[trigger] → [source node] → [transform node(s)] → [destination node]

Node types used:

  • Triggers: schedule, webhook
  • Sources: mysql, postgresql, mongodb, s3
  • Transforms: set-fields, filter, sort, to-file, aggregate
  • Destinations: mysql, postgresql, mongodb, s3

Source and destination variants of the same system share the same node id; the palette category (source vs destination) determines the behavior.

Pattern: AI Processing​

[trigger] → [llm] → [destination]

Node types used:

  • AI: llm, embeddings, vision, text-to-speech, speech-to-text
  • Supporting: set-fields for prompt construction, guardrails for content validation

Pattern: Agent with Tools​

[trigger] → [agent node] → [destination]
              |    |    |
           [llm] [memory] [mcp-tools-provider]
           (ai)  (memory)  (tools connector)

Node types used:

  • Agents: react-agent, agent-loop, supervisor-agent
  • Operators: rag, function-calling, tool-router
  • Tool providers: mcp-tools-provider, web-search, code-interpreter, calculator
  • Memory: context-buffer, knowledge-base, semantic-memory

Pattern: Event-Driven Routing​

[webhook] → [guardrails] → [switch-case] → [branch A]
                                          → [branch B]
                                          → [branch C]

Node types used:

  • Validation: guardrails
  • Routing: switch-case, conditional
  • Actions: api, slack, smtp, mongodb

Pattern: Batch Processing with Loop​

[schedule] → [source] → [loop] → [process each item] → [merge] → [destination]

Node types used:

  • Iteration: loop, map
  • Accumulation: merge (collects per-iteration outputs into one array)
  • Transform: set-fields, filter, code
  • Control: while-loop, retry

Configuration Patterns​

Data Source Connection​

Nodes that connect to external databases use the connectionType and dataSourceId pattern:

{
  "connectionType": "datasource",
  "dataSourceId": "<id-from-data-sources-page>"
}

Add-on Connection​

Nodes that support platform add-ons use:

{
  "connectionType": "addon",
  "addonId": "<id-from-addons-page>"
}

Input Mapping (JSONPath)​

Nodes reference data from upstream nodes using JSONPath expressions:

  • $.output - The full output object from the connected upstream node
  • $.output.rows - A specific field from the upstream output
  • $.output.body.text - Nested field access
  • $.metadata.processedAt - Metadata from the upstream node

Direct Data Source Selection (Special Case)​

Some nodes select their connection directly without a connectionType field. For example, the mongodb source takes only a data source, an operation, and a collection:

{
  "dataSourceId": "<your-mongodb-datasource-id>",
  "operation": "find",
  "database": "app",
  "collection": "users"
}

The s3 nodes and pinecone-dest follow the same pattern (dataSourceId directly, no connectionType).

Best Practices​

Node Selection​

  1. Use specific nodes over generic ones (e.g., postgresql over api for database queries)
  2. Use native destinations over MCP tools for deterministic operations (e.g., the slack destination over an MCP Slack server)
  3. Use agents when the workflow needs dynamic decision-making
  4. Use control flow nodes (switch-case, loop, parallel-branch) to build complex logic

Error Handling​

  1. Add guardrails nodes to screen content before processing
  2. Use conditional nodes to check for error conditions
  3. Configure retry nodes for transient failures
  4. Use stop-error to halt execution with a clear error message

Performance​

  1. Use parallel-branch for independent operations that can run concurrently
  2. Use filter early to reduce data volume before expensive operations
  3. Set appropriate timeouts in node configuration
  4. Use set-fields to select only needed fields before passing to downstream nodes

Data Flow​

  1. Use descriptive node labels to document the workflow
  2. Keep workflows linear when possible; use branching only when needed
  3. Use merge to accumulate loop iteration outputs into a single array; parallel-branch joins its own branches via its join strategy
  4. Use set-fields to reshape data between nodes with different schemas

Next Steps​