Websockets

Websockets provide a persistent, bidirectional connection between your application and Allium, enabling real-time data delivery with lower latency than traditional HTTP requests. Allium offers two websocket capabilities:

APIs over Websockets - Call any Allium API through a persistent websocket connection for faster response times
Data Streams over Websockets - Subscribe to real-time blockchain data streams (wraps Kafka topics)

If you need delivery guarantees and at-least-once semantics, we recommend using Kafka or Pub/Sub directly. Websockets are best for real-time monitoring where occasional message loss is acceptable.

APIs over Websockets

APIs over websockets allow you to make API requests through a persistent connection, reducing latency by eliminating the connection overhead of individual HTTP requests.

Connection

Connect to the websocket API endpoint with your API key:

import websocket
import json

API_KEY = "your_api_key_here"
WS_URL = "wss://api.allium.so/api/v1/developer/ws/api"

# Connect with X-API-KEY header
ws = websocket.create_connection(
    WS_URL,
    header={"X-API-KEY": API_KEY}
)

const WebSocket = require('ws');

const API_KEY = "your_api_key_here";
const WS_URL = "wss://api.allium.so/api/v1/developer/ws/api";

// Connect with X-API-KEY header
const ws = new WebSocket(WS_URL, {
    headers: {
        "X-API-KEY": API_KEY
    }
});

package main

import (
    "log"
    
    "github.com/gorilla/websocket"
)

func main() {
    apiKey := "your_api_key_here"
    wsURL := "wss://api.allium.so/api/v1/developer/ws/api"
    
    // Connect with X-API-KEY header
    headers := map[string][]string{
        "X-API-KEY": {apiKey},
    }
    
    conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
    if err != nil {
        log.Fatal("dial:", err)
    }
    defer conn.Close()
}

Request Format

Send requests as JSON messages with the following structure:

{
  "method": "POST",
  "id": "unique-request-id",
  "path": "/api/v1/developer/trading/hyperliquid/info/fills",
  "body": {
    "type": "userFills",
    "user": "0xD5994523d6B235862bE1790C06cFBE8547869449"
  }
}

Request Fields:

method (string, required): HTTP method - GET, POST, PUT, PATCH, or DELETE
id (string, required): Unique identifier for tracking the request/response
path (string, required): API endpoint path (e.g., /api/v1/developer/assets)
body (object, optional): Request parameters
- For GET requests: URL query parameters sent as key-value pairs in body
- For POST/PUT/PATCH/DELETE: Body parameters sent as key-value pairs in body

Response Format

Responses are returned as JSON messages:

{
  "id": "unique-request-id",
  "status": "success",
  "data": { ... }
}

Response Fields:

id (string): Matches the request ID
status (string): Either "success" or "error"
data (object): Response data or error message

Example: POST Request with Body Parameters

import websocket
import json
from datetime import datetime

API_KEY = "your_api_key_here"
WS_URL = "wss://api.allium.so/api/v1/developer/ws/api"

# Connect
ws = websocket.create_connection(
    WS_URL,
    header={"X-API-KEY": API_KEY}
)

# Prepare POST request with body parameters
request = {
    "method": "POST",
    "id": f"request-{datetime.now().timestamp()}",
    "path": "/api/v1/developer/trading/hyperliquid/info/fills",
    "body": {
        "type": "userFills",
        "user": "0xD5994523d6B235862bE1790C06cFBE8547869449"
    }
}

# Send request
ws.send(json.dumps(request))

# Receive response
response = ws.recv()
response_json = json.loads(response)

print(response_json)

ws.close()

const WebSocket = require('ws');

const API_KEY = "your_api_key_here";
const WS_URL = "wss://api.allium.so/api/v1/developer/ws/api";

// Connect
const ws = new WebSocket(WS_URL, {
    headers: {
        "X-API-KEY": API_KEY
    }
});

ws.on('open', () => {
    // Prepare POST request with body parameters
    const request = {
        method: "POST",
        id: `request-${Date.now()}`,
        path: "/api/v1/developer/trading/hyperliquid/info/fills",
        body: {
            type: "userFills",
            user: "0xD5994523d6B235862bE1790C06cFBE8547869449"
        }
    };
    
    // Send request
    ws.send(JSON.stringify(request));
});

ws.on('message', (data) => {
    const response = JSON.parse(data);
    console.log(response);
    ws.close();
});

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "time"
    
    "github.com/gorilla/websocket"
)

