Configuration reference

A pipeline config defines where data comes from (source), how it is processed (transforms), and where it is delivered (sinks).

source → transforms[] → sinks[]

BeamConfig

The top-level object returned by all config endpoints.

{
  "id": "abc123",
  "name": "USDC Transfer Monitor",
  "description": "Monitors USDC ERC20 transfers on Base",
  "tags": ["production", "base"],
  "static_egress_ip": false,
  "pod_size": "S",
  "delivery_guarantee": "at_least_once",
  "owner_org_team_user_id": "org_xyz",
  "organization_id": "org123",
  "created_at": "2025-01-15T10:30:00Z",
  "updated_at": "2025-01-15T10:30:00Z",
  "pipeline_config": {
    "source": { ... },
    "transforms": [ ... ],
    "sinks": [ ... ]
  }
}

Field	Type	Required	Description
`id`	`string`	Auto-generated	Pipeline ID. Omit on create; the server assigns it.
`name`	`string`	Yes	Pipeline name
`description`	`string`	Yes	Pipeline description
`tags`	`string[]`	No	Tags for organizing pipelines. Default: `null`
`static_egress_ip`	`boolean`	No	Send webhook traffic from a static outbound IP. Default: `false`
`pod_size`	`string`	No	Worker size: `XXS`, `XS`, `S`, `M`, `L`, or `XL`. Default: `XS`
`delivery_guarantee`	`string`	No	`at_least_once`: no data loss; if the pipeline errors, intake pauses until it recovers. `best_effort`: at-most-once; records may be dropped on errors and intake continues. Default: `at_least_once`
`owner_org_team_user_id`	`string`	Auto-set	Owner ID, set from the authenticated API key
`organization_id`	`string \| null`	Auto-set	Organization that owns the pipeline, set from the authenticated API key
`created_at`	`datetime`	Auto-set	Creation timestamp
`updated_at`	`datetime`	Auto-set	Last update timestamp
`pipeline_config`	`PipelineConfig`	Yes	The pipeline’s source, transforms, and sinks

CreateBeamConfigRequest

Request body for Create pipeline. Same as BeamConfig, without the server-set fields.

{
  "name": "USDC Transfer Monitor",
  "description": "Monitors USDC ERC20 transfers on Base",
  "tags": ["production", "base"],
  "static_egress_ip": false,
  "pod_size": "S",
  "delivery_guarantee": "at_least_once",
  "pipeline_config": {
    "source": { ... },
    "transforms": [ ... ],
    "sinks": [ ... ]
  }
}

Field	Type	Required	Description
`name`	`string`	Yes	Pipeline name
`description`	`string`	Yes	Pipeline description
`tags`	`string[]`	No	Tags for organizing pipelines
`static_egress_ip`	`boolean`	No	Send webhook traffic from a static outbound IP. Default: `false`
`pod_size`	`string`	No	Worker size: `XXS`, `XS`, `S`, `M`, `L`, or `XL`. Default: `XS`
`delivery_guarantee`	`string`	No	`at_least_once`: no data loss; if the pipeline errors, intake pauses until it recovers. `best_effort`: at-most-once; records may be dropped on errors and intake continues. Default: `at_least_once`
`pipeline_config`	`PipelineConfig`	Yes	The pipeline’s source, transforms, and sinks

UpdateBeamConfigRequest

Request body for Update pipeline. Same shape as CreateBeamConfigRequest. The pipeline is identified by config_id in the URL path. Server-set fields (id, owner_org_team_user_id, organization_id, created_at, updated_at) keep their existing values and are ignored if sent.

{
  "name": "USDC Transfer Monitor (updated)",
  "description": "Updated description",
  "tags": ["production", "base"],
  "static_egress_ip": false,
  "pod_size": "S",
  "delivery_guarantee": "at_least_once",
  "pipeline_config": {
    "source": { ... },
    "transforms": [ ... ],
    "sinks": [ ... ]
  }
}

PipelineConfig

{
  "source": { ... },
  "transforms": [ ... ],
  "sinks": [ ... ]
}

Field	Type	Required	Description
`source`	`Source`	Yes	Where data comes from
`transforms`	`Transform[]`	Yes	Processing steps applied in order
`sinks`	`Sink[]`	Yes	Where processed data is delivered. Exactly one sink is allowed per pipeline

Source

The Allium Datastream the pipeline reads from, identified by chain and entity.

{
  "type": "pubsub",
  "chain": "base",
  "entity": "erc20_token_transfer",
  "is_zerolag": false
}

Field	Type	Required	Description
`type`	`string`	Yes	`"pubsub"` or `"kafka"`. Use the `source_type` that List sources returns for the chain and entity
`chain`	`string`	Yes	Blockchain to read from
`entity`	`string`	Yes	Entity to stream
`is_zerolag`	`boolean`	No	`pubsub` sources only. Stream from the chain tip before finality. Lower latency, but may include reorged data. Default: `false`

