Overview

Creates a new stream with a specified blockchain network, dataset type, destination, and optional filter function. Use this endpoint to programmatically set up real-time or historical blockchain data pipelines.

Key capabilities:

• Start from a specific block (backfills) or latest block (real-time streaming)
• Choose destination: webhook, S3, Azure Blob Storage, or PostgreSQL
• Attach a base64-encoded JavaScript filter to transform data before delivery
• Configure batch sizes, reorg detection, distance-from-tip, and elastic batching
• Set up multi-destination delivery with extra_destinations
• Receive email alerts on stream termination

Endpoint

Send a POST request to the following URL with your stream configuration in the request body.

POST https://api.quicknode.com/streams/rest/v1/streams

Request Parameters

Parameter	Type	Required	Description
name	string	Yes	A name for your stream
network	string	Yes	Blockchain network (see Supported Chains)
dataset	string	Yes	Dataset type (see Data Sources)
region	string	Yes	usa_east
dataset_batch_size	integer	Yes	Number of blocks to batch per delivery
elastic_batch_enabled	boolean	Yes	Auto-adjust batch size to 1 near chain tip
destination	string	Yes	webhook, s3, azure, or postgres
destination_attributes	object	Yes	Configuration for selected destination
status	string	Yes	active or paused
filter_function	string	No	Base64-encoded JavaScript filter function
start_range	integer	No	Starting block number (omit for latest)
end_range	integer	No	Ending block number (omit for continuous)
fix_block_reorgs	integer	No	1 to enable reorg detection and re-streaming
keep_distance_from_tip	integer	No	Blocks to stay behind chain tip
notification_email	string	No	Email for stream termination alerts
extra_destinations	array	No	Additional destinations (same structure as primary)

Destination Attributes

Configuration options vary based on the selected destination type. Select a tab below to view the required and optional attributes for each destination.

Webhook

Parameter	Type	Required	Description
url	string	Yes	Webhook endpoint URL
compression	string	Yes	none or gzip
max_retry	integer	Yes	Max retry attempts
retry_interval_sec	integer	Yes	Seconds between retries
post_timeout_sec	integer	Yes	Request timeout in seconds
headers	object	No	Custom headers as key-value pairs

S3

Parameter	Type	Required	Description
endpoint	string	Yes	S3 endpoint (e.g., s3.amazonaws.com)
bucket	string	Yes	Bucket name
secret_key	string	Yes	S3 secret key
access_key	string	No	S3 access key
object_prefix	string	No	Prefix for object keys
file_compression_type	string	Yes	none or gzip
file_type	string	Yes	.json
max_retry	integer	Yes	Max retry attempts
retry_interval_sec	integer	Yes	Seconds between retries
use_ssl	boolean	Yes	Enable SSL

Azure

Parameter	Type	Required	Description
storage_account	string	Yes	Azure storage account name
container	string	Yes	Container name
sas_token	string	Yes	Shared Access Signature token
blob_prefix	string	No	Prefix for blob names
file_compression_type	string	Yes	none or gzip
file_type	string	Yes	.json
max_retry	integer	Yes	Max retry attempts
retry_interval_sec	integer	Yes	Seconds between retries

PostgreSQL

Parameter	Type	Required	Description
host	string	Yes	Database host
port	integer	Yes	Database port
username	string	Yes	Database username
password	string	Yes	Database password
table_name	string	Yes	Target table name
sslmode	string	Yes	require or disable
max_retry	integer	Yes	Max retry attempts
retry_interval_sec	integer	Yes	Seconds between retries

Example Request

cURL

curl -X POST "https://api.quicknode.com/streams/rest/v1/streams" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -H "x-api-key: YOUR_API_KEY" \
  -d '{
    "name": "My Stream",
    "network": "ethereum-mainnet",
    "dataset": "block",
    "filter_function": "ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==",
    "region": "usa_east",
    "start_range": 100,
    "end_range": 200,
    "dataset_batch_size": 1,
    "destination": "webhook",
    "fix_block_reorgs": 0,
    "keep_distance_from_tip": 0,
    "elastic_batch_enabled": true,
    "destination_attributes": {
      "url": "https://webhook.site",
      "compression": "none",
      "headers": { "Authorization": "Bearer token" },
      "max_retry": 3,
      "retry_interval_sec": 1,
      "post_timeout_sec": 10
    },
    "status": "active"
  }'

