Edge Delta HTTP Workflow Source

Overview

The HTTP Workflow source chains multiple HTTP requests into a single pipeline node. Each step in the workflow can read data from previous steps, so you can authenticate with one API and then use the returned token to call another.

Common use cases include:

• Authenticate first, then fetch data — for example, get an OAuth token and pass it to a protected API
• Get a list, then fetch details for each item — fan-out to a detail endpoint for every item in a list response
• React to incoming webhooks — receive a webhook and enrich the payload with additional API calls
• Cache tokens or responses — store frequently used values in Redis to reduce API calls

Use the configuration wizard at the bottom of this page to generate a starter YAML configuration.

HTTP Workflow vs. HTTP Pull

If your integration only needs to call one endpoint on a schedule, HTTP Pull is simpler. Use HTTP Workflow when you need multiple requests that depend on each other.

• Outgoing data types: log
• Minimum version: v2.8.0

Quick Start: Your First Workflow

Poll a Public API

The simplest workflow has one step and a polling interval:

nodes:
- name: github_status
  type: http_workflow_input
  workflow_pull_interval: 5m
  steps:
    - name: get_status
      endpoint: https://www.githubstatus.com/api/v2/status.json

This workflow runs every 5 minutes, calls the GitHub status API, and emits the JSON response as a log item into your pipeline.

Add Authentication

Most APIs require authentication. Use the {{ SECRET name }} syntax to reference a secret stored in Edge Delta without exposing it in your configuration:

nodes:
- name: my_api_workflow
  type: http_workflow_input
  workflow_pull_interval: 5m
  steps:
    - name: fetch_data
      endpoint: https://api.example.com/v1/data
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"
        Accept: application/json

In this example, {{ SECRET api_token }} is replaced at runtime with the value of a secret named api_token. Create and manage secrets in the Edge Delta app under Settings > Secrets.

Core Concepts

HTTP Workflow is built on three building blocks: a trigger that starts the workflow, a sequence of steps that make HTTP requests, and expressions that pass data between steps.

1. Triggers: What Starts the Workflow

Every workflow needs exactly one trigger that determines when it runs. Choose one of the following:

2. Steps: The Requests in Your Workflow

Each step defines one HTTP request. Steps run in order from top to bottom, and each step can read data returned by any step that ran before it.

steps:
  - name: step_one    # Runs first
    endpoint: https://api.example.com/auth

  - name: step_two    # Runs second — reads the token returned by step_one
    endpoint: https://api.example.com/data
    header_expressions:
      Authorization: Concat(["Bearer ", steps["step_one"]["response"]["token"]], "")

In this example, step_two builds its Authorization header dynamically using the token value from step_one’s response body.

Note: Step names must use snake_case with no spaces. Use underscores instead of spaces: get_user_list not Get User List.

3. Expressions: Passing Data Between Steps

Later steps can read data from earlier steps using OTTL expressions. These expressions are used in fields like endpoint_expression, header_expressions, and request_body_expression to build dynamic values at runtime.

Every completed step stores its response in a steps array. You access a step’s data by referencing its name or its zero-based index (the first step is 0, the second is 1, and so on).

Accessing step results

Accessing variables and environment values

Building dynamic strings with Concat

Use Concat to combine static text with expression values. The function takes an array of parts and a separator:

# Build a URL using a value from a previous step
endpoint_expression: Concat(["https://api.example.com/users/", steps["get_user"]["response"]["id"], "/profile"], "")

# Build an Authorization header using a token from step 1
header_expressions:
  Authorization: Concat(["Bearer ", steps["get_token"]["response"]["access_token"]], "")

The second argument ("") is the separator inserted between parts — an empty string means the parts are joined directly.

Warning: All expressions using Concat, steps[], etc. must be written on a single line. Multi-line expressions fail validation.

Common Patterns

Pattern 1: OAuth Token + Data Fetch

Many APIs require an OAuth token before you can access data. This pattern uses two steps: the first step authenticates and receives a token, and the second step passes that token in an Authorization header to fetch the actual data. An optional Redis cache avoids re-authenticating on every run.

