CoCoCo Workflow Engine Guide

Workflows are automated processes that execute a directed acyclic graph (DAG) of nodes. Each workflow can have multiple versions; the active version is the one that runs when triggered.

YAML Definition Structure

nodes:
  - id: "fetch_data"
    name: "Fetch Data"
    type: http_call
    config:
      url: "https://api.example.com/data"
      method: GET
    position:
      x: 100
      y: 100

  - id: "process"
    name: "Process Results"
    type: transform
    config:
      mapping:
        body: "{{ nodes.fetch_data.output.body }}"

  - id: "check_status"
    name: "Check Status"
    type: condition
    config:
      expression: "nodes.fetch_data.output.status == 200"

edges:
  - id: "e1"
    fromNodeId: "fetch_data"
    toNodeId: "check_status"
  - id: "e2"
    fromNodeId: "check_status"
    toNodeId: "process"
    condition: "true"

variables:
  - name: "api_url"
    type: "string"
    defaultValue: "https://api.example.com"
    description: "Base API URL"

Node Structure

Each node has:

• id — unique string identifier within the workflow
• name — human-readable display name
• type — node type (use list_node_types to see all available types)
• config — type-specific configuration object (use describe_node_type for schema)
• position — optional { x, y } for visual layout
• timeoutSeconds — optional execution timeout in seconds (number, e.g. 30)
• continueOnError — optional boolean, if true the workflow continues even if this node fails
• maxRetries — optional integer, number of times to retry on failure

A valid workflow is not the same as a correct config

Validation checks each node's config against its schema — required keys, types, enums. It does not reject keys a node does not have: an unrecognised key is dropped in silence, so a node configured with the wrong field names validates cleanly and then fails — or worse, runs and does nothing — at execution time. Take a node's field names from its own config schema — describe_node_type — rather than inferring them from a neighbouring node type.

The two HTTP nodes are not interchangeable

	http_call	http_request
target	any absolute URL	a device, addressed by deviceId
required config	url	deviceId
path	in the url	separate path field
fails when	the host HTTP policy rejects it	the device is not online

http_request has no url field at all. Reaching for it to call an external API is the natural mistake and it cannot work — use http_call.

An HTTP response body is a raw string

Both nodes put the body at nodes.<id>.output.body as a string, and the status code at nodes.<id>.output.status — not statusCode. A wrong field name here resolves to null silently, with no error and no hint, so it is worth reading twice.

To read a field of a JSON body, parse it first with a json_parse node (mode: parse, source: "{{ nodes.<id>.output.body }}") and read the parsed node's output. nodes.<id>.output.body.status is not a shorthand for that — it is null. Use csv_parse or yaml_parse for those wire formats.

Edge Structure

Edges connect nodes and define execution flow:

• id — unique string identifier
• fromNodeId — source node ID
• toNodeId — target node ID
• condition — optional string for conditional branching

Condition Node Edges

A condition node evaluates a workflow expression and branches:

• Edge with condition: "true" → taken when expression is truthy
• Edge with condition: "false" → taken when expression is falsy

Switch Node Edges

A switch node evaluates its cases in order — each is { name, expression } with a bare predicate such as nodes.classify.output.label == 'urgent' — and takes the first that is true:

• Edge with condition: "<case name>" → taken when that case is the first to match
• Edge with condition: "<config.default>" → the branch named in the node's default, taken when no case matches. Without a default, the node throws in that situation

Regular Edges

Edges without a condition are always followed (sequential flow).

Variable Structure

Workflow-level variables provide parameterization:

• name — variable identifier
• type — "string", "number", "boolean", "object", "array"
• defaultValue — optional default (JSON value)
• description — optional documentation

Expression Syntax

Workflows use a small expression dialect (a defined subset of Jinja). It is the single syntax for both predicates and value/text interpolation — there is no $.path syntax and no JSONLogic in node configs.

References (resolved against the execution context):

• input.key — workflow input parameters
• nodes.nodeId.output.field — a previous node's output (.output is required; nodeId may be a node id or its name)
• variables.name — workflow variable values