CLI

qn stream create --name "My Stream" \
  --network ethereum-mainnet \
  --dataset block \
  --start 100 \
  --end 200 \
  --region usa_east \
  --webhook https://webhook.site \
  --status active \
  -o json

JavaScript

const headers = new Headers()
headers.append('accept', 'application/json')
headers.append('Content-Type', 'application/json')
headers.append('x-api-key', 'YOUR_API_KEY')

const body = JSON.stringify({
  name: 'My Stream',
  network: 'ethereum-mainnet',
  dataset: 'block',
  filter_function: 'ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==',
  region: 'usa_east',
  start_range: 100,
  end_range: 200,
  dataset_batch_size: 1,
  destination: 'webhook',
  fix_block_reorgs: 0,
  keep_distance_from_tip: 0,
  elastic_batch_enabled: true,
  destination_attributes: {
    url: 'https://webhook.site',
    compression: 'none',
    headers: { 'Authorization': 'Bearer token' },
    max_retry: 3,
    retry_interval_sec: 1,
    post_timeout_sec: 10
  },
  status: 'active'
})

fetch('https://api.quicknode.com/streams/rest/v1/streams', {
  method: 'POST',
  headers: headers,
  body: body
})
  .then(response => response.json())
  .then(data => console.log(data))
  .catch(error => console.error('Error:', error))

Python

import requests

url = "https://api.quicknode.com/streams/rest/v1/streams"

payload = {
    "name": "My Stream",
    "network": "ethereum-mainnet",
    "dataset": "block",
    "filter_function": "ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==",
    "region": "usa_east",
    "start_range": 100,
    "end_range": 200,
    "dataset_batch_size": 1,
    "destination": "webhook",
    "fix_block_reorgs": 0,
    "keep_distance_from_tip": 0,
    "elastic_batch_enabled": True,
    "destination_attributes": {
        "url": "https://webhook.site",
        "compression": "none",
        "headers": { "Authorization": "Bearer token" },
        "max_retry": 3,
        "retry_interval_sec": 1,
        "post_timeout_sec": 10
    },
    "status": "active"
}

headers = {
    "accept": "application/json",
    "Content-Type": "application/json",
    "x-api-key": "YOUR_API_KEY"
}

response = requests.post(url, headers=headers, json=payload)
print(response.json())

Ruby

require "uri"
require "net/http"
require "json"

url = URI("https://api.quicknode.com/streams/rest/v1/streams")
https = Net::HTTP.new(url.host, url.port)
https.use_ssl = true

request = Net::HTTP::Post.new(url)
request["accept"] = "application/json"
request["Content-Type"] = "application/json"
request["x-api-key"] = "YOUR_API_KEY"

request.body = {
  "name" => "My Stream",
  "network" => "ethereum-mainnet",
  "dataset" => "block",
  "filter_function" => "ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==",
  "region" => "usa_east",
  "start_range" => 100,
  "end_range" => 200,
  "dataset_batch_size" => 1,
  "destination" => "webhook",
  "fix_block_reorgs" => 0,
  "keep_distance_from_tip" => 0,
  "elastic_batch_enabled" => true,
  "destination_attributes" => {
    "url" => "https://webhook.site",
    "compression" => "none",
    "headers" => { "Authorization" => "Bearer token" },
    "max_retry" => 3,
    "retry_interval_sec" => 1,
    "post_timeout_sec" => 10
  },
  "status" => "active"
}.to_json

response = https.request(request)
puts response.body

SDK (JS)

import { QuicknodeSdk, StreamDataset, StreamRegion, StreamStatus } from "@quicknode/sdk"

const qn = new QuicknodeSdk({ apiKey: "YOUR_API_KEY" })

const filterFunction = Buffer.from(`
function main(data) {
  return {
    hash: data.streamData.hash,
    number: parseInt(data.streamData.number, 16),
  }
}
`).toString("base64")