For kafka sources, set only type, chain, and entity. The connection fields keep their defaults.

Supported chains and entities

Chain	Entities	Zerolag
Polygon	`log`, `decoded_log`, `erc20_token_transfer`, `erc721_token_transfer`, `erc1155_token_transfer`	Yes
Base	`log`, `decoded_log`, `erc20_token_transfer`, `erc721_token_transfer`, `erc1155_token_transfer`	Yes
Solana	`nonvoting_transaction`	No
Hyperliquid	`block`, `trade`, `fill`, `order`, `misc_event`	No

Coverage expands regularly. For current availability, see the Datastreams catalog or call GET /api/v1/beam/sources, which requires a Beam-enabled organization.

Transforms

Transforms run in order. Two types are available. Omit uid when adding a transform; the server assigns one.

Set filter (deprecated)
JavaScript

Set filters (redis_set_filter) are deprecated. New pipelines start with a JavaScript transform. Move your filter values to a dataset and check them with beam.contains() in a JavaScript transform instead.

Passes only records whose selected field matches a value in your set. Sets hold 10M+ values.

{
  "type": "redis_set_filter",
  "uid": "tf-001",
  "filter_expr": "root = this.token_address"
}

Field	Type	Required	Description
`type`	`"redis_set_filter"`	Yes	Must be `redis_set_filter`
`uid`	`string`	No	Auto-generated if omitted. Include to reference an existing transform. Max 128 characters, unique within `transforms`
`filter_expr`	`string`	Yes	Expression that selects the field to match, in the form `root = this.<field>`

Common filter_expr patterns:

Expression	Filters by
`root = this.address`	Contract address
`root = this.topic0`	Event signature
`root = this.from_address`	Transaction sender
`root = this.to_address`	Transaction recipient

Manage filter values with the filter values endpoints. Changes take effect immediately, with no redeploy.

Use lowercase values when filtering by addresses, labels, or symbols.

Modify, enrich, or reshape each record with a JavaScript function. Return null to drop the record.

{
  "type": "v8",
  "uid": "tf-002",
  "script": "function transform(record) { record.parsed = true; return record; }"
}

Field	Type	Required	Description
`type`	`"v8"`	Yes	Must be `v8`
`uid`	`string`	No	Auto-generated if omitted. Include to reference an existing transform. Max 128 characters, unique within `transforms`
`script`	`string`	Yes	JavaScript code that processes each record

Example: tag each transfer by size.

function transform(record) {
  const amount = parseFloat(record.amount || "0");
  if (amount > 1000000) {
    record.size_tag = "whale";
  } else if (amount > 10000) {
    record.size_tag = "medium";
  } else {
    record.size_tag = "small";
  }
  return record;
}

Sinks

A sink is where the pipeline delivers records. Each pipeline has exactly one sink, so sinks takes a single entry. Omit uid when adding a sink; the server assigns one. Five types are available:

Kafka
SNS
External Kafka
Webhook
NATS JetStream

Delivers to an Allium-managed Kafka topic. After deployment, the Allium App shows connection credentials and consumer code samples.

{
  "type": "kafka",
  "uid": "sk-001",
  "name": "usdc-transfers"
}

Field	Type	Required	Description
`type`	`"kafka"`	Yes	Must be `kafka`
`uid`	`string`	No	Auto-generated if omitted. Max 128 characters, unique within `sinks`
`name`	`string`	Yes	Topic name suffix. Full topic: `beam.{config_id}.{name}`. Letters, digits, `.`, `_` and `-`; max 100 characters

Delivers to an Allium-managed SNS topic.

{
  "type": "sns",
  "uid": "sk-002",
  "name": "my-topic"
}

Field	Type	Required	Description
`type`	`"sns"`	Yes	Must be `sns`
`uid`	`string`	No	Auto-generated if omitted. Max 128 characters, unique within `sinks`
`name`	`string`	Yes	Topic name suffix. Full topic: `beam-{config_id}-{name}`. Letters, digits, `_` and `-` (no `.`); max 100 characters

Delivers to your own Kafka cluster.

{
  "type": "external_kafka",
  "uid": "sk-003",
  "name": "my-external-topic",
  "bootstrap_servers": "broker1.example.com:9092,broker2.example.com:9092",
  "username_secret_id": "my-kafka-username",
  "password_secret_id": "my-kafka-password",
  "sasl_mechanism": "PLAIN"
}

Field	Type	Required	Description
`type`	`"external_kafka"`	Yes	Must be `external_kafka`
`uid`	`string`	No	Auto-generated if omitted. Max 128 characters, unique within `sinks`
`name`	`string`	Yes	Kafka topic name on your cluster. Letters, digits, `.`, `_` and `-`; max 100 characters
`bootstrap_servers`	`string`	Yes	Comma-separated broker addresses
`username_secret_id`	`string`	Yes	Organization secret id for the SASL username
`password_secret_id`	`string`	Yes	Organization secret id for the SASL password
`security_protocol`	`string`	No	Default: `SASL_SSL`
`sasl_mechanism`	`string`	No	`PLAIN` or `SCRAM-SHA-512`. Default: `PLAIN`

