Connector SDK Development Guide¶

The Helix Connector SDK provides a framework for integrating external systems -- APIs, databases, message brokers, cloud services -- into Helix workflows. Connectors are higher-level than components: they handle connection lifecycle, health checks, reconnection, and state management out of the box.

Helix ships with 441 connectors across 33 categories. This guide covers building custom connectors from scratch.

Table of Contents¶

Architecture Overview
Core Interfaces
Connector
OutboundConnector
InboundConnector
BidirectionalConnector
MessageHandler
Descriptor
BaseConnector and State Machine
Building an Outbound Connector: GitHub API
Building an Inbound Connector: Webhook Receiver
Registration and Discovery
LifecycleManager for Reconnection
Health Checks
Using Connectors in Workflows
Testing Connectors
Connector Categories Catalog

Architecture Overview¶

YAML flow file
  |
  +-- connectors:
  |     - name: my-github
  |       type: github
  |       config: { token: "..." }
  |
  +-- workflows:
        - from: timer://tick?period=60s
          steps:
            - target: connector://my-github?operation=list-issues

The connector system has three layers:

Registry (pkg/connector/registry.go): A global registry where connector factories are registered via init(). Each factory knows how to create a connector instance from a config map.
ConnectorMeta (pkg/connector/component.go): A meta-component registered in the engine under the connector:// scheme. It resolves connector://name?operation=X URIs to the correct connector instance.
LifecycleManager (pkg/connector/lifecycle.go): Optional wrapper that adds automatic reconnection, health monitoring, and exponential backoff.

Core Interfaces¶

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

Connector¶

The base interface every connector must implement:

type Connector interface {
    // ID returns the unique instance identifier.
    ID() string

    // Type returns the connector type name (e.g. "github").
    Type() string

    // Connect initializes the connection to the external system.
    Connect(ctx context.Context) error

    // Disconnect gracefully shuts down the connection.
    Disconnect(ctx context.Context) error

    // IsConnected reports whether the connector has an active connection.
    IsConnected() bool

    // HealthCheck probes the external system and returns nil if healthy.
    HealthCheck(ctx context.Context) error

    // Descriptor returns catalog metadata.
    Descriptor() *Descriptor
}

OutboundConnector¶

For connectors that send messages to external systems:

type OutboundConnector interface {
    Connector

    // Send transmits a message and returns an optional response.
    Send(ctx context.Context, msg *exchange.Message) (*exchange.Message, error)
}

InboundConnector¶

For connectors that receive messages from external systems:

type InboundConnector interface {
    Connector

    // Subscribe registers a handler called for every inbound message.
    Subscribe(ctx context.Context, handler MessageHandler) error
}

BidirectionalConnector¶

For connectors that support both directions:

type BidirectionalConnector interface {
    OutboundConnector
    InboundConnector
}

MessageHandler¶

The callback signature for inbound message delivery:

type MessageHandler func(ctx context.Context, msg *exchange.Message) error

Descriptor¶

Catalog metadata used for discovery and documentation:

type Descriptor struct {
    Type        string            `json:"type"`
    Name        string            `json:"name"`
    Description string            `json:"description"`
    Category    string            `json:"category"`
    Schema      map[string]any    `json:"schema,omitempty"`
}

BaseConnector and State Machine¶

The SDK provides BaseConnector as an embeddable struct that handles common concerns:

type BaseConnector struct {
    id         string
    connType   string
    descriptor *Descriptor
    state      atomic.Int32  // 0=disconnected, 1=connecting, 2=connected, 3=disconnecting
}

State transitions are enforced atomically:

Disconnected -> Connecting -> Connected -> Disconnecting -> Disconnected
     0              1            2              3               0

Embed BaseConnector to get ID(), Type(), Descriptor(), IsConnected(), and state management for free:

type MyConnector struct {
    connector.BaseConnector
    token  string
    client *http.Client
}

Call SetConnected(true) at the end of Connect() and SetConnected(false) at the end of Disconnect() to update the state machine.