nodes:
- name: oauth_workflow
  type: http_workflow_input
  workflow_pull_interval: 15m

  # Redis for token caching (optional but recommended)
  redis_address: "redis://localhost:6379"

  # Global variables for reuse (secrets resolved at runtime)
  variables:
    client_id: "{{ SECRET oauth_client_id }}"
    client_secret: "{{ SECRET oauth_client_secret }}"

  steps:
    # Step 1: Get OAuth token (cached for 50 minutes)
    - name: get_token
      endpoint: https://auth.example.com/oauth/token
      method: POST
      headers:
        Content-Type: application/x-www-form-urlencoded
      request_body_expression: Concat(["grant_type=client_credentials&client_id=", variables["client_id"], "&client_secret=", variables["client_secret"]], "")
      redis_cache:
        key_expression: '"oauth-token"'
        ttl: 50m
        check_before_request: true
        populate_on_success: true

    # Step 2: Fetch data using the token
    - name: fetch_data
      endpoint: https://api.example.com/v1/data
      header_expressions:
        Authorization: Concat(["Bearer ", steps["get_token"]["response"]["access_token"]], "")

Why cache the token?

• OAuth tokens typically last 1 hour
• Caching for 50 minutes avoids re-authenticating every request
• If cache is empty or expired, the HTTP request runs automatically

Pattern 2: Webhook-Triggered Processing

Instead of polling on a schedule, you can configure the workflow to listen for incoming HTTP requests. When an external system sends a webhook, the workflow runs its steps using the webhook payload as input. This is useful for event-driven integrations where you want to enrich or forward incoming data in real time.

nodes:
- name: alert_webhook
  type: http_workflow_input

  webhook:
    path: /webhooks/alerts
    port: 8888
    auth_type: bearer
    bearer_token: "{{ SECRET webhook_bearer_token }}"
    rate_limit_enabled: true
    rate_limit_max_per_min: 100

  steps:
    # Access webhook payload with webhook_payload["field"]
    - name: enrich_alert
      endpoint_expression: Concat(["https://api.example.com/users/", webhook_payload["user_id"]], "")
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"

    - name: send_to_slack
      endpoint: "https://hooks.slack.com/services/{{ SECRET slack_webhook_path }}"
      method: POST
      headers:
        Content-Type: application/json
      request_body_expression: Concat(["{\"text\":\"Alert from ", steps["enrich_alert"]["response"]["username"], ": ", webhook_payload["message"], "\"}"], "")

Webhook authentication options:

Pattern 3: Fan-Out (Get List, Then Details)

A fan-out workflow fetches a list of items in one step, then iterates over that list to make a separate API call for each item. For example, you might fetch a list of user IDs and then call a profile endpoint for each one. The iterator configuration controls how many requests run in parallel.

nodes:
- name: user_details_workflow
  type: http_workflow_input
  workflow_pull_interval: 30m

  steps:
    # Step 1: Get list of users
    - name: get_users
      endpoint: https://api.example.com/users?limit=50
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"

    # Step 2: For each user, get their profile
    - name: get_profile
      endpoint_expression: Concat(["https://api.example.com/users/", iterator_item["id"], "/profile"], "")
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"
      is_last_step: true
      iterator:
        source: steps["get_users"]["response"]["users"]
        variable_name: iterator_item
        max_parallel: 10
        continue_on_error: true

Inside an iterator step, these variables are available:

The default variable name is iterator_item, but you can change it with variable_name. For example, if you set variable_name: user, reference the current item as user["id"] instead of iterator_item["id"].

Warning: The variable name in your expressions must exactly match the variable_name you configure. Mismatched names cause silent failures where the expression resolves to nil.

Warning: With max_parallel: 10, the workflow makes 10 concurrent requests. If the target API has rate limits, lower this value or add retry_http_code: [429] so the agent retries when rate-limited.

Fan-Out with Per-Item Caching

You can combine fan-out with Redis caching so that each item gets its own cache entry. On subsequent runs, items that are already cached are served from Redis and only new or expired items trigger API calls.

nodes:
- name: cached_fan_out
  type: http_workflow_input
  workflow_pull_interval: 30m
  redis_address: "redis://localhost:6379"

  steps:
    - name: get_users
      endpoint: https://api.example.com/users
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"

    - name: get_profile
      endpoint_expression: Concat(["https://api.example.com/users/", iterator_item["id"], "/profile"], "")
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"
      is_last_step: true
      iterator:
        source: steps["get_users"]["response"]["users"]
        variable_name: iterator_item
        max_parallel: 10
      # Cache each user's profile by their ID
      redis_cache:
        key_expression: Concat(["user-profile-", iterator_item["id"]], "")
        ttl: 1h
        check_before_request: true      # Check cache first
        populate_on_success: true       # Cache new responses

