Custom Processor Development Guide¶

Custom processors define the logic steps within agent pipelines, tool chains, and data flows.

Processors are the building blocks of Helix workflow pipelines. Every step in a workflow -- transforms, filters, routers, enrichers -- is a processor. This guide covers building custom processors, integrating them with the YAML DSL, and testing them.

Table of Contents¶

Processor Interface
ProcessorFunc Shorthand
Exchange Anatomy
Body
Headers
Properties
Simple Example: Uppercase Processor
Complex Example: JSON Validation Processor
Error Handling Patterns
Adding YAML DSL Support
Testing Processors
Built-in Processor Catalog

Processor Interface¶

The processor interface is defined in pkg/engine/interfaces.go:

type Processor interface {
    Process(ctx context.Context, exchange *exchange.Exchange) error
}

That is the entire contract. A processor receives a context and an exchange, does its work, and returns nil on success or an error on failure.

The engine also provides a function adapter:

type ProcessorFunc func(ctx context.Context, exchange *exchange.Exchange) error

func (f ProcessorFunc) Process(ctx context.Context, exchange *exchange.Exchange) error {
    return f(ctx, exchange)
}

ProcessorFunc Shorthand¶

For simple processors that do not need struct state, use ProcessorFunc:

var toUpperCase engine.Processor = engine.ProcessorFunc(
    func(ctx context.Context, ex *exchange.Exchange) error {
        if s, ok := ex.In().Body().(string); ok {
            ex.In().SetBody(strings.ToUpper(s))
        }
        return nil
    },
)

This is useful for inline processors in tests or one-off workflow steps.

Exchange Anatomy¶

The exchange.Exchange (defined in pkg/exchange/) is the data envelope that flows through every workflow. It carries:

Body¶

The primary payload. Can be any Go value (any / interface{}).

// Read the body
body := ex.In().Body()

// Set the body
ex.In().SetBody(map[string]any{"status": "ok"})

// Type-assert as needed
switch v := body.(type) {
case string:
    // string body
case []byte:
    // binary body
case map[string]any:
    // JSON-like map
default:
    // use fmt.Sprintf("%v", v) as fallback
}

Headers¶

String-keyed metadata attached to the message. Headers are per-message and travel with the exchange through the workflow.

// Read a header
contentType := ex.In().Header("Content-Type")

// Set a header
ex.In().SetHeader("X-Processed-By", "my-processor")

// Check existence (empty string means absent)
if ex.In().Header("Authorization") == "" {
    return fmt.Errorf("missing Authorization header")
}

Properties¶

Exchange-level key-value pairs that are NOT part of the message. Properties are used for routing decisions, processor state, and metadata that should not be forwarded to external systems.

// Set a property
ex.SetProperty("validation.passed", true)

// Read a property
if passed, ok := ex.Property("validation.passed").(bool); ok && passed {
    // ...
}

Headers vs. Properties:

Aspect	Headers	Properties
Scope	Per-message	Per-exchange
Forwarded to external systems	Yes	No
Use for	Content-Type, auth tokens, routing keys	Internal flags, processor state

Simple Example: Uppercase Processor¶

A processor that converts string bodies to uppercase.

// pkg/processor/uppercase.go
package processor

import (
    "context"
    "fmt"
    "strings"

    "github.com/arvenza/helix/pkg/exchange"
)

// UppercaseProcessor converts the exchange body to uppercase.
type UppercaseProcessor struct{}

func NewUppercaseProcessor() *UppercaseProcessor {
    return &UppercaseProcessor{}
}

func (p *UppercaseProcessor) Process(ctx context.Context, ex *exchange.Exchange) error {
    body := ex.In().Body()
    switch v := body.(type) {
    case string:
        ex.In().SetBody(strings.ToUpper(v))
    case []byte:
        ex.In().SetBody([]byte(strings.ToUpper(string(v))))
    default:
        return fmt.Errorf("uppercase: unsupported body type %T, expected string or []byte", body)
    }
    return nil
}

Complex Example: JSON Validation Processor¶

A processor that validates the exchange body against a set of required fields and type constraints.

// pkg/processor/json_validator.go
package processor

import (
    "context"
    "encoding/json"
    "fmt"
    "strings"

    "github.com/arvenza/helix/pkg/exchange"
)