Missing references resolve to null. Member/index access works: input.items[0], nodes.n1.output.user.email.

Two ways strings are used in config:

• Predicate fields (condition/switch/assert expression) take a bare expression that is evaluated for truthiness, e.g. input.age >= 18 and input.role == 'admin'.
• Value / text fields (transform leaves, message, url, headers, …) use {{ … }} templates. A field that is exactly one {{ expr }} returns the typed value (object, number, array); mixed text interpolates to a string, e.g. "Order {{ input.orderId }} is {{ nodes.fetch.output.status }}". A string with no {{ }} is a literal.

Operators: == != < <= > >=, and or not, in / not in, arithmetic (+ - * / // %), string concat ~, and ternary a if cond else b.

Filters (value | name(args), chainable): default, length, min, max, lower, upper, trim, join, abs, round, and the structured filters merge, pick, omit, concat. Example: input.tags | join(', '), nodes.a.output | merge(nodes.b.output), input.name | default('anonymous').

Trigger filter config (EVENT / DEVICE_MQTT triggers, below) is a separate mechanism that still uses JSONLogic — it is not the workflow expression dialect.

DAG Requirement

The workflow graph must be a valid DAG — no cycles allowed. The validator checks this automatically. Nodes with no incoming edges are root nodes (executed first). Nodes with no outgoing edges are terminal nodes.

Trigger Types

Triggers start workflow executions automatically:

The input takes exactly one of these keys; the key decides the kind (TriggerConfigInput is @oneOf, there is no separate type field).

scheduled (cron)

config: { scheduled: { cronExpression: "0 9 * * MON-FRI", timezone: "Europe/Berlin", overlapPolicy: SKIP } }

• cronExpression — 5 or 6 whitespace-separated fields (minute hour day month weekday [year])
• timezone — IANA timezone (optional; omitted = the scheduler's default timezone)
• overlapPolicy — required: ALLOW, SKIP (skip if the previous run is still going) or CANCEL (cancel the previous run)

A scheduled run starts with an empty input. Anything time-dependent (for example "now") has to be computed inside the workflow — see Example 2.

event

config: { event: { topic: "job.created", dataQuery: "query($id: JobID!) { getJob(id: $id) { id name status } }" } }

• topic — <resource>.<verb>; the verbs are created, updated and deleted. listDataChangeEventTopics lists the topics this deployment publishes.
• dataQuery — optional GraphQL query run when the event fires. The event's resource ID is bound to $id, and the result is merged into the workflow input under data. Without it the input carries nothing about the record, so the workflow cannot tell which one changed.
• filter — optional JSONLogic expression. It is evaluated against the event itself, not against the record, so it cannot test fields such as status. Check record fields inside the workflow instead (on input.data).

deviceMqtt

config: { deviceMqtt: { topic: "sensors/temperature", filter: "{\">\": [{\"var\": \"mqtt.payload.temperature\"}, 85]}" } }

• topic — device-relative MQTT topic filter (+ and # allowed); a value starting with t/ is rejected
• deviceId — optional device ID (dev_…); omitted = every device in the tenant
• filter — optional JSONLogic expression, evaluated against the envelope { mqtt: { topic, deviceId, payload, … } } — so rules read mqtt.payload.<field>

In the workflow input, the keys of a JSON-object payload sit at the top level (input.temperature), and the envelope is under input.mqtt (input.mqtt.deviceId). input.payload is only set when the payload is not a JSON object.

edgeAppEvent

config: { edgeAppEvent: { controllerId: "ctl_...", edgeAppHandle: "my-edge-app", topic: "readings" } }

Fires on a bridge.publish from an edge app. All three keys are optional exact-match scopes; omit one to match any controller, edge app or topic.

folderWatch

config: { folderWatch: { folderId: "fld_...", prefix: "incoming/" } }

• folderId — the folder to watch
• prefix — folder-relative prefix watched for new files; the watch is shallow (no subdirectories)

A folderWatch trigger must use concurrencyPolicy: ALLOW and leave maxConcurrentExecutions unset.

Concurrency Control

All trigger types support:

• concurrencyPolicy — "ALLOW" (default), "QUEUE", "SKIP", "CANCEL"
• maxConcurrentExecutions — optional limit (null = unlimited)

MCP Workflow Tools

Workflow Management

• list_workflows — list all workflows
• read_workflow(id) — read workflow metadata + current version YAML
• import_workflow(workflowId, yaml) — create new version from YAML
• validate_workflow(yaml) — validate YAML without saving

Node Type Discovery

• list_node_types — list every node type this deployment has, with descriptions and categories
• describe_node_type(nodeType) — get full JSON schema for a node type's config

Execution Debugging

• get_workflow_execution(id) — see execution results, per-node outputs, and errors

Trigger Management (via GraphQL)

Use execute_graphql_query / execute_graphql_mutation for trigger CRUD. Use describe_type("CreateWorkflowTriggerInput") to explore the full input schema.

A trigger's config is a union. On the way out, __typename names the member and you select fields through an inline fragment; on the way in, the input takes exactly one key (it is @oneOf), and that key is what decides the kind. There is no separate type field to keep in step with it.

# List triggers for a workflow
query { workflowTriggers(workflowId: "wfl_...") {
  id name isEnabled
  config { __typename ... on ScheduledTriggerConfig { cronExpression timezone overlapPolicy } }
} }

# Read a trigger
query { workflowTrigger(id: "wft_...") { id name isEnabled config { __typename } } }

# Create a trigger
mutation { createWorkflowTrigger(input: {
  workflowId: "wfl_...", name: "Daily report",
  config: { scheduled: { cronExpression: "0 9 * * MON-FRI", timezone: "Europe/Berlin", overlapPolicy: SKIP } }
}) { trigger { id name isEnabled } errors { path message } } }

# Enable / disable
mutation { enableWorkflowTrigger(input: { id: "wft_..." }) { trigger { id isEnabled } errors { path message } } }

mutation { disableWorkflowTrigger(input: { id: "wft_..." }) { trigger { id isEnabled } errors { path message } } }

# Delete
mutation { deleteWorkflowTrigger(input: { id: "wft_..." }) { trigger { id } errors { path message } } }

Example 1: Fetch API → Check Status → Update Job

nodes:
  - id: fetch_order
    name: Fetch Order Status
    type: http_call
    config:
      url: "https://api.supplier.com/orders/{{ input.orderId }}"
      method: GET
      headers:
        Authorization: "Bearer {{ variables.api_token }}"

  # The response body is a raw string, so it has to be parsed before
  # any field of it can be read. Skipping this node is the mistake
  # that reads as `output.body.status` and silently yields null.
  - id: parse_order
    name: Parse Order JSON
    type: json_parse
    config:
      mode: parse
      source: "{{ nodes.fetch_order.output.body }}"

  - id: check_status
    name: Check If Ready
    type: condition
    config:
      expression: "nodes.parse_order.output.status == 'ready'"

  - id: update_job
    name: Update Job
    type: graphql
    config:
      operation: mutation
      query: |
        mutation($id: JobID!, $input: UpdateJobInput!) {
          updateJob(id: $id, input: $input) {
            job { id status }
            errors { path message }
          }
        }
      variables:
        id: "{{ input.jobId }}"
        input:
          status: "CONFIRMED"

  - id: log_skipped
    name: Log Not Ready
    type: log
    config:
      level: info
      message: "Order {{ input.orderId }} not ready yet: {{ nodes.parse_order.output.status }}"

edges:
  - id: e1
    fromNodeId: fetch_order
    toNodeId: parse_order
  - id: e1a
    fromNodeId: parse_order
    toNodeId: check_status
  - id: e2
    fromNodeId: check_status
    toNodeId: update_job
    condition: "true"
  - id: e3
    fromNodeId: check_status
    toNodeId: log_skipped
    condition: "false"

variables:
  - name: api_token
    type: string
    description: Supplier API bearer token

Example 2: Scheduled Report — Query Jobs → Transform → Send Email

nodes:
  # A scheduled run starts with an empty input, so "now" is computed here.
  - id: clock
    name: Current Time
    type: script
    config:
      inline: |
        return { now = ctx.time.nowIso() }

  - id: query_jobs
    name: Query Overdue Jobs
    type: graphql
    config:
      operation: query
      query: |
        query($now: ISO8601!) {
          listJobs(first: 100, filter: { status: { neq: "COMPLETED" }, dueAt: { lt: $now } }) {
            edges { node { id name dueAt status } }
          }
        }
      variables:
        now: "{{ nodes.clock.output.now }}"

  - id: build_report
    name: Build Report Data
    type: transform
    config:
      mapping:
        count: "{{ nodes.query_jobs.output.data.listJobs.edges | length }}"
        edges: "{{ nodes.query_jobs.output.data.listJobs.edges }}"
        subject: "Overdue Jobs Report"

  # A message node addresses a notification group or a single
  # recipient — it has no `to:` and no `channel:`. Both are dropped
  # silently and the node then throws "requires either 'groupId' or
  # 'recipientId'" at run time, after validation has passed.
  - id: send_email
    name: Send Report Email
    type: message
    config:
      groupId: "ngr_..."
      subject: "{{ nodes.build_report.output.subject }}"
      body: "There are {{ nodes.build_report.output.count }} overdue jobs. See dashboard for details."

edges:
  - id: e0
    fromNodeId: clock
    toNodeId: query_jobs
  - id: e1
    fromNodeId: query_jobs
    toNodeId: build_report
  - id: e2
    fromNodeId: build_report
    toNodeId: send_email

Trigger for this workflow (via GraphQL mutation):

mutation { createWorkflowTrigger(input: {
  workflowId: "wfl_...", name: "Daily 9am report",
  config: { scheduled: { cronExpression: "0 9 * * MON-FRI", timezone: "Europe/Berlin", overlapPolicy: SKIP } }
}) { trigger { id name isEnabled } errors { path message } } }

Example 3: MQTT Sensor → Script → Conditional Alert

nodes:
  # Lua is the only language, so there is no `language:` field, and
  # the source field is `inline`. The script's input is its first
  # vararg — not a global. For a root node it is the workflow input, and
  # a JSON-object MQTT payload arrives with its keys at the top level.
  - id: evaluate
    name: Check Threshold
    type: script
    config:
      inline: |
        local input = ...
        local temp = input.temperature
        local threshold = 85.0
        return {
          temperature = temp,
          exceeded = temp > threshold,
          severity = temp > 95 and "critical" or "warning"
        }

  - id: should_alert
    name: Alert Needed?
    type: condition
    config:
      expression: "nodes.evaluate.output.exceeded == true"

  - id: send_alert
    name: Send Alert
    type: mqtt_publish
    config:
      topic: "alerts/temperature"
      message: |
        {
          "severity": "{{ nodes.evaluate.output.severity }}",
          "temperature": "{{ nodes.evaluate.output.temperature }}",
          "deviceId": "{{ input.mqtt.deviceId }}"
        }

edges:
  - id: e1
    fromNodeId: evaluate
    toNodeId: should_alert
  - id: e2
    fromNodeId: should_alert
    toNodeId: send_alert
    condition: "true"

Typical Workflow Building Process

1. list_node_types → understand available node types
2. describe_node_type("http_call") → see config schema for nodes you need
3. Write YAML definition
4. validate_workflow(yaml) → check for errors
5. list_workflows → find the target workflow (or create one via GraphQL)
6. import_workflow(workflowId, yaml) → save as new version
7. setActiveVersion(workflowId:, versionId:) mutation → required.import_workflow only stores a version; it does not activate one, and an execution against a workflow with no active version fails with "No active version set for workflow".
8. createWorkflowTrigger mutation → set up trigger(s)