Result: On the first run, all 100 profiles are fetched. On subsequent runs within the TTL, only new/changed users trigger API calls—cached profiles are reused.

Pattern 3b: Caching Expensive API Calls

If an API is rate-limited, slow, or charges per call, use Redis caching to reduce how often the workflow makes requests. When a cached response exists and has not expired, the agent uses the cached value and skips the HTTP call entirely.

nodes:
- name: expensive_api_workflow
  type: http_workflow_input
  workflow_pull_interval: 5m           # Check every 5 minutes
  redis_address: "redis://localhost:6379"

  steps:
    # This API costs $0.01 per call - cache for 1 hour
    - name: fetch_expensive_data
      endpoint: https://api.expensive-service.com/data
      headers:
        Authorization: "Bearer {{ SECRET expensive_api_key }}"
      redis_cache:
        key_expression: '"expensive-data-cache"'
        ttl: 1h                         # Cache for 1 hour
        check_before_request: true      # Always check cache first
        populate_on_success: true       # Store successful responses

    # This API is rate-limited to 10 req/min - cache for 5 minutes
    - name: fetch_rate_limited
      endpoint: https://api.strict-limits.com/metrics
      headers:
        Authorization: "Bearer {{ SECRET metrics_api_key }}"
      redis_cache:
        key_expression: '"metrics-cache"'
        ttl: 5m
        check_before_request: true
        populate_on_success: true

Cost savings example:

Pattern 4: Conditional Execution

Use run_condition to control whether a step runs based on the results of a previous step. If the condition evaluates to false, the step is skipped and the workflow continues to the next step.

steps:
  - name: check_status
    endpoint: https://api.example.com/status

  # Only runs if status is "active"
  - name: fetch_active_data
    endpoint: https://api.example.com/active-data
    run_condition: steps["check_status"]["response"]["status"] == "active"

  # Only runs if status is NOT "active"
  - name: send_alert
    endpoint: https://alerts.example.com/webhook
    method: POST
    run_condition: steps["check_status"]["response"]["status"] != "active"
    request_body: '{"message": "Service is not active"}'

Pattern 5: State Persistence (Markers/Cursors)

You can persist state between workflow runs using Redis-only steps — steps that interact with Redis but make no HTTP request. This lets you remember where you left off between runs. Common use cases include:

• Pagination cursors — store the last page or offset so the next run picks up where the previous one stopped
• Timestamps — record the timestamp of the most recent record so you only fetch new data
• Deduplication — track processed record IDs to avoid duplicates

nodes:
- name: incremental_sync
  type: http_workflow_input
  workflow_pull_interval: 5m
  redis_address: "redis://localhost:6379"

  steps:
    # Step 1: Read last processed timestamp from Redis (no HTTP call)
    - name: get_marker
      redis_cache:
        key_expression: '"last-sync-timestamp"'
        check_before_request: true

    # Step 2: Fetch new records since the marker
    - name: fetch_new_records
      endpoint: https://api.example.com/records
      parameter_expressions:
        since: steps["get_marker"]["response"]
        limit: '"100"'
      headers:
        Authorization: "Bearer {{ SECRET api_token }}"

    # Step 3: Save the new marker to Redis (no HTTP call)
    - name: save_marker
      redis_cache:
        key_expression: '"last-sync-timestamp"'
        value_expression: steps["fetch_new_records"]["response"]["latest_timestamp"]
        populate_on_success: true
        ttl: 720h  # 30 days

How Redis-only steps work:

Note: Steps with redis_cache configured but no endpoint are Redis-only steps. They don’t make HTTP calls—they only interact with Redis. Use value_expression to specify what to store.

Redis Caching

Redis caching stores API responses so the workflow can reuse them on subsequent runs without making another HTTP call. This is especially important for OAuth tokens, which are expensive to request but can be reused for the duration of their validity period.

Quick Setup

nodes:
- name: cached_workflow
  type: http_workflow_input
  workflow_pull_interval: 5m

  # Global Redis connection
  redis_address: "redis://localhost:6379"
  redis_password: "{{ SECRET redis_password }}"  # Optional
  redis_tls: false

  steps:
    - name: cached_request
      endpoint: https://api.example.com/data
      redis_cache:
        key_expression: '"my-cache-key"'      # Cache key (must evaluate to string)
        ttl: 10m                               # Cache duration
        check_before_request: true             # Check cache first
        populate_on_success: true              # Store response on success

