Custom Component Development Guide¶

Custom components are the building blocks that agents, tools, and pipelines connect to. They are registered via hc.AddComponent() and used in triggers, tool connectors, and pipeline steps.

This guide walks through building custom Helix components from scratch. Components are the primary extension mechanism in Helix -- they expose URI-addressable endpoints that can produce and consume messages within workflows.

Table of Contents¶

Overview
Core Interfaces
Component
Endpoint
Producer
Consumer
PollingConsumer
Component Lifecycle
URI Parsing
Building a Weather Component
Project Structure
Component Implementation
Producer Implementation (Outbound)
Consumer Implementation (Inbound Polling)
Registration
Wiring into main.go
Testing
Best Practices

Overview¶

A Helix component is a factory for endpoints. Each endpoint represents a communication channel identified by a URI. Endpoints create producers (send messages out) and consumers (receive messages in).

The URI scheme determines which component handles a given endpoint:

weather://current?city=London&units=metric
^^^^^^^   ^^^^^^^  ^^^^^^^^^^^^^^^^^^^^^^^^^^^
scheme    path     query parameters

When a workflow references weather://current?city=London, the engine looks up the component registered under the weather scheme, calls CreateEndpoint, and uses the resulting endpoint to create a producer or consumer depending on the workflow direction.

Core Interfaces¶

All interfaces are defined in pkg/engine/interfaces.go.

Component¶

type Component interface {
    // Scheme returns the URI scheme this component handles (e.g. "weather").
    Scheme() string

    // CreateEndpoint builds an Endpoint from the parsed URI.
    // - uri: the full original URI string
    // - remaining: the path portion after the scheme (e.g. "current")
    // - params: query parameters as a string map
    CreateEndpoint(uri string, remaining string, params map[string]string) (Endpoint, error)

    // Start is called once when the Helix engine starts.
    Start(ctx context.Context) error

    // Stop is called once during graceful shutdown.
    Stop(ctx context.Context) error
}

Endpoint¶

type Endpoint interface {
    // URI returns the full URI string for this endpoint.
    URI() string

    // Component returns the parent component that created this endpoint.
    Component() Component

    // CreateProducer creates a Producer for sending messages TO this endpoint.
    CreateProducer() (Producer, error)

    // CreateConsumer creates a Consumer that receives messages FROM this endpoint
    // and dispatches them to the given processor (the downstream workflow pipeline).
    CreateConsumer(processor Processor) (Consumer, error)
}

Producer¶

type Producer interface {
    // Process sends a message through this producer (inherited from Processor).
    Process(ctx context.Context, exchange *exchange.Exchange) error

    // Start initializes the producer (open connections, etc.).
    Start(ctx context.Context) error

    // Stop tears down the producer.
    Stop(ctx context.Context) error

    // Endpoint returns the parent endpoint.
    Endpoint() Endpoint
}

Note that Producer embeds the Processor interface. This means a producer IS a processor -- it can be placed directly in a processing pipeline.

Consumer¶

type Consumer interface {
    // Start begins consuming messages (may spawn goroutines).
    Start(ctx context.Context) error

    // Stop halts consumption and cleans up.
    Stop(ctx context.Context) error

    // Endpoint returns the parent endpoint.
    Endpoint() Endpoint
}

PollingConsumer¶

type PollingConsumer interface {
    Consumer

    // Poll actively checks for a new message. Returns nil if none available.
    Poll(ctx context.Context) (*exchange.Exchange, error)
}

Use PollingConsumer when the external system requires active polling (e.g., checking an API on an interval) rather than push-based delivery.

Component Lifecycle¶

Engine.Start()
  |
  +-- Component.Start(ctx)        -- called once per component
  |
  +-- For each workflow:
  |     +-- CreateEndpoint(...)    -- parse URI, build endpoint
  |     +-- CreateConsumer(proc)   -- if this is a "from" URI
  |     +-- CreateProducer()       -- if this is a "to" URI
  |     +-- Consumer.Start(ctx)    -- begin receiving
  |     +-- Producer.Start(ctx)    -- open connections
  |
  ...running...
  |
Engine.Stop()
  |
  +-- For each workflow (reverse order):
  |     +-- Consumer.Stop(ctx)
  |     +-- Producer.Stop(ctx)
  |
  +-- Component.Stop(ctx)          -- called once per component

Key points:

Component.Start/Stop are called exactly once, regardless of how many endpoints use the component.
Producer.Start/Stop and Consumer.Start/Stop are called per-endpoint.
Shutdown proceeds in reverse order to ensure consumers stop receiving before producers close their connections.

URI Parsing¶

The engine parses URIs before calling CreateEndpoint. The component receives three pieces:

Parameter	Example URI: `weather://current/forecast?city=London&units=metric`
`uri`	`weather://current/forecast?city=London&units=metric`
`remaining`	`current/forecast`
`params`	`{"city": "London", "units": "metric"}`

The remaining string is the path after scheme://. Use it to determine the operation or resource. The params map contains query parameters.

Building a Weather Component¶

This complete example builds a weather:// component that:

Producer: fetches current weather from an HTTP API when a message is sent to it.
Consumer: polls the weather API on a configurable interval and feeds updates into a workflow.

Project Structure¶

pkg/component/weather/
    weather.go       -- Component + Endpoint
    producer.go      -- Producer (outbound)
    consumer.go      -- Consumer (inbound polling)
    weather_test.go  -- Tests

Component Implementation¶

// pkg/component/weather/weather.go
package weather

import (
    "context"
    "fmt"
    "net/http"
    "sync"

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

// Component implements engine.Component for the "weather" scheme.
type Component struct {
    client     *http.Client
    baseURL    string
    initOnce   sync.Once
}

// New creates a new weather component.
func New() *Component {
    return &Component{
        baseURL: "https://api.weatherapi.com/v1",
    }
}

func (c *Component) Scheme() string { return "weather" }

func (c *Component) CreateEndpoint(uri, remaining string, params map[string]string) (engine.Endpoint, error) {
    operation := remaining
    if operation == "" {
        operation = "current"
    }

    // Validate operation
    switch operation {
    case "current", "forecast":
        // ok
    default:
        return nil, fmt.Errorf("weather: unknown operation %q (use current or forecast)", operation)
    }

    return &Endpoint{
        uri:       uri,
        component: c,
        operation: operation,
        params:    params,
    }, nil
}

func (c *Component) Start(ctx context.Context) error {
    c.initOnce.Do(func() {
        c.client = &http.Client{}
    })
    return nil
}

func (c *Component) Stop(ctx context.Context) error {
    if c.client != nil {
        c.client.CloseIdleConnections()
    }
    return nil
}

// Endpoint implements engine.Endpoint.
type Endpoint struct {
    uri       string
    component *Component
    operation string
    params    map[string]string
}

func (e *Endpoint) URI() string              { return e.uri }
func (e *Endpoint) Component() engine.Component { return e.component }

func (e *Endpoint) CreateProducer() (engine.Producer, error) {
    apiKey := e.params["apiKey"]
    if apiKey == "" {
        return nil, fmt.Errorf("weather: apiKey parameter is required")
    }
    return &Producer{
        endpoint:  e,
        client:    e.component.client,
        baseURL:   e.component.baseURL,
        apiKey:    apiKey,
        operation: e.operation,
        city:      e.params["city"],
    }, nil
}

func (e *Endpoint) CreateConsumer(processor engine.Processor) (engine.Consumer, error) {
    apiKey := e.params["apiKey"]
    if apiKey == "" {
        return nil, fmt.Errorf("weather: apiKey parameter is required")
    }
    return &Consumer{
        endpoint:  e,
        processor: processor,
        client:    e.component.client,
        baseURL:   e.component.baseURL,
        apiKey:    apiKey,
        operation: e.operation,
        city:      e.params["city"],
        interval:  e.params["interval"],
    }, nil
}

Producer Implementation (Outbound)¶

The producer is invoked when a workflow sends a message TO the weather endpoint (- to: weather://current?city=London&apiKey=...).

// pkg/component/weather/producer.go
package weather

import (
    "context"
    "encoding/json"
    "fmt"
    "io"
    "net/http"

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

type Producer struct {
    endpoint  *Endpoint
    client    *http.Client
    baseURL   string
    apiKey    string
    operation string
    city      string
}

func (p *Producer) Endpoint() engine.Endpoint { return p.endpoint }

func (p *Producer) Start(ctx context.Context) error { return nil }
func (p *Producer) Stop(ctx context.Context) error  { return nil }

func (p *Producer) Process(ctx context.Context, ex *exchange.Exchange) error {
    // Allow the city to be overridden by exchange header or body.
    city := p.city
    if c := ex.In().HeaderString("city"); c != "" {
        city = c
    }
    if city == "" {
        if s, ok := ex.In().Body().(string); ok && s != "" {
            city = s
        }
    }
    if city == "" {
        return fmt.Errorf("weather: city is required (set via URI param, header, or body)")
    }

    url := fmt.Sprintf("%s/%s.json?key=%s&q=%s", p.baseURL, p.operation, p.apiKey, city)

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return fmt.Errorf("weather: failed to create request: %w", err)
    }

    resp, err := p.client.Do(req)
    if err != nil {
        return fmt.Errorf("weather: request failed: %w", err)
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return fmt.Errorf("weather: failed to read response: %w", err)
    }

    if resp.StatusCode != http.StatusOK {
        ex.SetProperty("weather.error", true)
        ex.In().SetBody(string(body))
        return fmt.Errorf("weather: API returned %d: %s", resp.StatusCode, string(body))
    }

    // Parse JSON and set as exchange body.
    var result map[string]any
    if err := json.Unmarshal(body, &result); err != nil {
        ex.In().SetBody(string(body))
        return nil
    }

    ex.In().SetBody(result)
    ex.In().SetHeader("weather.city", city)
    ex.In().SetHeader("weather.operation", p.operation)
    return nil
}

Consumer Implementation (Inbound Polling)¶

The consumer is used when a workflow starts FROM the weather endpoint (from: weather://current?city=London&interval=60s&apiKey=...). It polls the API on an interval.

// pkg/component/weather/consumer.go
package weather

import (
    "context"
    "encoding/json"
    "fmt"
    "io"
    "log/slog"
    "net/http"
    "time"

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

type Consumer struct {
    endpoint  *Endpoint
    processor engine.Processor
    client    *http.Client
    baseURL   string
    apiKey    string
    operation string
    city      string
    interval  string
    cancel    context.CancelFunc
}

func (c *Consumer) Endpoint() engine.Endpoint { return c.endpoint }

func (c *Consumer) Start(ctx context.Context) error {
    interval := 60 * time.Second
    if c.interval != "" {
        d, err := time.ParseDuration(c.interval)
        if err != nil {
            return fmt.Errorf("weather: invalid interval %q: %w", c.interval, err)
        }
        interval = d
    }

    ctx, c.cancel = context.WithCancel(ctx)

    go func() {
        ticker := time.NewTicker(interval)
        defer ticker.Stop()

        // Poll immediately on start, then on each tick.
        c.poll(ctx)

        for {
            select {
            case <-ctx.Done():
                return
            case <-ticker.C:
                c.poll(ctx)
            }
        }
    }()

    return nil
}

func (c *Consumer) Stop(ctx context.Context) error {
    if c.cancel != nil {
        c.cancel()
    }
    return nil
}

func (c *Consumer) poll(ctx context.Context) {
    url := fmt.Sprintf("%s/%s.json?key=%s&q=%s", c.baseURL, c.operation, c.apiKey, c.city)

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        slog.Error("weather: failed to create poll request", "error", err)
        return
    }

    resp, err := c.client.Do(req)
    if err != nil {
        slog.Error("weather: poll request failed", "error", err)
        return
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        slog.Error("weather: failed to read poll response", "error", err)
        return
    }

    var result map[string]any
    if err := json.Unmarshal(body, &result); err != nil {
        slog.Error("weather: failed to parse poll response", "error", err)
        return
    }

    ex := exchange.New(context.Background())
    ex.In().SetBody(result)
    ex.In().SetHeader("weather.city", c.city)
    ex.In().SetHeader("weather.operation", c.operation)

    if err := c.processor.Process(ctx, ex); err != nil {
        slog.Error("weather: processor failed", "error", err)
    }
}

Registration¶

Register the component in the engine so that workflows can reference weather:// URIs.

There are two approaches:

Option A: Register in main.go (recommended for project-specific components)¶

// In cmd/helix/main.go, inside registerDefaultComponents():
h.AddComponent("weather", weather.New())

Option B: Register via init() (recommended for reusable library components)¶

// pkg/component/weather/register.go
package weather

import "github.com/arvenza/helix/pkg/engine"

// Register the component with a HelixContext:
//   h.AddComponent("weather", weather.New())

Then import the package for side effects:

import _ "github.com/arvenza/helix/pkg/component/weather"

Wiring into main.go¶

Add the import and registration call in cmd/helix/main.go:

import (
    // ... existing imports ...
    "github.com/arvenza/helix/pkg/component/weather"
)

func registerDefaultComponents(h *engine.Helix) {
    // ... existing registrations ...

    // Custom components
    h.AddComponent("weather", weather.New())
}

Now workflows can use the component:

workflows:
  # Outbound: fetch weather on demand
  - id: get-weather
    trigger:
      uri: http-server://0.0.0.0:8080/api/weather
    steps:
      - target: weather://current?city=London&apiKey=${env:WEATHER_API_KEY}
      - target: log://output?showBody=true

  # Inbound: poll weather every 5 minutes
  - id: weather-monitor
    trigger:
      uri: weather://current?city=Tokyo&interval=5m&apiKey=${env:WEATHER_API_KEY}
    steps:
      - transform:
          starlang: |
            data = read(payload, "application/json")
            payload = {"temp": data["current"]["temp_c"], "city": "Tokyo"}
      - target: kafka://weather-updates?brokers=localhost:9092

Testing¶

Use net/http/httptest to mock the external API and test your component in isolation.

// pkg/component/weather/weather_test.go
package weather

import (
    "context"
    "encoding/json"
    "net/http"
    "net/http/httptest"
    "testing"

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

func TestProducer_FetchWeather(t *testing.T) {
    // Set up a mock weather API server.
    server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path != "/current.json" {
            t.Errorf("unexpected path: %s", r.URL.Path)
        }
        city := r.URL.Query().Get("q")
        if city != "London" {
            t.Errorf("unexpected city: %s", city)
        }
        json.NewEncoder(w).Encode(map[string]any{
            "current": map[string]any{
                "temp_c":    15.0,
                "condition": map[string]any{"text": "Partly cloudy"},
            },
        })
    }))
    defer server.Close()

    // Create the component with the mock URL.
    comp := New()
    comp.baseURL = server.URL
    comp.client = server.Client()

    // Create endpoint and producer.
    ep, err := comp.CreateEndpoint(
        "weather://current?city=London&apiKey=test-key",
        "current",
        map[string]string{"city": "London", "apiKey": "test-key"},
    )
    if err != nil {
        t.Fatal(err)
    }

    prod, err := ep.CreateProducer()
    if err != nil {
        t.Fatal(err)
    }

    // Process an exchange.
    ctx := context.Background()
    ex := exchange.New(context.Background())

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

    // Verify the result.
    result, ok := ex.In().Body().(map[string]any)
    if !ok {
        t.Fatalf("expected map body, got %T", ex.In().Body())
    }

    current := result["current"].(map[string]any)
    if current["temp_c"] != 15.0 {
        t.Errorf("expected temp 15.0, got %v", current["temp_c"])
    }

    if ex.In().Header("weather.city") != "London" {
        t.Errorf("expected city header London, got %s", ex.In().Header("weather.city"))
    }
}

func TestProducer_MissingCity(t *testing.T) {
    comp := New()
    comp.client = &http.Client{}

    ep, err := comp.CreateEndpoint(
        "weather://current?apiKey=test-key",
        "current",
        map[string]string{"apiKey": "test-key"},
    )
    if err != nil {
        t.Fatal(err)
    }

    prod, err := ep.CreateProducer()
    if err != nil {
        t.Fatal(err)
    }

    ex := exchange.New(context.Background())
    err = prod.Process(context.Background(), ex)
    if err == nil {
        t.Fatal("expected error for missing city")
    }
}

func TestCreateEndpoint_InvalidOperation(t *testing.T) {
    comp := New()
    _, err := comp.CreateEndpoint(
        "weather://invalid",
        "invalid",
        map[string]string{"apiKey": "test"},
    )
    if err == nil {
        t.Fatal("expected error for invalid operation")
    }
}

Run tests:

go test ./pkg/component/weather/ -v

Best Practices¶

Use sync.Once for lazy initialization¶

Defer expensive setup (HTTP clients, connection pools) until Start() or first use:

func (c *Component) Start(ctx context.Context) error {
    c.initOnce.Do(func() {
        c.client = &http.Client{Timeout: 30 * time.Second}
    })
    return nil
}

Propagate context everywhere¶

Always use the provided context.Context for HTTP requests, database queries, and any blocking operation. This enables proper timeout and cancellation handling:

req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

Set meaningful headers on the exchange¶

Headers provide metadata that downstream processors can inspect without parsing the body:

ex.In().SetHeader("weather.city", city)
ex.In().SetHeader("weather.status_code", fmt.Sprintf("%d", resp.StatusCode))

Return structured errors with component prefix¶

Prefix all errors with the component name for clear diagnostics:

return fmt.Errorf("weather: API returned %d: %s", resp.StatusCode, body)

Handle both InOnly and InOut exchange patterns¶

Check the exchange pattern to determine whether a response is expected:

if ex.Pattern() == exchange.InOut {
    // Set the Out message for request-reply patterns
    ex.Out().SetBody(result)
} else {
    // Modify the In message for one-way patterns
    ex.In().SetBody(result)
}

Clean up resources in Stop()¶

Close connections, cancel goroutines, and flush buffers:

func (c *Component) Stop(ctx context.Context) error {
    if c.client != nil {
        c.client.CloseIdleConnections()
    }
    return nil
}

Validate parameters early¶

Validate required parameters in CreateEndpoint or CreateProducer/CreateConsumer, not at runtime:

func (e *Endpoint) CreateProducer() (engine.Producer, error) {
    if e.params["apiKey"] == "" {
        return nil, fmt.Errorf("weather: apiKey parameter is required")
    }
    // ...
}

Make parameters configurable via URI¶

Follow the Helix convention of URI query parameters for configuration:

weather://current?city=London&apiKey=xxx&timeout=10s&retries=3

This allows all configuration to live in the YAML flow file alongside the workflow definition.