const stream = await qn.streams.createStream({
  name: "My Stream",
  network: "ethereum-mainnet",
  dataset: StreamDataset.Block,
  region: StreamRegion.UsaEast,
  startRange: 100,
  endRange: 200,
  datasetBatchSize: 1,
  elasticBatchEnabled: true,
  fixBlockReorgs: 0,
  keepDistanceFromTip: 0,
  filterFunction,
  status: StreamStatus.Active,
  destinationAttributes: {
    destination: "webhook",
    attributes: {
      url: "https://webhook.site",
      compression: "none",
      maxRetry: 3,
      retryIntervalSec: 1,
      postTimeoutSec: 10,
    },
  },
})

console.log(stream.id, stream.status)

SDK (Python)

import asyncio
import base64
from quicknode_sdk import QuicknodeSdk, SdkFullConfig, StreamWebhookDestination, WebhookAttributes

async def main():
    qn = QuicknodeSdk(SdkFullConfig(api_key="YOUR_API_KEY"))

    filter_function = base64.b64encode(b"""
function main(data) {
    return {
        hash: data.streamData.hash,
        number: parseInt(data.streamData.number, 16)
    };
}
""").decode()

    stream = await qn.streams.create_stream(
        name="My Stream",
        network="ethereum-mainnet",
        dataset="block",
        region="usa_east",
        start_range=100,
        end_range=200,
        dataset_batch_size=1,
        elastic_batch_enabled=True,
        fix_block_reorgs=0,
        keep_distance_from_tip=0,
        filter_function=filter_function,
        status="active",
        destination_attributes=StreamWebhookDestination(
            WebhookAttributes(
                url="https://webhook.site",
                max_retry=3,
                retry_interval_sec=1,
                post_timeout_sec=10,
                compression="none",
            )
        ),
    )

    print(stream.id, stream.status)

asyncio.run(main())

SDK (Ruby)

require "base64"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_config(api_key: "YOUR_API_KEY")

filter_function = Base64.strict_encode64(<<~JS)
  function main(data) {
    return {
      hash: data.streamData.hash,
      number: parseInt(data.streamData.number, 16)
    };
  }
JS

destination = QuicknodeSdk::DestinationAttributes.webhook(
  url: "https://webhook.site",
  max_retry: 3,
  retry_interval_sec: 1,
  post_timeout_sec: 10,
  compression: "none",
)

stream = qn.streams.create_stream(
  name: "My Stream",
  network: "ethereum-mainnet",
  dataset: "block",
  region: "usa_east",
  start_range: 100,
  end_range: 200,
  dataset_batch_size: 1,
  elastic_batch_enabled: true,
  fix_block_reorgs: 0,
  keep_distance_from_tip: 0,
  filter_function: filter_function,
  status: "active",
  destination_attributes: destination,
)

puts "#{stream["id"]} #{stream["status"]}"

SDK (Rust)

use quicknode_sdk::{
    streams::{
        CreateStreamParams, DestinationAttributes, StreamDataset, StreamRegion, StreamStatus,
        WebhookAttributes,
    },
    QuicknodeSdk,
    SdkFullConfig,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::new(&SdkFullConfig::builder().api_key("YOUR_API_KEY").build())?;

    let filter_function = "ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==".to_string();

    let params = CreateStreamParams::builder()
        .name("My Stream".to_string())
        .network("ethereum-mainnet".to_string())
        .dataset(StreamDataset::Block)
        .region(StreamRegion::UsaEast)
        .start_range(100)
        .end_range(200)
        .dataset_batch_size(1)
        .elastic_batch_enabled(true)
        .fix_block_reorgs(0)
        .keep_distance_from_tip(0)
        .filter_function(filter_function)
        .status(StreamStatus::Active)
        .destination_attributes(DestinationAttributes::Webhook(WebhookAttributes {
            url: "https://webhook.site".to_string(),
            compression: "none".to_string(),
            max_retry: 3,
            retry_interval_sec: 1,
            post_timeout_sec: 10,
            security_token: None,
        }))
        .build();

    let stream = qn.streams.create_stream(&params).await?;

    println!("{} {}", stream.id, stream.status);
    Ok(())
}