If a referenced secret changes after deployment, the sink is flagged as stale. Redeploy to apply the new value.

Messages can be up to 1 MB. Set the topic’s max.message.bytes to at least 1 MB, or your cluster may reject deliveries.

Delivers each record as a POST request to your HTTPS endpoint.

{
  "type": "webhook",
  "uid": "sk-004",
  "name": "my-webhook",
  "url": "https://example.com/webhook/receive",
  "hash_key": "YOUR_SECRET_ID",
  "enable_http_encoding": false,
  "max_in_flight": 64,
  "retries": 3,
  "retry_period": "1s",
  "timeout": "5s"
}

Field	Type	Required	Description
`type`	`"webhook"`	Yes	Must be `webhook`
`uid`	`string`	No	Auto-generated if omitted. Max 128 characters, unique within `sinks`
`name`	`string`	Yes	Descriptive name for the webhook sink. Letters, digits, `.`, `_` and `-`; max 100 characters
`url`	`string`	Yes	HTTPS endpoint URL. Must use `https://`
`hash_key`	`string`	No	Organization secret id holding the HMAC signing key. When set, Beam signs each request — see Webhook authentication
`enable_http_encoding`	`boolean`	No	When `true`, Beam compresses each request body with zstd and sets `Content-Encoding: zstd` on the POST. Your endpoint must decompress the payload before parsing. Default: `false`
`max_in_flight`	`integer`	No	Maximum number of requests Beam sends to your endpoint at once. Must be `1` or more. Default: `64`
`retries`	`integer`	No	How many times Beam retries a request that fails. Must be `1` or more. Default: `3`
`retry_period`	`string`	No	How long Beam waits between retries, as a Go duration such as `500ms`, `1s` or `1m30s` (units `ns`, `us`, `µs`, `ms`, `s`, `m`, `h`). Default: `"1s"`
`timeout`	`string`	No	How long Beam waits for your endpoint to respond to each request, as a Go duration such as `500ms`, `5s` or `1m30s` (units `ns`, `us`, `µs`, `ms`, `s`, `m`, `h`). Default: `"5s"`

The URL must use HTTPS. Plain HTTP is not supported.

If the hash_key secret is rotated after deployment, the sink is flagged as stale. Redeploy to apply the new value.

Publishes each record to a NATS JetStream subject on your own broker.

{
  "type": "nats_jetstream",
  "uid": "sk-005",
  "name": "my-jetstream",
  "url": "nats://broker.example.com:4222",
  "subject": "allium.ethereum.blocks",
  "token_secret_id": "MY_NATS_TOKEN"
}

Field	Type	Required	Description
`type`	`"nats_jetstream"`	Yes	Must be `nats_jetstream`
`uid`	`string`	No	Auto-generated if omitted. Max 128 characters, unique within `sinks`
`name`	`string`	Yes	Descriptive name for the sink. Letters, digits, `.`, `_`, `-` and inner spaces; max 100 characters
`url`	`string`	Yes	Broker URL, e.g. `nats://broker.example.com:4222`
`subject`	`string`	Yes	JetStream subject the pipeline publishes to, e.g. `allium.ethereum.blocks`
`token_secret_id`	`string`	No	Organization secret id holding the auth token. When omitted, the sink connects to the broker without authentication

Don’t put the token in url. When token_secret_id is set, Beam adds the token to the connection at deploy time: with a token of your-token, nats://your-broker.example.com:4222 becomes nats://[email protected]:4222.

If the token_secret_id secret is rotated after deployment, redeploy to apply the new value.

Need a different sink type? Contact [email protected].

IDs

When you create a pipeline, transform, or sink, omit id or uid and the server assigns one. When you update, include it to reference the existing resource.

Resource	ID field	Behavior when omitted
Pipeline	`id`	New pipeline created with auto-generated ID
Transform	`uid`	New transform created with auto-generated UID
Sink	`uid`	New sink created with auto-generated UID

Realtime API

Beam API

Explorer API

Account API

Deep Dives

BeamConfig

CreateBeamConfigRequest

UpdateBeamConfigRequest

PipelineConfig

Source

Supported chains and entities

Transforms

Sinks

IDs

Realtime API

Beam API

Explorer API

Account API

Deep Dives

​BeamConfig

​CreateBeamConfigRequest

​UpdateBeamConfigRequest

​PipelineConfig

​Source

​Supported chains and entities

​Transforms

​Sinks

​IDs

BeamConfig

CreateBeamConfigRequest

UpdateBeamConfigRequest

PipelineConfig

Source

Supported chains and entities

Transforms

Sinks

IDs