type Request struct {
    Method string                 `json:"method"`
    ID     string                 `json:"id"`
    Path   string                 `json:"path"`
    Body   map[string]interface{} `json:"body"`
}

type Response struct {
    ID     string          `json:"id"`
    Status string          `json:"status"`
    Data   json.RawMessage `json:"data"`
}

func main() {
    apiKey := "your_api_key_here"
    wsURL := "wss://api.allium.so/api/v1/developer/ws/api"
    
    // Connect
    headers := map[string][]string{
        "X-API-KEY": {apiKey},
    }
    
    conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
    if err != nil {
        log.Fatal("dial:", err)
    }
    defer conn.Close()
    
    // Prepare POST request with body parameters
    request := Request{
        Method: "POST",
        ID:     fmt.Sprintf("request-%d", time.Now().Unix()),
        Path:   "/api/v1/developer/trading/hyperliquid/info/fills",
        Body: map[string]interface{}{
            "type": "userFills",
            "user": "0xD5994523d6B235862bE1790C06cFBE8547869449",
        },
    }
    
    // Send request
    err = conn.WriteJSON(request)
    if err != nil {
        log.Fatal("write:", err)
    }
    
    // Receive response
    var response Response
    err = conn.ReadJSON(&response)
    if err != nil {
        log.Fatal("read:", err)
    }
    
    fmt.Printf("%+v\n", response)
}

Example: GET Request with Query Parameters

# GET request - parameters in body become URL query parameters
request = {
    "method": "GET",
    "id": "get-assets-1",
    "path": "/api/v1/developer/assets",
    "body": {
        "blockchain": "ethereum",
        "limit": "10"
    }
}

ws.send(json.dumps(request))
response = ws.recv()
response_json = json.loads(response)

// GET request - parameters in body become URL query parameters
const request = {
    method: "GET",
    id: "get-assets-1",
    path: "/api/v1/developer/assets",
    body: {
        blockchain: "ethereum",
        limit: "10"
    }
};

ws.send(JSON.stringify(request));

ws.on('message', (data) => {
    const response = JSON.parse(data);
    console.log(response);
});

// GET request - parameters in body become URL query parameters
request := Request{
    Method: "GET",
    ID:     "get-assets-1",
    Path:   "/api/v1/developer/assets",
    Body: map[string]interface{}{
        "blockchain": "ethereum",
        "limit":      "10",
    },
}

err = conn.WriteJSON(request)
if err != nil {
    log.Fatal("write:", err)
}

var response Response
err = conn.ReadJSON(&response)
if err != nil {
    log.Fatal("read:", err)
}

fmt.Printf("%+v\n", response)

Concurrency & Rate Limits

Maximum 1,000 concurrent requests in flight per connection
However even with the above your account rate limits still apply, so you might not be able to reach 1,000 concurrent requests in flight per connection.
If you try to exceed this limit, you’ll receive an error response:

{
  "id": "your-request-id",
  "status": "error",
  "data": "max inflight requests reached, please wait for requests to finish"
}

Benefits

Lower latency: No connection overhead for each request
Request/response matching: Use unique IDs to track async responses

Data Streams over Websockets

Stream real-time blockchain data directly to your application. All Kafka topics are available via websockets for easier integration.

Connection

Connect to a data stream by specifying a topic:

from websocket import WebSocketApp

API_KEY = "your_api_key_here"
WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"

