개요

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

매개변수	유형	Required	설명
이름	문자열	네	A name for your stream
네트워크	문자열	네	Blockchain network (see Supported Chains)
dataset	문자열	네	Dataset type (see Data Sources)
region	문자열	네	usa_east
dataset_batch_size	정수	네	Number of blocks to batch per delivery
elastic_batch_enabled	부울	네	Auto-adjust batch size to 1 near chain tip
destination	문자열	네	webhook, s3, azure, or postgres
destination_attributes	객체	네	Configuration for selected destination
상태	문자열	네	active or paused
filter_function	문자열	아니요	Base64-encoded JavaScript filter function
start_range	정수	아니요	Starting block number (omit for latest)
end_range	정수	아니요	Ending block number (omit for continuous)
fix_block_reorgs	정수	아니요	1 to enable reorg detection and re-streaming
keep_distance_from_tip	정수	아니요	Blocks to stay behind chain tip
알림 이메일	문자열	아니요	Email for stream termination alerts
extra_destinations	배열	아니요	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.

웹훅

매개변수	유형	Required	설명
url	문자열	네	Webhook endpoint URL
compression	문자열	네	none or gzip
max_retry	정수	네	Max retry attempts
retry_interval_sec	정수	네	Seconds between retries
post_timeout_sec	정수	네	Request timeout in seconds
headers	객체	아니요	Custom headers as key-value pairs
mtls	부울	아니요	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	문자열	아니요	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.

S3

매개변수	유형	Required	설명
endpoint	문자열	네	S3 endpoint (e.g., s3.amazonaws.com)
bucket	문자열	네	Bucket name
secret_key	문자열	네	S3 secret key
access_key	문자열	아니요	S3 access key
object_prefix	문자열	아니요	Prefix for object keys
file_compression_type	문자열	네	none or gzip
file_type	문자열	네	.json
max_retry	정수	네	Max retry attempts
retry_interval_sec	정수	네	Seconds between retries
use_ssl	부울	네	Enable SSL

Azure

매개변수	유형	Required	설명
storage_account	문자열	네	Azure storage account name
container	문자열	네	Container name
sas_token	문자열	네	Shared Access Signature token
blob_prefix	문자열	아니요	Prefix for blob names
file_compression_type	문자열	네	none or gzip
file_type	문자열	네	.json
max_retry	정수	네	Max retry attempts
retry_interval_sec	정수	네	Seconds between retries

PostgreSQL

매개변수	유형	Required	설명
host	문자열	네	Database host
port	정수	네	Database port
username	문자열	네	Database username
password	문자열	네	Database password
table_name	문자열	네	Target table name
sslmode	문자열	네	require or disable
max_retry	정수	네	Max retry attempts
retry_interval_sec	정수	네	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,
      "mtls": true
    },
    "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

자바스크립트

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

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,
                mtls=True,
                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,
  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"]}"

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

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

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

답변

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"
}

응답 필드

분야	유형	설명
id	문자열	Unique stream identifier (UUID)
이름	문자열	Stream name
네트워크	문자열	Blockchain network
dataset	문자열	Dataset type
region	문자열	Deployment region
filter_function	문자열	Base64-encoded filter function (if configured)
start_range	정수	Starting block number
end_range	정수	Ending block number (null for continuous)
dataset_batch_size	정수	Number of blocks per batch
elastic_batch_enabled	부울	Whether elastic batching is enabled
fix_block_reorgs	정수	Reorg detection setting (1 = enabled)
keep_distance_from_tip	정수	Blocks to stay behind chain tip
destination	문자열	Destination type (webhook, s3, azure, postgres)
destination_attributes	객체	Destination configuration (varies by type)
상태	문자열	Stream status (active or paused)
순서	정수	Last delivered block from the data range
current_hash	문자열	Hash of the current block being processed
알림 이메일	문자열	Email for stream termination alerts
extra_destinations	배열	Additional destinations (if configured)
created_at	문자열	Stream creation timestamp (ISO 8601)
업데이트 일시	문자열	Last update timestamp (ISO 8601)

Response Destination Attributes

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

웹훅

분야	유형	설명
url	문자열	Webhook endpoint URL
compression	문자열	Compression type (none or gzip)
headers	객체	Custom headers (if configured)
max_retry	정수	Maximum retry attempts
retry_interval_sec	정수	Seconds between retries
post_timeout_sec	정수	Request timeout in seconds
security_token	문자열	Auto-generated token for validating webhook authenticity

S3

분야	유형	설명
endpoint	문자열	S3 endpoint
bucket	문자열	Bucket name
access_key	문자열	S3 access key
object_prefix	문자열	Prefix for object keys
file_compression_type	문자열	Compression type (none or gzip)
file_type	문자열	File format (.json)
max_retry	정수	Maximum retry attempts
retry_interval_sec	정수	Seconds between retries
use_ssl	부울	SSL enabled status

Azure

분야	유형	설명
storage_account	문자열	Azure storage account name
container	문자열	Container name
blob_prefix	문자열	Prefix for blob names
file_compression_type	문자열	Compression type (none or gzip)
file_type	문자열	File format (.json)
max_retry	정수	Maximum retry attempts
retry_interval_sec	정수	Seconds between retries

PostgreSQL

분야	유형	설명
host	문자열	Database host
port	정수	Database port
username	문자열	Database username
table_name	문자열	Target table name
sslmode	문자열	SSL mode (require or disable)
max_retry	정수	Maximum retry attempts
retry_interval_sec	정수	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.