Building an Outbound Connector: GitHub API¶

This example builds a connector that interacts with the GitHub API to list issues, create issues, and get repository information.

// pkg/connector/connectors/github/github.go
package github

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

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

type GitHubConnector struct {
    connector.BaseConnector
    token  string
    owner  string
    repo   string
    client *http.Client
}

func NewGitHubConnector(id string, config map[string]any) (connector.Connector, error) {
    token, _ := config["token"].(string)
    if token == "" {
        return nil, fmt.Errorf("github: token is required")
    }
    owner, _ := config["owner"].(string)
    repo, _ := config["repo"].(string)

    c := &GitHubConnector{
        token: token,
        owner: owner,
        repo:  repo,
    }
    c.InitBase(id, "github", &connector.Descriptor{
        Type:        "github",
        Name:        "GitHub",
        Description: "GitHub API connector for issues, PRs, and repository management",
        Category:    "devops",
    })
    return c, nil
}

func (c *GitHubConnector) Connect(ctx context.Context) error {
    c.client = &http.Client{}

    // Validate credentials by calling the user endpoint.
    req, err := http.NewRequestWithContext(ctx, "GET", "https://api.github.com/user", nil)
    if err != nil {
        return fmt.Errorf("github: %w", err)
    }
    req.Header.Set("Authorization", "Bearer "+c.token)
    req.Header.Set("Accept", "application/vnd.github+json")

    resp, err := c.client.Do(req)
    if err != nil {
        return fmt.Errorf("github: connection failed: %w", err)
    }
    resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("github: authentication failed (status %d)", resp.StatusCode)
    }

    c.SetState(connector.StateConnected)
    return nil
}

func (c *GitHubConnector) Disconnect(ctx context.Context) error {
    if c.client != nil {
        c.client.CloseIdleConnections()
    }
    c.SetState(connector.StateDisconnected)
    return nil
}

func (c *GitHubConnector) HealthCheck(ctx context.Context) error {
    req, err := http.NewRequestWithContext(ctx, "GET", "https://api.github.com/rate_limit", nil)
    if err != nil {
        return err
    }
    req.Header.Set("Authorization", "Bearer "+c.token)

    resp, err := c.client.Do(req)
    if err != nil {
        return fmt.Errorf("github: health check failed: %w", err)
    }
    resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("github: unhealthy (status %d)", resp.StatusCode)
    }
    return nil
}

func (c *GitHubConnector) Send(ctx context.Context, msg *exchange.Message) (*exchange.Message, error) {
    operation := msg.Header("operation")
    if operation == "" {
        operation = "list-issues"
    }

    owner := c.owner
    if o := msg.Header("owner"); o != "" {
        owner = o
    }
    repo := c.repo
    if r := msg.Header("repo"); r != "" {
        repo = r
    }

    if owner == "" || repo == "" {
        return nil, fmt.Errorf("github: owner and repo are required")
    }

    switch operation {
    case "list-issues":
        return c.listIssues(ctx, owner, repo)
    case "create-issue":
        return c.createIssue(ctx, owner, repo, msg)
    case "get-repo":
        return c.getRepo(ctx, owner, repo)
    default:
        return nil, fmt.Errorf("github: unknown operation %q", operation)
    }
}

func (c *GitHubConnector) listIssues(ctx context.Context, owner, repo string) (*exchange.Message, error) {
    url := fmt.Sprintf("https://api.github.com/repos/%s/%s/issues", owner, repo)
    body, err := c.doGet(ctx, url)
    if err != nil {
        return nil, err
    }

    resp := exchange.NewMessage()
    resp.SetBody(body)
    resp.SetHeader("github.operation", "list-issues")
    return resp, nil
}