// FieldRule defines a validation rule for a single JSON field.
type FieldRule struct {
    Name     string // JSON field name (dot-separated for nested: "address.city")
    Required bool   // Whether the field must be present
    Type     string // Expected type: "string", "number", "bool", "array", "object"
}

// JSONValidatorProcessor validates exchange bodies against field rules.
type JSONValidatorProcessor struct {
    Rules         []FieldRule
    RejectOnError bool // If true, set an error on the exchange instead of returning an error
}

func NewJSONValidatorProcessor(rules []FieldRule) *JSONValidatorProcessor {
    return &JSONValidatorProcessor{
        Rules:         rules,
        RejectOnError: false,
    }
}

func (p *JSONValidatorProcessor) Process(ctx context.Context, ex *exchange.Exchange) error {
    // Parse body to map.
    var data map[string]any
    switch v := ex.In().Body().(type) {
    case map[string]any:
        data = v
    case string:
        if err := json.Unmarshal([]byte(v), &data); err != nil {
            return p.handleError(ex, fmt.Errorf("json-validator: body is not valid JSON: %w", err))
        }
    case []byte:
        if err := json.Unmarshal(v, &data); err != nil {
            return p.handleError(ex, fmt.Errorf("json-validator: body is not valid JSON: %w", err))
        }
    default:
        return p.handleError(ex, fmt.Errorf("json-validator: unsupported body type %T", ex.In().Body()))
    }

    // Validate each rule.
    var errors []string
    for _, rule := range p.Rules {
        value, found := resolveField(data, rule.Name)

        if rule.Required && !found {
            errors = append(errors, fmt.Sprintf("missing required field %q", rule.Name))
            continue
        }

        if found && rule.Type != "" {
            if err := checkType(value, rule.Type); err != nil {
                errors = append(errors, fmt.Sprintf("field %q: %s", rule.Name, err))
            }
        }
    }

    if len(errors) > 0 {
        return p.handleError(ex, fmt.Errorf("json-validator: %s", strings.Join(errors, "; ")))
    }

    // Mark as validated.
    ex.SetProperty("json.validated", true)
    ex.In().SetHeader("X-Validated", "true")
    return nil
}

func (p *JSONValidatorProcessor) handleError(ex *exchange.Exchange, err error) error {
    if p.RejectOnError {
        ex.SetProperty("validation.error", err.Error())
        ex.In().SetHeader("X-Validated", "false")
        return nil // Don't stop the pipeline
    }
    return err // Stop the pipeline
}

// resolveField traverses a map using dot-separated keys.
func resolveField(data map[string]any, path string) (any, bool) {
    parts := strings.Split(path, ".")
    current := any(data)
    for _, part := range parts {
        m, ok := current.(map[string]any)
        if !ok {
            return nil, false
        }
        current, ok = m[part]
        if !ok {
            return nil, false
        }
    }
    return current, true
}

func checkType(value any, expected string) error {
    switch expected {
    case "string":
        if _, ok := value.(string); !ok {
            return fmt.Errorf("expected string, got %T", value)
        }
    case "number":
        switch value.(type) {
        case float64, int, int64, float32:
            // ok
        default:
            return fmt.Errorf("expected number, got %T", value)
        }
    case "bool":
        if _, ok := value.(bool); !ok {
            return fmt.Errorf("expected bool, got %T", value)
        }
    case "array":
        if _, ok := value.([]any); !ok {
            return fmt.Errorf("expected array, got %T", value)
        }
    case "object":
        if _, ok := value.(map[string]any); !ok {
            return fmt.Errorf("expected object, got %T", value)
        }
    }
    return nil
}

Error Handling Patterns¶

Stop the pipeline¶

Return an error to halt processing. The exchange is marked as failed and the error handler (if configured) takes over:

func (p *MyProcessor) Process(ctx context.Context, ex *exchange.Exchange) error {
    return fmt.Errorf("my-processor: something went wrong")
}

Set error without stopping¶

Store the error as a property and let downstream processors decide:

ex.SetProperty("my-processor.error", "invalid input")
ex.In().SetHeader("X-Error", "true")
return nil // Pipeline continues

Wrap errors with context¶

