Flow File

An integration is defined by a YAML flow file, the config the octo run and octo invoke commands load. This page is the canonical reference for every key in that file. Values documented as expressions are CEL, evaluated per message.

--config can also point at a directory: every .yaml/.yml file directly inside it (sorted by name, subdirectories ignored) is loaded and merged into a single config. The config's directory is the resource root that relative resource ids resolve against.

Files ending in _test.yaml are not config. They are test suites for the flows beside them, and the loader skips them.

Top-level keys

service:     # service identity
env:         # declared environment variables
resources:   # imported env files and templates
connectors:  # connector instances
processors:  # reusable named blocks
flows:       # the pipelines

service

service:
  name: orders-service
  environment: production

env

Declares the environment variables the config depends on. A variable must be declared here before any value may reference it as ${NAME}; referencing an undeclared variable fails the load with config references undeclared environment variable.

env:
  - name: HTTP_PORT
    default: "8080"
  - name: NOTION_TOKEN
    required: true

Resolution precedence: OS environment > .env file > default. The .env chain is ./.env (relative to the working directory), overlaid by the file named in $OCTO_ENV_FILE when set, overlaid by any resources.env files in listed order. A referenced variable that resolves to no value and has no default fails the load.

Substitution: ${NAME} references are rewritten in every value in the document: settings at any depth, a block's condition, a flow's workers, service.name. The two exceptions are the env: and resources: sections themselves, which are read to build the environment in the first place and so must be written literally.

A value that is exactly one placeholder is replaced with the variable's natural YAML type, so port: ${HTTP_PORT} fills an integer setting and workers: ${FLOW_WORKERS} an integer flow field. Quoting the placeholder makes no difference. A placeholder embedded in a longer string is substituted textually and stays a string.

In expressions: the resolved variables are also readable in CEL as env.NAME; see Octo Extensions. Inside an expression setting, prefer env.NAME over ${NAME}. A value substituted into an expression lands as raw CEL source (database: '"${NOTION_DATABASE_ID}"' produces a CEL string literal).

connectors

Connector instances own the machinery flows use to talk to the outside world (HTTP servers, database pools, API clients). Each entry instantiates one connector type under a unique name. Flows reference the name, so the same type can be instantiated several times with different settings.

connectors:
  - name: api
    type: http
    settings:
      port: ${HTTP_PORT}
  - name: orders-db
    type: database
    settings:
      driver: sqlite
      dsn: file:orders.db

processors

Reusable, named block definitions: declared once, referenced from any flow with ref. Names must be unique.

A referencing block takes its type and base settings from the definition; any settings on the block itself override the referenced ones key-by-key (shallow merge). A block sets either ref or type, not both. The one allowed overlap is an inline type equal to the referenced definition's type.

processors:
  - name: audit
    type: log
    settings:
      logger: out
      message: '"processed event " + eventID'

flows:
  - name: example
    process:
      - ref: audit                    # use as-is
      - ref: audit                    # override one setting
        settings:
          message: '"done: " + eventID'

flows

Each entry under flows is a root flow: a source feeding a chain of blocks, run by a pool of workers.

source

flows:
  - name: submit-order
    source:
      connector: api
      type: http
      settings:
        path: /orders
    process:
      - type: log
        settings: { message: '"received " + toJson(body)' }

error

The root flow's error path. When the process chain returns an error, the runtime exposes the failure as vars.error (an object with message, flow, and block) and runs this chain with the failing message. If the error chain succeeds, its output becomes the flow's result (recovery); for an HTTP-sourced flow, set vars.httpStatus to control the response status.

    error:
      - type: set-variable
        settings: { name: httpStatus, value: "502" }
      - type: set-payload
        settings:
          value: '{"error": vars.error.message}'

error (like workers, buffer, pool, and source) is valid on root flows only. For inline recovery around a few steps, use the handle-errors block instead. See Error Handling.

Block entries

Each step under process (or any sub-flow) is a block. Every block shares four base fields:

Leaf blocks (log, set-payload, sql, rest, jwt-validate, the Notion and Slack blocks, …) use only these four fields; everything else goes under settings.