func (c *GitHubConnector) createIssue(ctx context.Context, owner, repo string, msg *exchange.Message) (*exchange.Message, error) {
    url := fmt.Sprintf("https://api.github.com/repos/%s/%s/issues", owner, repo)

    var payload []byte
    switch v := msg.Body().(type) {
    case string:
        payload = []byte(v)
    case []byte:
        payload = v
    case map[string]any:
        var err error
        payload, err = json.Marshal(v)
        if err != nil {
            return nil, fmt.Errorf("github: failed to marshal body: %w", err)
        }
    default:
        return nil, fmt.Errorf("github: unsupported body type %T for create-issue", msg.Body())
    }

    req, err := http.NewRequestWithContext(ctx, "POST", url, strings.NewReader(string(payload)))
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+c.token)
    req.Header.Set("Accept", "application/vnd.github+json")
    req.Header.Set("Content-Type", "application/json")

    httpResp, err := c.client.Do(req)
    if err != nil {
        return nil, fmt.Errorf("github: create issue failed: %w", err)
    }
    defer httpResp.Body.Close()

    respBody, _ := io.ReadAll(httpResp.Body)

    var result any
    json.Unmarshal(respBody, &result)

    resp := exchange.NewMessage()
    resp.SetBody(result)
    resp.SetHeader("github.operation", "create-issue")
    resp.SetHeader("github.status", fmt.Sprintf("%d", httpResp.StatusCode))
    return resp, nil
}

func (c *GitHubConnector) getRepo(ctx context.Context, owner, repo string) (*exchange.Message, error) {
    url := fmt.Sprintf("https://api.github.com/repos/%s/%s", owner, repo)
    body, err := c.doGet(ctx, url)
    if err != nil {
        return nil, err
    }

    resp := exchange.NewMessage()
    resp.SetBody(body)
    resp.SetHeader("github.operation", "get-repo")
    return resp, nil
}

func (c *GitHubConnector) doGet(ctx context.Context, url string) (any, error) {
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+c.token)
    req.Header.Set("Accept", "application/vnd.github+json")

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

    body, _ := io.ReadAll(resp.Body)

    var result any
    json.Unmarshal(body, &result)
    return result, nil
}

Building an Inbound Connector: Webhook Receiver¶

This example builds a connector that listens for incoming HTTP webhooks and delivers them into Helix workflows.

// pkg/connector/connectors/webhook/webhook.go
package webhook

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

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

type WebhookConnector struct {
    connector.BaseConnector
    addr    string
    path    string
    secret  string
    server  *http.Server
    handler connector.MessageHandler
    mu      sync.Mutex
}

func NewWebhookConnector(id string, config map[string]any) (connector.Connector, error) {
    addr, _ := config["addr"].(string)
    if addr == "" {
        addr = ":8090"
    }
    path, _ := config["path"].(string)
    if path == "" {
        path = "/webhook"
    }
    secret, _ := config["secret"].(string)

    c := &WebhookConnector{
        addr:   addr,
        path:   path,
        secret: secret,
    }
    c.InitBase(id, "webhook", &connector.Descriptor{
        Type:        "webhook",
        Name:        "Webhook Receiver",
        Description: "Receives incoming HTTP webhooks and delivers them into workflows",
        Category:    "communication",
    })
    return c, nil
}

func (c *WebhookConnector) Connect(ctx context.Context) error {
    mux := http.NewServeMux()
    mux.HandleFunc(c.path, c.handleWebhook)

    c.server = &http.Server{
        Addr:    c.addr,
        Handler: mux,
    }

    go func() {
        slog.Info("webhook: listening", "addr", c.addr, "path", c.path)
        if err := c.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            slog.Error("webhook: server error", "error", err)
        }
    }()

    c.SetState(connector.StateConnected)
    return nil
}

func (c *WebhookConnector) Disconnect(ctx context.Context) error {
    if c.server != nil {
        if err := c.server.Shutdown(ctx); err != nil {
            return fmt.Errorf("webhook: shutdown error: %w", err)
        }
    }
    c.SetState(connector.StateDisconnected)
    return nil
}

func (c *WebhookConnector) HealthCheck(ctx context.Context) error {
    if !c.IsConnected() {
        return fmt.Errorf("webhook: not connected")
    }
    return nil
}

