Create a Stream – Streams REST API

Présentation générale

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

Paramètres de requête

Paramètre	Type	Obligatoire	Description
nom	chaîne de caractères	Oui	A name for your stream
network	chaîne de caractères	Oui	Blockchain network (see Supported Chains)
dataset	chaîne de caractères	Oui	Dataset type (see Data Sources)
region	chaîne de caractères	Oui	usa_east
dataset_batch_size	entier	Oui	Number of blocks to batch per delivery
elastic_batch_enabled	booléen	Oui	Auto-adjust batch size to 1 near chain tip
destination	chaîne de caractères	Oui	webhook, s3, azure, or postgres
destination_attributes	objet	Oui	Configuration for selected destination
statut	chaîne de caractères	Oui	active or paused
filter_function	chaîne de caractères	Non	Base64-encoded JavaScript filter function
start_range	entier	Non	Starting block number (omit for latest)
end_range	entier	Non	Ending block number (omit for continuous)
fix_block_reorgs	entier	Non	1 to enable reorg detection and re-streaming
keep_distance_from_tip	entier	Non	Blocks to stay behind chain tip
notification_email	chaîne de caractères	Non	Email for stream termination alerts
extra_destinations	tableau	Non	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
S3
Azure
PostgreSQL

Paramètre	Type	Obligatoire	Description
url	chaîne de caractères	Oui	Webhook endpoint URL
compression	chaîne de caractères	Oui	none or gzip
max_retry	entier	Oui	Max retry attempts
retry_interval_sec	entier	Oui	Seconds between retries
post_timeout_sec	entier	Oui	Request timeout in seconds
headers	objet	Non	Custom headers as key-value pairs
mtls	booléen	Non	Enable mutual TLS: Quicknode authenticates to your endpoint with its client certificate. Requires an HTTPS URL. See the Webhook destination page for server setup.
ssl_ca_pem	chaîne de caractères	Non	CA certificate (PEM) used to verify your endpoint's server certificate when it is signed by a private CA instead of a public one

For mTLS server configuration and how to download the Quicknode CA certificate, see Mutual TLS on the Webhook destination page.

Paramètre	Type	Obligatoire	Description
endpoint	chaîne de caractères	Oui	S3 endpoint (e.g., s3.amazonaws.com)
bucket	chaîne de caractères	Oui	Bucket name
secret_key	chaîne de caractères	Oui	S3 secret key
access_key	chaîne de caractères	Non	S3 access key
object_prefix	chaîne de caractères	Non	Prefix for object keys
file_compression_type	chaîne de caractères	Oui	none or gzip
file_type	chaîne de caractères	Oui	.json
max_retry	entier	Oui	Max retry attempts
retry_interval_sec	entier	Oui	Seconds between retries
use_ssl	booléen	Oui	Enable SSL

Paramètre	Type	Obligatoire	Description
storage_account	chaîne de caractères	Oui	Azure storage account name
container	chaîne de caractères	Oui	Container name
sas_token	chaîne de caractères	Oui	Shared Access Signature token
blob_prefix	chaîne de caractères	Non	Prefix for blob names
file_compression_type	chaîne de caractères	Oui	none or gzip
file_type	chaîne de caractères	Oui	.json
max_retry	entier	Oui	Max retry attempts
retry_interval_sec	entier	Oui	Seconds between retries

Paramètre	Type	Obligatoire	Description
host	chaîne de caractères	Oui	Database host
port	entier	Oui	Database port
username	chaîne de caractères	Oui	Database username
password	chaîne de caractères	Oui	Database password
table_name	chaîne de caractères	Oui	Target table name
sslmode	chaîne de caractères	Oui	require or disable
max_retry	entier	Oui	Max retry attempts
retry_interval_sec	entier	Oui	Seconds between retries

Example Request

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,
      "mtls": true
    },
    "status": "active"
  }'

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

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,
    mtls: true
  },
  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))

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,
        "mtls": true
    },
    "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())

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,
    "mtls" => true
  },
  "status" => "active"
}.to_json

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

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)

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,
                mtls=True,
                compression="none",
            )
        ),
    )

    print(stream.id, stream.status)

asyncio.run(main())

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,
  mtls: true,
  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"]}"

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,
            mtls: true,
            security_token: None,
        }))
        .build();

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

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

Réponse

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,
    "mtls": true,
    "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"
}

Champs de réponse

Champ	Type	Description
id	chaîne de caractères	Unique stream identifier (UUID)
nom	chaîne de caractères	Stream name
network	chaîne de caractères	Blockchain network
dataset	chaîne de caractères	Dataset type
region	chaîne de caractères	Deployment region
filter_function	chaîne de caractères	Base64-encoded filter function (if configured)
start_range	entier	Starting block number
end_range	entier	Ending block number (null for continuous)
dataset_batch_size	entier	Number of blocks per batch
elastic_batch_enabled	booléen	Whether elastic batching is enabled
fix_block_reorgs	entier	Reorg detection setting (1 = enabled)
keep_distance_from_tip	entier	Blocks to stay behind chain tip
destination	chaîne de caractères	Destination type (webhook, s3, azure, postgres)
destination_attributes	objet	Destination configuration (varies by type)
statut	chaîne de caractères	Stream status (active or paused)
séquence	entier	Last delivered block from the data range
current_hash	chaîne de caractères	Hash of the current block being processed
notification_email	chaîne de caractères	Email for stream termination alerts
extra_destinations	tableau	Additional destinations (if configured)
created_at	chaîne de caractères	Stream creation timestamp (ISO 8601)
date_de_mise_à_jour	chaîne de caractères	Last update timestamp (ISO 8601)

Response Destination Attributes

Le 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
S3
Azure
PostgreSQL

Champ	Type	Description
url	chaîne de caractères	Webhook endpoint URL
compression	chaîne de caractères	Compression type (none or gzip)
headers	objet	Custom headers (if configured)
max_retry	entier	Maximum retry attempts
retry_interval_sec	entier	Seconds between retries
post_timeout_sec	entier	Request timeout in seconds
security_token	chaîne de caractères	Auto-generated token for validating webhook authenticity

Champ	Type	Description
endpoint	chaîne de caractères	S3 endpoint
bucket	chaîne de caractères	Bucket name
access_key	chaîne de caractères	S3 access key
object_prefix	chaîne de caractères	Prefix for object keys
file_compression_type	chaîne de caractères	Compression type (none or gzip)
file_type	chaîne de caractères	File format (.json)
max_retry	entier	Maximum retry attempts
retry_interval_sec	entier	Seconds between retries
use_ssl	booléen	SSL enabled status

Champ	Type	Description
storage_account	chaîne de caractères	Azure storage account name
container	chaîne de caractères	Container name
blob_prefix	chaîne de caractères	Prefix for blob names
file_compression_type	chaîne de caractères	Compression type (none or gzip)
file_type	chaîne de caractères	File format (.json)
max_retry	entier	Maximum retry attempts
retry_interval_sec	entier	Seconds between retries

Champ	Type	Description
host	chaîne de caractères	Database host
port	entier	Database port
username	chaîne de caractères	Database username
table_name	chaîne de caractères	Target table name
sslmode	chaîne de caractères	SSL mode (require or disable)
max_retry	entier	Maximum retry attempts
retry_interval_sec	entier	Seconds between retries

remarque

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

Create a Stream – Streams REST API

Présentation générale​

Endpoint​

Paramètres de requête​

Destination Attributes​

Example Request​

Réponse​

Champs de réponse​

Response Destination Attributes​