ws = WebSocketApp(
    WS_URL,
    header={"X-API-KEY": API_KEY},
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

ws.run_forever()

const WebSocket = require('ws');

const API_KEY = "your_api_key_here";
const WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks";

const ws = new WebSocket(WS_URL, {
    headers: {
        "X-API-KEY": API_KEY
    }
});

ws.on('open', onOpen);
ws.on('message', onMessage);
ws.on('error', onError);
ws.on('close', onClose);

package main

import (
    "log"
    
    "github.com/gorilla/websocket"
)

func main() {
    apiKey := "your_api_key_here"
    wsURL := "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"
    
    headers := map[string][]string{
        "X-API-KEY": {apiKey},
    }
    
    conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
    if err != nil {
        log.Fatal("dial:", err)
    }
    defer conn.Close()
    
    // Handle connection events
    go handleMessages(conn)
}

Available Topics

All Kafka topics are available. Use the format: {blockchain}.{data_type} Examples:

ethereum.blocks
ethereum.transactions
ethereum.logs
base.dex_trades
solana.transactions
hyperliquid.fills
bitcoin.blocks

See the Kafka documentation for the complete list of available topics and schemas.

Starting the Stream

After connecting, send a start message to begin receiving data:

def on_open(ws):
    print("Connected to WebSocket")
    
    # Start receiving messages
    start_message = {"action": "start"}
    ws.send(json.dumps(start_message))

function onOpen() {
    console.log("Connected to WebSocket");
    
    // Start receiving messages
    const startMessage = {action: "start"};
    ws.send(JSON.stringify(startMessage));
}

func onOpen(conn *websocket.Conn) {
    fmt.Println("Connected to WebSocket")
    
    // Start receiving messages
    startMessage := map[string]string{"action": "start"}
    err := conn.WriteJSON(startMessage)
    if err != nil {
        log.Println("write start:", err)
    }
}

Stopping the Stream

You can pause the stream without disconnecting by sending a stop message:

# Stop receiving messages (connection remains open)
stop_message = {"action": "stop"}
ws.send(json.dumps(stop_message))

# Resume later by sending start again
start_message = {"action": "start"}
ws.send(json.dumps(start_message))

// Stop receiving messages (connection remains open)
const stopMessage = {action: "stop"};
ws.send(JSON.stringify(stopMessage));

// Resume later by sending start again
const startMessage = {action: "start"};
ws.send(JSON.stringify(startMessage));

// Stop receiving messages (connection remains open)
stopMessage := map[string]string{"action": "stop"}
conn.WriteJSON(stopMessage)

// Resume later by sending start again
startMessage := map[string]string{"action": "start"}
conn.WriteJSON(startMessage)

This is useful when you need to temporarily pause data delivery, update filters, or reduce load without closing the websocket connection.

Filtering Data (Optional)

You can filter the stream to only receive messages matching specific criteria. Send a filter with the setFilter action before starting the stream:

def on_open(ws):
    # Set a filter before starting
    filter_message = {
        "action": "setFilter",
        "data": {
            "field": "to_address",
            "operator": "exists",
            "value": True
        }
    }
    ws.send(json.dumps(filter_message))
    
    # Start the stream
    start_message = {"action": "start"}
    ws.send(json.dumps(start_message))

function onOpen() {
    // Set a filter before starting
    const filterMessage = {
        action: "setFilter",
        data: {
            field: "to_address",
            operator: "exists",
            value: true
        }
    };
    ws.send(JSON.stringify(filterMessage));
    
    // Start the stream
    const startMessage = {action: "start"};
    ws.send(JSON.stringify(startMessage));
}

func onOpen(conn *websocket.Conn) {
    // Set a filter before starting
    filterMessage := map[string]interface{}{
        "action": "setFilter",
        "data": map[string]interface{}{
            "field":    "to_address",
            "operator": "exists",
            "value":    true,
        },
    }
    conn.WriteJSON(filterMessage)
    
    // Start the stream
    startMessage := map[string]string{"action": "start"}
    conn.WriteJSON(startMessage)
}

The data field contains the filter definition. Filters support comparison operators (=, !=, >, >=, <, <=), the in operator for arrays, the exists operator, and compound AND/OR logic. Example with compound filter:

# Filter for large Ethereum transactions
filter_message = {
    "action": "setFilter",
    "data": {
        "op": "AND",
        "conditions": [
            {
                "field": "blockchain",
                "operator": "=",
                "value": "ethereum"
            },
            {
                "field": "value",
                "operator": ">",
                "value": "1000000000000000000"
            }
        ]
    }
}
ws.send(json.dumps(filter_message))

// Filter for large Ethereum transactions
const filterMessage = {
    action: "setFilter",
    data: {
        op: "AND",
        conditions: [
            {
                field: "blockchain",
                operator: "=",
                value: "ethereum"
            },
            {
                field: "value",
                operator: ">",
                value: "1000000000000000000"
            }
        ]
    }
};
ws.send(JSON.stringify(filterMessage));

// Filter for large Ethereum transactions
filterMessage := map[string]interface{}{
    "action": "setFilter",
    "data": map[string]interface{}{
        "op": "AND",
        "conditions": []map[string]interface{}{
            {
                "field":    "blockchain",
                "operator": "=",
                "value":    "ethereum",
            },
            {
                "field":    "value",
                "operator": ">",
                "value":    "1000000000000000000",
            },
        },
    },
}
conn.WriteJSON(filterMessage)

See the Filter Syntax documentation for complete details on all operators, nested conditions, and real-world examples.

Receiving Messages

Messages are delivered as JSON:

def on_message(ws, message):
    data = json.loads(message)
    print(f"Received: {data}")
    
    # Process your blockchain data
    process_data(data)

function onMessage(data) {
    const message = JSON.parse(data);
    console.log("Received:", message);
    
    // Process your blockchain data
    processData(message);
}

func onMessage(conn *websocket.Conn) {
    var data map[string]interface{}
    err := conn.ReadJSON(&data)
    if err != nil {
        log.Println("read:", err)
        return
    }
    
    fmt.Println("Received:", data)
    
    // Process your blockchain data
    processData(data)
}

Complete Example: Streaming Ethereum Blocks

from websocket import WebSocketApp
import json

API_KEY = "your_api_key_here"
WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"

def on_open(ws):
    print("✅ Connected to stream")
    start_message = {"action": "start"}
    ws.send(json.dumps(start_message))

def on_message(ws, message):
    data = json.loads(message)
    block_number = data.get('number')
    print(f"New block: {block_number}")

def on_error(ws, error):
    print(f"❌ Error: {error}")

def on_close(ws, close_status_code, close_msg):
    print(f"🔌 Disconnected: {close_msg}")

ws = WebSocketApp(
    WS_URL,
    header={"X-API-KEY": API_KEY},
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

ws.run_forever()

const WebSocket = require('ws');

const API_KEY = "your_api_key_here";
const WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks";

const ws = new WebSocket(WS_URL, {
    headers: {
        "X-API-KEY": API_KEY
    }
});

ws.on('open', () => {
    console.log("✅ Connected to stream");
    const startMessage = {action: "start"};
    ws.send(JSON.stringify(startMessage));
});

ws.on('message', (message) => {
    const data = JSON.parse(message);
    const blockNumber = data.number;
    console.log(`New block: ${blockNumber}`);
});

ws.on('error', (error) => {
    console.log(`❌ Error: ${error}`);
});

ws.on('close', (code, reason) => {
    console.log(`🔌 Disconnected: ${reason}`);
});

package main

import (
    "encoding/json"
    "fmt"
    "log"
    
    "github.com/gorilla/websocket"
)

func main() {
    apiKey := "your_api_key_here"
    wsURL := "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"
    
    headers := map[string][]string{
        "X-API-KEY": {apiKey},
    }
    
    conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
    if err != nil {
        log.Fatal("dial:", err)
    }
    defer conn.Close()
    
    // Send start message
    fmt.Println("✅ Connected to stream")
    startMessage := map[string]string{"action": "start"}
    err = conn.WriteJSON(startMessage)
    if err != nil {
        log.Fatal("write start:", err)
    }
    
    // Read messages
    for {
        var data map[string]interface{}
        err := conn.ReadJSON(&data)
        if err != nil {
            log.Println("❌ Error:", err)
            break
        }
        
        blockNumber := data["number"]
        fmt.Printf("New block: %v\n", blockNumber)
    }
    
    fmt.Println("🔌 Disconnected")
}

Connection Management

Compression

Websocket compression is supported and automatically negotiated:

# Python websocket-client handles compression automatically
ws = websocket.create_connection(
    WS_URL,
    header={
        "X-API-KEY": API_KEY,
        "Sec-WebSocket-Extensions": "permessage-deflate"
    }
)

// Node.js ws library handles compression automatically
const ws = new WebSocket(WS_URL, {
    headers: {
        "X-API-KEY": API_KEY,
        "Sec-WebSocket-Extensions": "permessage-deflate"
    },
    perMessageDeflate: true
});

// Gorilla websocket handles compression automatically
conn, _, err := websocket.DefaultDialer.Dial(wsURL, map[string][]string{
    "X-API-KEY": {apiKey},
    "Sec-WebSocket-Extensions": {"permessage-deflate"},
})

Reconnection

Implement reconnection logic for production applications:

import time

def connect_with_retry(url, headers, max_retries=5):
    retries = 0
    while retries < max_retries:
        try:
            ws = websocket.create_connection(url, header=headers)
            print("Connected successfully")
            return ws
        except Exception as e:
            retries += 1
            wait_time = min(2 ** retries, 30)
            print(f"Connection failed. Retrying in {wait_time}s...")
            time.sleep(wait_time)
    
    raise Exception("Failed to connect after max retries")

function connectWithRetry(url, headers, maxRetries = 5) {
    let retries = 0;
    
    function connect() {
        const ws = new WebSocket(url, {headers});
        
        ws.on('open', () => {
            console.log("Connected successfully");
            retries = 0;
        });
        
        ws.on('error', (error) => {
            console.log(`Connection failed: ${error}`);
        });
        
        ws.on('close', () => {
            if (retries < maxRetries) {
                const waitTime = Math.min(2 ** retries, 30) * 1000;
                console.log(`Retrying in ${waitTime/1000}s...`);
                setTimeout(connect, waitTime);
                retries++;
            } else {
                console.log("Failed to connect after max retries");
            }
        });
        
        return ws;
    }
    
    return connect();
}

func connectWithRetry(url string, headers map[string][]string, maxRetries int) (*websocket.Conn, error) {
    var conn *websocket.Conn
    var err error
    
    for retries := 0; retries < maxRetries; retries++ {
        conn, _, err = websocket.DefaultDialer.Dial(url, headers)
        if err == nil {
            fmt.Println("Connected successfully")
            return conn, nil
        }
        
        waitTime := time.Duration(math.Min(math.Pow(2, float64(retries)), 30)) * time.Second
        fmt.Printf("Connection failed. Retrying in %v...\n", waitTime)
        time.Sleep(waitTime)
    }
    
    return nil, fmt.Errorf("failed to connect after max retries")
}

Error Handling

Handle common websocket errors:

try:
    ws.send(json.dumps(request))
    response = ws.recv()
except websocket.WebSocketTimeoutException:
    print("Request timeout")
except websocket.WebSocketConnectionClosedException:
    print("Connection closed unexpectedly")
    # Implement reconnection
except Exception as e:
    print(f"Error: {e}")

ws.on('error', (error) => {
    console.log(`Error: ${error}`);
    // Implement reconnection logic
});

ws.on('close', (code, reason) => {
    if (code !== 1000) {
        console.log(`Connection closed unexpectedly: ${code} - ${reason}`);
        // Implement reconnection logic
    }
});

// Timeout handling
const timeout = setTimeout(() => {
    console.log("Request timeout");
    ws.close();
}, 30000);

ws.on('message', (data) => {
    clearTimeout(timeout);
    // Process message
});

// Read with timeout
conn.SetReadDeadline(time.Now().Add(30 * time.Second))

var response Response
err := conn.ReadJSON(&response)
if err != nil {
    if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
        log.Println("Connection closed normally")
    } else if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
        log.Println("Request timeout")
    } else {
        log.Println("Connection closed unexpectedly:", err)
        // Implement reconnection logic
    }
}