func (c *WebhookConnector) Subscribe(ctx context.Context, handler connector.MessageHandler) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.handler = handler
    return nil
}

func (c *WebhookConnector) handleWebhook(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
        return
    }

    // Optional secret validation.
    if c.secret != "" {
        provided := r.Header.Get("X-Webhook-Secret")
        if provided != c.secret {
            http.Error(w, "unauthorized", http.StatusUnauthorized)
            return
        }
    }

    body, err := io.ReadAll(io.LimitReader(r.Body, 10*1024*1024)) // 10MB limit
    if err != nil {
        http.Error(w, "failed to read body", http.StatusBadRequest)
        return
    }
    defer r.Body.Close()

    c.mu.Lock()
    handler := c.handler
    c.mu.Unlock()

    if handler == nil {
        http.Error(w, "no handler registered", http.StatusServiceUnavailable)
        return
    }

    // Build a message from the webhook payload.
    msg := exchange.NewMessage()

    // Try to parse as JSON, fall back to raw string.
    var jsonBody any
    if err := json.Unmarshal(body, &jsonBody); err == nil {
        msg.SetBody(jsonBody)
    } else {
        msg.SetBody(string(body))
    }

    // Forward select HTTP headers.
    msg.SetHeader("webhook.content-type", r.Header.Get("Content-Type"))
    msg.SetHeader("webhook.method", r.Method)
    msg.SetHeader("webhook.path", r.URL.Path)
    msg.SetHeader("webhook.remote-addr", r.RemoteAddr)
    for key, vals := range r.Header {
        if len(vals) > 0 {
            msg.SetHeader("webhook.header."+key, vals[0])
        }
    }

    if err := handler(r.Context(), msg); err != nil {
        slog.Error("webhook: handler error", "error", err)
        http.Error(w, "processing error", http.StatusInternalServerError)
        return
    }

    w.WriteHeader(http.StatusOK)
    w.Write([]byte(`{"status":"accepted"}`))
}

Registration and Discovery¶

Register your connector factory in an init() function so it is discovered automatically:

// pkg/connector/connectors/github/register.go
package github

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

func init() {
    connector.Register("github", NewGitHubConnector, &connector.Descriptor{
        Type:        "github",
        Name:        "GitHub",
        Description: "GitHub API connector for issues, PRs, and repository management",
        Category:    "devops",
    })
}

Then add the import to the aggregator file:

// pkg/connector/connectors/all/all.go
import (
    // ... existing imports ...
    _ "github.com/arvenza/helix/pkg/connector/connectors/github"
    _ "github.com/arvenza/helix/pkg/connector/connectors/webhook"
)

The main CLI binary imports all to trigger all registrations:

import _ "github.com/arvenza/helix/pkg/connector/connectors/all"

At runtime, connector.Create("github", "my-github", config) looks up the factory and creates an instance.

LifecycleManager for Reconnection¶

The LifecycleManager wraps a connector with automatic reconnection and periodic health checks:

cfg := connector.DefaultLifecycleConfig()
cfg.HealthCheckInterval = 30 * time.Second
cfg.MaxReconnectAttempts = 10
cfg.InitialReconnectDelay = 2 * time.Second

connMeta.AddWithLifecycle("my-github", conn, cfg)

The lifecycle manager:

Calls Connect() on startup.
Runs HealthCheck() on a configurable interval.
If HealthCheck() fails, calls Disconnect() then Connect() with exponential backoff.
Stops retrying after MaxReconnectAttempts (0 = unlimited).

In YAML, enable lifecycle management per connector:

connectors:
  - name: my-github
    type: github
    config:
      token: "${env:GITHUB_TOKEN}"
      owner: myorg
      repo: myrepo
    lifecycle:
      reconnect: true
      health_interval: 30s
      max_retries: 10
      retry_delay: 2s

Health Checks¶

Every connector must implement HealthCheck(ctx) error. Best practices:

Probe the external system with a lightweight operation (e.g., API ping, SELECT 1, list buckets).
Respect the context deadline for timeout.
Return nil for healthy, an error for unhealthy.
Do not modify connector state in HealthCheck.

func (c *GitHubConnector) HealthCheck(ctx context.Context) error {
    req, err := http.NewRequestWithContext(ctx, "GET", "https://api.github.com/rate_limit", nil)
    if err != nil {
        return err
    }
    req.Header.Set("Authorization", "Bearer "+c.token)

    resp, err := c.client.Do(req)
    if err != nil {
        return fmt.Errorf("github: health check failed: %w", err)
    }
    resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("github: unhealthy (status %d)", resp.StatusCode)
    }
    return nil
}

Using Connectors in Workflows¶

Connectors are accessed in workflows via the connector:// URI scheme:

connectors:
  - name: my-github
    type: github
    config:
      token: "${env:GITHUB_TOKEN}"
      owner: myorg
      repo: myrepo

workflows:
  # Outbound: send to connector
  - id: create-github-issue
    trigger:
      uri: http-server://0.0.0.0:8080/api/issues
    steps:
      - set-header:
          name: operation
          constant: create-issue
      - target: connector://my-github

  # Outbound: list issues on a timer
  - id: poll-issues
    trigger:
      uri: timer://poll?period=5m
    steps:
      - set-header:
          name: operation
          constant: list-issues
      - target: connector://my-github
      - target: log://issues?showBody=true

  # Inbound: receive webhooks
  - id: webhook-handler
    trigger:
      uri: connector://my-webhook
    steps:
      - log:
          message: "Received webhook"
      - target: kafka://webhook-events?brokers=localhost:9092

The operation header or ?operation=X query parameter on the URI tells the connector which action to perform:

- target: connector://my-github?operation=list-issues

Testing Connectors¶

Unit testing with httptest¶

package github

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

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

func TestGitHubConnector_ListIssues(t *testing.T) {
    // Mock GitHub API.
    server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        switch r.URL.Path {
        case "/user":
            w.WriteHeader(http.StatusOK)
            json.NewEncoder(w).Encode(map[string]any{"login": "testuser"})
        case "/repos/myorg/myrepo/issues":
            json.NewEncoder(w).Encode([]map[string]any{
                {"number": 1, "title": "Bug report"},
                {"number": 2, "title": "Feature request"},
            })
        default:
            w.WriteHeader(http.StatusNotFound)
        }
    }))
    defer server.Close()

    // Create connector (you would need to make baseURL configurable for testing).
    conn, err := NewGitHubConnector("test-gh", map[string]any{
        "token": "test-token",
        "owner": "myorg",
        "repo":  "myrepo",
    })
    if err != nil {
        t.Fatal(err)
    }

    ctx := context.Background()
    if err := conn.Connect(ctx); err != nil {
        t.Fatal(err)
    }
    defer conn.Disconnect(ctx)

    // Send a list-issues request.
    outbound := conn.(connector.OutboundConnector)
    msg := exchange.NewMessage()
    msg.SetHeader("operation", "list-issues")

    resp, err := outbound.Send(ctx, msg)
    if err != nil {
        t.Fatal(err)
    }

    // Verify response.
    issues, ok := resp.Body().([]any)
    if !ok {
        t.Fatalf("expected []any, got %T", resp.Body())
    }
    if len(issues) != 2 {
        t.Errorf("expected 2 issues, got %d", len(issues))
    }
}

Testing inbound connectors¶

func TestWebhookConnector_ReceivesPayload(t *testing.T) {
    conn, err := NewWebhookConnector("test-wh", map[string]any{
        "addr": ":0", // random port
        "path": "/hook",
    })
    if err != nil {
        t.Fatal(err)
    }

    ctx := context.Background()
    if err := conn.Connect(ctx); err != nil {
        t.Fatal(err)
    }
    defer conn.Disconnect(ctx)

    // Subscribe to inbound messages.
    received := make(chan *exchange.Message, 1)
    inbound := conn.(connector.InboundConnector)
    inbound.Subscribe(ctx, func(ctx context.Context, msg *exchange.Message) error {
        received <- msg
        return nil
    })

    // Send a webhook POST (you would get the actual port from the server).
    // ... POST to http://localhost:<port>/hook with JSON body ...

    // Assert the message was received.
    // msg := <-received
    // assert msg.Body() contains expected data
}