Always include the processor name and relevant details:

return fmt.Errorf("json-validator: field %q: expected %s, got %T", field, expectedType, value)

Respect context cancellation¶

Check for context cancellation in long-running processors:

func (p *SlowProcessor) Process(ctx context.Context, ex *exchange.Exchange) error {
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
    }
    // ... do work ...
    return nil
}

Adding YAML DSL Support¶

To make your processor available as a YAML step, register a reifier function in pkg/dsl/reifier.go.

Step 1: Define the YAML step structure¶

In pkg/workflow/definition.go, add a new field to the StepDefinition struct (or use the existing generic mechanism):

type StepDefinition struct {
    // ... existing fields ...
    JSONValidate *JSONValidateStep `yaml:"json-validate,omitempty" json:"json-validate,omitempty"`
}

type JSONValidateStep struct {
    Fields        []FieldRuleDef `yaml:"fields" json:"fields"`
    RejectOnError bool           `yaml:"reject_on_error,omitempty" json:"reject_on_error,omitempty"`
}

type FieldRuleDef struct {
    Name     string `yaml:"name" json:"name"`
    Required bool   `yaml:"required,omitempty" json:"required,omitempty"`
    Type     string `yaml:"type,omitempty" json:"type,omitempty"`
}

Step 2: Register the reifier¶

In pkg/dsl/reifier.go, add handling for the new step in the reifyStep method:

case step.JSONValidate != nil:
    rules := make([]processor.FieldRule, len(step.JSONValidate.Fields))
    for i, f := range step.JSONValidate.Fields {
        rules[i] = processor.FieldRule{
            Name:     f.Name,
            Required: f.Required,
            Type:     f.Type,
        }
    }
    p := processor.NewJSONValidatorProcessor(rules)
    p.RejectOnError = step.JSONValidate.RejectOnError
    processors = append(processors, p)

Step 3: Use in YAML¶

workflows:
  - id: validate-orders
    trigger:
      uri: kafka://orders?brokers=localhost:9092&group=validator
    steps:
      - json-validate:
          fields:
            - name: order_id
              required: true
              type: string
            - name: amount
              required: true
              type: number
            - name: customer.email
              required: true
              type: string
          reject_on_error: false
      - choice:
          when:
            - starlang: 'properties["validation.error"] != None'
              steps:
                - target: kafka://invalid-orders?brokers=localhost:9092
          otherwise:
            steps:
              - target: kafka://valid-orders?brokers=localhost:9092

Testing Processors¶

Unit testing¶

Create an exchange, run the processor, and assert on the result:

package processor

import (
    "context"
    "testing"

    "github.com/arvenza/helix/pkg/exchange"
)

func TestUppercaseProcessor(t *testing.T) {
    p := NewUppercaseProcessor()
    ctx := context.Background()

    t.Run("converts string body", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody("hello world")

        if err := p.Process(ctx, ex); err != nil {
            t.Fatal(err)
        }

        if ex.In().Body() != "HELLO WORLD" {
            t.Errorf("expected HELLO WORLD, got %v", ex.In().Body())
        }
    })

    t.Run("converts byte body", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody([]byte("hello"))

        if err := p.Process(ctx, ex); err != nil {
            t.Fatal(err)
        }

        got := string(ex.In().Body().([]byte))
        if got != "HELLO" {
            t.Errorf("expected HELLO, got %s", got)
        }
    })

    t.Run("rejects unsupported type", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody(42)

        err := p.Process(ctx, ex)
        if err == nil {
            t.Fatal("expected error for int body")
        }
    })
}