Composite blocks additionally use top-level keys, called slots, that hold expressions or sub-flows. By convention slots sit on the block entry itself rather than under settings; the runtime folds the two together, so either spelling decodes the same way, and a key given in both places is an error:

- type: if
  name: any-orders
  condition: "size(body.orders) > 0"   # slot: top-level, not under settings
  then:                                # slot: a sub-flow
    process:
      - type: log
        settings:                      # leaf settings stay under settings
          message: '"processing orders"'

Across the composites the slots are process, error, branches, condition, then, else, cases, default, items, as, mode, body, setBody, setVars, key, ttl, rules, onReject, rejectStatus, delimiter, chunkSize, onError, buildResponse, correlation, strategy, expression, completionSize, completionTimeout, completionExpression, timeoutFrom, storeKey, maxGroups, onOverflow, connector, prompt, guardrail, routes, tools, skills, maxIterations, maxAttempts, serverName, resources, prompts, memoryThreadId, contextMaxTokens, memoryCompaction, and the further ai-agent memory, streaming and authorization slots listed on AI. Each composite accepts only its own: a key it does not declare fails the build (e.g. block "if": decode settings: json: unknown field "thne"). A leaf block reads only the settings it declares, so a slot written on one is ignored rather than refused.

Sub-flows

A slot documented as a sub-flow holds the same shape as a flow (an optional name and a process list) but must not declare source, workers, buffer, pool, or error; those are root-flow-only, enforced at build time. Composition recurses without limit: any block chain may contain more composites.

handle-errors

Inline recovery: the process block list is the happy path, and the error block list, also required, runs on failure with vars.error set. See Control Flow.

fork

branches holds a list of sub-flows, each run in parallel on a copy of the message, scheduled on the root flow's shared pool. See Control Flow.

if

condition is a boolean expression; then is the sub-flow run when it holds and the optional else the one run when it does not. See Control Flow.

switch

cases is an ordered list, each case a when expression plus inline sub-flow fields (name, process); the optional default sub-flow runs when no case matched. See Control Flow.

foreach

items evaluates to a list and body runs once per element, bound to vars.<as> (as defaults to item). mode is iterate (the default) or map. See Control Flow.

enrich

body runs on an isolated scope of the message; setBody and setVars name what comes back. See Control Flow.

validate

A filter: every rule in rules must hold or the message is rejected with rejectStatus, or with the onReject sub-flow when one is set. See Control Flow.

The jwt-validate block is also a filter but is a leaf: its rejection response is configured through its settings, not these slots.

cache-scope

body runs on a cache miss and its result is stored under key for ttl. See Control Flow.

split and aggregate are composites too, and take their slots the same way; see Control Flow.

ai-router

An LLM picks one of the named routes for the message, or the default sub-flow when the guardrail applies. Slots: connector, prompt, guardrail, routes, default. See AI.

ai-agent

An LLM agent that calls the flows in tools in a loop until it produces a result. Slots: connector, prompt, guardrail, tools, skills, default, maxIterations, memoryThreadId, contextMaxTokens, memoryCompaction. See AI and AI Agents.

ai-retry

Runs process; on failure an LLM revises the message and the chain re-runs, up to maxAttempts, before falling through to error. See AI.

mcp-router

Serves the flow as an MCP server: tools, resources and prompts, named by serverName. At least one tool, resource or prompt is required. See AI and MCP Server.

resources

Declares the external resources the config imports. Resource ids are paths relative to the config's directory (the resource root); an id that escapes the root with .. is rejected. Every declared resource is loaded when the config loads, so a missing or malformed template fails at deployment, not when a message first uses it.

Each templates entry:

resources:
  env:
    - config/extra.env
  templates:
    - resource: templates/receipt.tmpl
      as: receipt

Complete example

A single flow exercising every structural feature: declared env, resources, connectors, a reusable processor, a sourced root flow with tuning fields and an error path, and the core composites. This config loads and runs as-is (create config/extra.env and templates/receipt.tmpl next to it):

service:
  name: orders-service
  environment: production

env:
  - name: HTTP_PORT
    default: "8080"                      # used when not set by OS env or .env
  - name: PAYMENTS_URL
    default: https://payments.example.com
  - name: DB_DRIVER
    default: sqlite
  - name: DB_DSN
    default: file:orders.db