Connector Categories Catalog¶

Helix includes 441 connectors across these 33 categories:

Category	Count	Examples
ai	14	openai, anthropic, ollama, huggingface, cohere, mistral
aws	46	s3, sqs, sns, dynamodb, lambda, kinesis, ec2, rds, secrets_manager, eventbridge
azure	21	blob_storage, service_bus, cosmos_db, functions, event_hubs, key_vault
blockchain	2	ethereum, hyperledger
caching	6	redis, memcached, hazelcast, ehcache, caffeine, redis_json
cdc	6	debezium_mysql, debezium_postgres, debezium_mongodb, dynamodb_streams, mongodb_change_stream, pg_logical
communication	25	slack, teams, discord, twilio, telegram, email, webhook, sendgrid
content	3	wordpress, contentful, strapi
core	12	direct, seda, timer, cron, mock, log, bean, ref, language, control_bus, browse, dataset
database	27	postgres, mysql, mongodb, elasticsearch, clickhouse, neo4j, cassandra, couchdb, influxdb, sqlite
dataformat	17	json, xml, csv, yaml, avro, protobuf, parquet, msgpack, toml, bson
devops	12	github, gitlab, jenkins, docker, ansible, terraform, vault, consul, pagerduty, opsgenie
ecommerce	13	shopify, stripe, paypal, woocommerce, magento, square, braintree
edi	5	hl7, fhir, as2, rosettanet, tradacoms (X12/EDIFACT → `edi://` component)
enterprise	13	sap, servicenow, jira, confluence, sharepoint, dynamics365, netsuite
erp	12	sap_rfc, sap_idoc, oracle_ebs, odoo, sage, infor, epicor
file	23	file, ftp, sftp, s3, gcs, azure_blob, hdfs, nfs, smb, minio, webdav
financial	3	plaid, quickbooks, xero
google	34	bigquery, pubsub, gcs, datastore, spanner, firebase, sheets, drive, gmail, calendar
infrastructure	11	ssh, snmp, ldap, dns, ntp, syslog, netconf, prometheus_remote
iot	2	mqtt, opc_ua
kubernetes	20	pods, deployments, services, configmaps, secrets, jobs, cronjobs, ingresses, namespaces, nodes
messaging	19	kafka, rabbitmq, nats, activemq, pulsar, redis_streams, zeromq, mqtt
microsoft	22	teams, outlook, graph_api, sharepoint, onedrive, power_automate, excel
misc	31	http, grpc, graphql, websocket, sse, jsonrpc, soap, thrift
monitoring	10	prometheus, datadog, grafana, new_relic, splunk, elastic_apm, cloudwatch
openstack	6	nova, swift, neutron, cinder, glance, keystone
productivity	12	notion, airtable, trello, asana, monday, clickup, linear, basecamp
salesforce	10	sobject, bulk_api, streaming, apex, reports, metadata, chatter, platform_events
scheduling	2	quartz, temporal
search	5	elasticsearch, solr, algolia, meilisearch, typesense
storage	8	minio, ceph, glusterfs, lustre, seaweedfs, garage, lakeFS, delta_lake
workflow	4	temporal, airflow, prefect, dagster

Each connector is registered in its own package under pkg/connector/connectors/<category>/<type>/ and auto-discovered via the all.go aggregator import.

To list all available connector types at runtime:

types := connector.RegisteredTypes()
for _, t := range types {
    desc := connector.GetDescriptor(t)
    fmt.Printf("%-20s %-12s %s\n", desc.Type, desc.Category, desc.Description)
}

Or from the CLI, connectors appear alongside components when referenced in flow files. The helix validate command will report any connectors defined in the flow.