Authentication

Both websocket types require an API key passed in the connection headers:

header = {"X-API-KEY": "your_api_key_here"}

const headers = {
    "X-API-KEY": "your_api_key_here"
};

headers := map[string][]string{
    "X-API-KEY": {"your_api_key_here"},
}

APIs over Websockets: Requires it to be enabled on your account
Streams over Websockets: Requires it to be enabled and topic access

Contact account representative to enable websocket features on your account.

Next Steps

Filter Syntax

Learn how to filter streams with complex conditions

Kafka Streams

Production-grade streaming with delivery guarantees

API Reference

Explore available API endpoints

Use Cases

Allium App

Allium Datashares

Allium Datastreams

Allium Beam

APIs over Websockets

Connection

Request Format

Response Format

Example: POST Request with Body Parameters

Example: GET Request with Query Parameters

Concurrency & Rate Limits

Benefits

Data Streams over Websockets

Connection

Available Topics

Starting the Stream

Stopping the Stream

Filtering Data (Optional)

Receiving Messages

Complete Example: Streaming Ethereum Blocks

Connection Management

Compression

Reconnection

Error Handling

Authentication

Next Steps

Filter Syntax

Kafka Streams

API Reference

​APIs over Websockets

​Connection

​Request Format

​Response Format

​Example: POST Request with Body Parameters

​Example: GET Request with Query Parameters

​Concurrency & Rate Limits

​Benefits

​Data Streams over Websockets

​Connection

​Available Topics

​Starting the Stream

​Stopping the Stream

​Filtering Data (Optional)

​Receiving Messages

​Complete Example: Streaming Ethereum Blocks

​Connection Management

​Compression

​Reconnection

​Error Handling

​Authentication

​Next Steps

Filter Syntax

Kafka Streams

API Reference

APIs over Websockets

Connection

Request Format

Response Format

Example: POST Request with Body Parameters

Example: GET Request with Query Parameters

Concurrency & Rate Limits

Benefits

Data Streams over Websockets

Connection

Available Topics

Starting the Stream

Stopping the Stream

Filtering Data (Optional)

Receiving Messages

Complete Example: Streaming Ethereum Blocks

Connection Management

Compression

Reconnection

Error Handling

Authentication

Next Steps