func TestJSONValidatorProcessor(t *testing.T) {
    rules := []FieldRule{
        {Name: "name", Required: true, Type: "string"},
        {Name: "age", Required: true, Type: "number"},
        {Name: "address.city", Required: false, Type: "string"},
    }

    p := NewJSONValidatorProcessor(rules)
    ctx := context.Background()

    t.Run("valid input", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody(map[string]any{
            "name": "Alice",
            "age":  30.0,
            "address": map[string]any{
                "city": "London",
            },
        })

        if err := p.Process(ctx, ex); err != nil {
            t.Fatal(err)
        }

        if ex.In().Header("X-Validated") != "true" {
            t.Error("expected X-Validated header to be true")
        }
    })

    t.Run("missing required field", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody(map[string]any{
            "name": "Alice",
            // missing "age"
        })

        err := p.Process(ctx, ex)
        if err == nil {
            t.Fatal("expected error for missing required field")
        }
    })

    t.Run("wrong type", func(t *testing.T) {
        ex := exchange.New(ctx)
        ex.In().SetBody(map[string]any{
            "name": 123,    // should be string
            "age":  30.0,
        })

        err := p.Process(ctx, ex)
        if err == nil {
            t.Fatal("expected error for wrong field type")
        }
    })

    t.Run("reject mode sets property", func(t *testing.T) {
        p2 := NewJSONValidatorProcessor(rules)
        p2.RejectOnError = true

        ex := exchange.New(ctx)
        ex.In().SetBody(map[string]any{"name": 123})

        // Should NOT return an error in reject mode.
        if err := p2.Process(ctx, ex); err != nil {
            t.Fatalf("reject mode should not return error, got: %v", err)
        }

        if ex.Property("validation.error") == nil {
            t.Error("expected validation.error property to be set")
        }
    })
}

Integration testing with the test framework¶

Use the Helix declarative test format:

workflows:
  - id: validate-and-workflow
    trigger:
      uri: direct://input
    steps:
      - json-validate:
          fields:
            - name: id
              required: true
              type: string
      - target: mock://output

tests:
  - id: valid-json-passes
    description: "Valid JSON body passes validation"
    workflow: validate-and-workflow
    input:
      body: '{"id": "abc-123"}'
    expect:
      body_contains: "abc-123"
      headers:
        X-Validated: "true"

  - id: invalid-json-fails
    description: "Missing required field fails validation"
    workflow: validate-and-workflow
    input:
      body: '{"other": "value"}'
    expect:
      error_contains: "missing required field"

Run with:

helix test validate_test.yaml

Built-in Processor Catalog¶

Helix ships with 60+ processors across 14 source files in pkg/processor/. Here is an overview by category:

Message Transformation¶

Processor	Description
`transform`	Set body from constant, Starlang expression, or header
`set-body`	Replace body with a value
`set-header`	Set a message header
`remove-header`	Remove a single header
`remove-headers`	Remove headers matching a pattern
`convert-body-to`	Convert body to string, bytes, int, float, map, etc.
`marshal` / `unmarshal`	Serialize/deserialize with JSON, XML, CSV, YAML

Routing¶

Processor	Description
`choice`	Content-based router (when/otherwise)
`filter`	Drop messages that don't match a predicate
`routing-slip`	Dynamic routing based on header
`dynamic-router`	Iterative routing with feedback
`recipient-list`	Send to multiple dynamic recipients
`load-balancer`	Round-robin, random, weighted, sticky, failover

Splitting & Aggregation¶

Processor	Description
`split`	Split a message into parts
`aggregate`	Combine related messages
`multicast`	Send to multiple static endpoints in parallel or sequentially
`scatter-gather`	Multicast + aggregate pattern
`composed-message`	Split, process, re-aggregate

Error Handling¶

Processor	Description
`on-exception`	Per-exception-type handling
`error-handler`	Default, dead-letter-channel, no-error-handler
`try-catch-finally`	Structured error handling
`do-catch`	Catch specific exception types
`do-finally`	Always-execute block

Resilience¶

Processor	Description
`circuit-breaker`	Fail fast when downstream is down
`retry`	Retry with configurable backoff
`timeout`	Time-limit processing
`throttle`	Rate-limit messages
`idempotent-consumer`	Deduplicate by message ID

Advanced Orchestration Patterns¶

Processor	Description
`saga`	Distributed transaction with compensation
`resequencer`	Reorder messages by sequence number
`claim-check`	Store/retrieve large payloads
`normalize`	Convert different message formats to a common format
`enrich`	Enrich message with data from another endpoint
`poll-enrich`	Enrich by polling an endpoint
`wire-tap`	Copy message to a secondary endpoint
`log`	Log message to slog
`delay`	Delay message delivery
`step`	Group processors into a named unit
`pipeline`	Chain processors sequentially
`on-completion`	Execute after exchange completes (success or failure)