Cache Configuration Reference

Complete Reference

Node-Level Parameters

Step-Level Parameters

Webhook Configuration

Iterator Configuration

Security

Built-in SSRF Protection

HTTP Workflow blocks requests to:

• Localhost (127.0.0.1, ::1, localhost)
• Private IPs (10.x, 172.16-31.x, 192.168.x)
• Cloud metadata (169.254.169.254)
• Non-HTTP schemes (file://, ftp://)

Credential Best Practices

# DO: Use Edge Delta secrets (recommended)
headers:
  Authorization: "Bearer {{ SECRET api_token }}"

# DO: Use EDXEnv for environment variables
header_expressions:
  Authorization: Concat(["Bearer ", EDXEnv("API_TOKEN", "")], "")

# DON'T: Hardcode secrets
headers:
  Authorization: Bearer sk_live_abc123  # Never do this!

Note: Create secrets in the Edge Delta app: Settings > Secrets. Use lowercase names with underscores (e.g., api_token, oauth_client_secret).

Troubleshooting

Debugging Workflow Execution

By default, only the final step’s response is emitted as a log item. If your workflow isn’t producing the expected output, these techniques help you see what each step is doing.

1. Enable Emit Intermediate Responses

Set emit_intermediate: true to emit the response from every step, not just the last one. This makes it easier to identify which step is failing or returning unexpected data:

nodes:
- name: my_workflow
  type: http_workflow_input
  emit_intermediate: true    # Emit data from all steps
  steps:
    - name: step_one
      endpoint: https://api.example.com/auth
    - name: step_two
      endpoint: https://api.example.com/data

In the UI, toggle Emit Intermediate Responses under Advanced Settings.

2. Add Debug Attributes

Use emit_transform to attach the full workflow state to each emitted log. This adds all step responses and variable values as attributes you can inspect in the Edge Delta app:

nodes:
- name: my_workflow
  type: http_workflow_input
  emit_transform:
    - set(attributes["debug_steps"], steps)
    - set(attributes["debug_vars"], variables)
  steps:
    # ...

This adds debug_steps and debug_vars attributes to your output, showing the complete state of all steps and variables at emit time.

No Data Retrieved

1. Verify endpoint is accessible: Test with curl first
2. Check authentication: Ensure secrets/env vars are set correctly
3. Review step names: Names are case-sensitive and must use snake_case
4. Enable emit_intermediate: See if earlier steps are succeeding
5. Enable debug logging:

   log:
     level: debug
   

Token Not Being Cached

1. Verify Redis is reachable: redis-cli ping
2. Check check_before_request: true is set
3. Ensure key_expression evaluates to a string
4. Confirm TTL is less than token expiry

Iterator Not Working

1. Check variable name: Ensure the variable used in expressions matches variable_name exactly
2. Verify source returns an array: Use ParseJSON() if needed, and ensure it returns an array not an object
3. Check previous step: The step providing data must complete successfully
4. Enable emit_intermediate: Verify the list step is returning expected data
5. Try continue_on_error: true for partial success

Webhook Not Receiving Requests

1. Check port availability: lsof -i:8888
2. Verify firewall allows incoming connections
3. Confirm auth headers match configuration
4. Check agent logs for 401 or 429 responses

Testing Your Workflow

Before deploying a workflow, verify that each endpoint is reachable and returns the expected response by testing with curl:

# Test OAuth endpoint
curl -X POST "https://auth.example.com/oauth/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials&client_id=<YOUR_CLIENT_ID>&client_secret=<YOUR_CLIENT_SECRET>"

# Test data endpoint with token
curl "https://api.example.com/data" \
  -H "Authorization: Bearer <YOUR_ACCESS_TOKEN>"

# Test webhook locally
curl -X POST "http://localhost:8888/webhooks/test" \
  -H "Authorization: Bearer <YOUR_WEBHOOK_SECRET>" \
  -H "Content-Type: application/json" \
  -d '{"event": "test", "user_id": "123"}'

Configuration Wizard

Use this interactive wizard to generate a starter configuration:

See Also

• HTTP Pull Source — Single-request HTTP polling
• OTTL Language Guide — Complete expression reference