resources:
  env:
    - config/extra.env                   # overlays ./.env; relative to this file's dir
  templates:
    - resource: templates/receipt.tmpl
      as: receipt                        # referenced as templateResource("receipt")

connectors:
  - name: api                            # flows reference this name, not the type
    type: http
    settings:
      port: ${HTTP_PORT}                 # exact placeholder -> typed (int) value
  - name: payments
    type: http-client
    settings:
      baseURL: ${PAYMENTS_URL}
      timeout: 10s
  - name: orders-db
    type: database
    settings:
      driver: ${DB_DRIVER}
      dsn: ${DB_DSN}
  - name: out
    type: logger
    settings:
      format: json
      level: info

processors:
  - name: audit                          # reusable block, used via `ref` below
    type: log
    settings:
      logger: out
      level: info
      message: '"processed event " + eventID'

flows:
  - name: submit-order
    workers: 4                           # default 8
    buffer: 128                          # default 64
    pool: 8                              # shared pool for concurrent composites
    source:
      connector: api                     # connector instance name
      type: http
      settings:
        path: /orders
        correlationIdHeader: X-Request-Id
    process:
      # validate: filter, all rules must hold or the flow stops with 422
      - type: validate
        name: check-order
        rules:
          - expr: has(body.item)
            message: item is required
          - expr: 'body.amount > 0.0'
            message: amount must be positive
        rejectStatus: 422

      # if/else: slots are top-level keys, sub-flows have their own `process`
      - type: if
        name: needs-review
        condition: 'body.amount > 1000.0'
        then:
          process:
            - type: set-variable
              settings: { name: review, value: "true" }
        else:
          process:
            - type: set-variable
              settings: { name: review, value: "false" }

      # enrich: fetch on an isolated copy, write back only chosen vars
      - type: enrich
        name: load-customer
        body:
          process:
            - type: sql
              settings:
                connector: orders-db
                query: "SELECT 1 AS tier"
                single: true
        setVars:
          customerTier: body.tier

      # handle-errors: inline recovery, failure exposed as vars.error
      - type: handle-errors
        name: charge-safely
        process:
          - type: rest
            name: call-payments
            settings:
              connector: payments
              method: POST
              path: /charges
              body: '{"amount": body.amount}'
        error:
          - type: set-payload
            settings:
              value: '{"item": body.item, "amount": body.amount, "status": "degraded", "reason": vars.error.message}'

      # fork: branches run in parallel on message copies
      - type: fork
        name: notify
        branches:
          - name: log-branch
            process:
              - ref: audit
          - name: receipt-branch
            process:
              - type: log
                settings:
                  logger: out
                  message: 'templateResource("receipt")'

      # switch: first true case wins, else default
      - type: switch
        name: classify
        cases:
          - when: 'vars.review == "true"'
            process:
              - type: log
                settings: { logger: out, message: '"order flagged for review"' }
        default:
          process:
            - type: log
              settings: { logger: out, message: '"order accepted"' }

      # foreach: iterate an expression's list, element bound to vars.tag
      - type: foreach
        name: each-tag
        items: 'has(body.tags) ? body.tags : []'
        as: tag
        body:
          process:
            - type: log
              settings:
                logger: out
                message: '"tag: " + vars.tag'

      # cache-scope: body runs on a miss, result cached under the key
      - type: cache-scope
        name: fx-rate
        key: '"usd-eur"'
        ttl: 5m
        body:
          process:
            - type: set-payload
              settings:
                value: '{"rate": 0.92, "amount": body.amount}'

    # flow-level error path (root only): recovery + response status
    error:
      - type: set-variable
        settings: { name: httpStatus, value: "502" }
      - type: set-payload
        settings:
          value: '{"error": vars.error.message, "flow": vars.error.flow}'

Exercise it without binding the HTTP port:

bin/octo invoke --config orders.yaml --flow submit-order \
  --data '{"item": "widget", "amount": 50, "tags": ["a", "b"]}'

{"event_id":"5c12a0e3f47b4d19a6c8f0b21d3e4a97","body":{"amount":50,"rate":0.92}}