Response

On success, the API returns the created stream object with a unique ID, timestamps, and all configured parameters.

{
  "id": "f6ad6459-b5ad-4183-b370-1c1388e47e83",
  "name": "My Stream",
  "network": "ethereum-mainnet",
  "dataset": "block",
  "region": "usa_east",
  "filter_function": "ZnVuY3Rpb24gbWFpbihkYXRhKSB7Li4ufQ==",
  "start_range": 100,
  "end_range": 200,
  "dataset_batch_size": 1,
  "elastic_batch_enabled": true,
  "fix_block_reorgs": 0,
  "keep_distance_from_tip": 0,
  "destination": "webhook",
  "destination_attributes": {
    "url": "https://webhook.site",
    "compression": "none",
    "headers": { "Authorization": "Bearer token" },
    "max_retry": 3,
    "retry_interval_sec": 1,
    "post_timeout_sec": 10,
    "security_token": "qn_xxxxxxxxxxxxxx"
  },
  "status": "active",
  "sequence": 0,
  "current_hash": null,
  "notification_email": null,
  "extra_destinations": [],
  "created_at": "2024-01-15T10:30:00Z",
  "updated_at": "2024-01-15T10:30:00Z"
}

Response Fields

Field	Type	Description
id	string	Unique stream identifier (UUID)
name	string	Stream name
network	string	Blockchain network
dataset	string	Dataset type
region	string	Deployment region
filter_function	string	Base64-encoded filter function (if configured)
start_range	integer	Starting block number
end_range	integer	Ending block number (null for continuous)
dataset_batch_size	integer	Number of blocks per batch
elastic_batch_enabled	boolean	Whether elastic batching is enabled
fix_block_reorgs	integer	Reorg detection setting (1 = enabled)
keep_distance_from_tip	integer	Blocks to stay behind chain tip
destination	string	Destination type (webhook, s3, azure, postgres)
destination_attributes	object	Destination configuration (varies by type)
status	string	Stream status (active or paused)
sequence	integer	Last delivered block from the data range
current_hash	string	Hash of the current block being processed
notification_email	string	Email for stream termination alerts
extra_destinations	array	Additional destinations (if configured)
created_at	string	Stream creation timestamp (ISO 8601)
updated_at	string	Last update timestamp (ISO 8601)

Response Destination Attributes

The destination_attributes object in the response varies based on the destination type. Select a tab below to view the fields returned for each destination.

Webhook

Field	Type	Description
url	string	Webhook endpoint URL
compression	string	Compression type (none or gzip)
headers	object	Custom headers (if configured)
max_retry	integer	Maximum retry attempts
retry_interval_sec	integer	Seconds between retries
post_timeout_sec	integer	Request timeout in seconds
security_token	string	Auto-generated token for validating webhook authenticity

S3

Field	Type	Description
endpoint	string	S3 endpoint
bucket	string	Bucket name
access_key	string	S3 access key
object_prefix	string	Prefix for object keys
file_compression_type	string	Compression type (none or gzip)
file_type	string	File format (.json)
max_retry	integer	Maximum retry attempts
retry_interval_sec	integer	Seconds between retries
use_ssl	boolean	SSL enabled status

Azure

Field	Type	Description
storage_account	string	Azure storage account name
container	string	Container name
blob_prefix	string	Prefix for blob names
file_compression_type	string	Compression type (none or gzip)
file_type	string	File format (.json)
max_retry	integer	Maximum retry attempts
retry_interval_sec	integer	Seconds between retries

PostgreSQL

Field	Type	Description
host	string	Database host
port	integer	Database port
username	string	Database username
table_name	string	Target table name
sslmode	string	SSL mode (require or disable)
max_retry	integer	Maximum retry attempts
retry_interval_sec	integer	Seconds between retries

note

Sensitive fields like secret_key (S3), sas_token (Azure), and password (PostgreSQL) are not returned in the response for security reasons.