Docs / SDK guides

Streaming

GET https://farbeamlabs.com/api/live is a Server-Sent Events stream of new transfers and blocks. It sits outside /api/v1, needs no key and has no unit cost. The live transfers page uses it.

#Protocol

Item	Value
Content type	`text/event-stream`
Events	`transfer` and `block`
Filtering	None. You receive every event, and you filter on your side.
First line	`retry: 5000`, so browsers reconnect after 5 seconds
Keep-alive	A comment line, `: ping`, every 25 seconds
Limit	6 open streams per network, 400 in total
At the limit	HTTP 429, `retry-after: 30`, no body
Event ids	None. `Last-Event-ID` has no effect.
Replay	None. Fill gaps with REST.

#Events

A real transfer event holds the same fields as the transfer object, with ts as integer milliseconds.

Text

event: transfer
data: {"uid":"ethereum:0xcc96...841a:210","ts":1793821007000,"chain":"ethereum","block":26343583,"tx":"0xcc96...841a","kind":"token","token":"WETH","token_address":"0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2","amount":743.51,"usd":1800098.74,"from":"0x76f3...5b1a","to":"0x0000...594e","sampled":false,"id":3485,"from_label":null,"from_type":null,"to_label":null,"to_type":null}

event: block
data: {"chain":"base","number":53515362,"ts":1793820071000,"txs":207,"gas_used":27387633,"gas_limit":400000000,"base_fee_gwei":0.005}

block events arrive once per poll per chain. On Solana, they hold number, ts, txs and hash.

#In the browser

Use EventSource. It reconnects by itself.

JavaScript

const stream = new EventSource("/api/live");

stream.addEventListener("transfer", (message) => {
  const row = JSON.parse(message.data);
  if (row.usd >= 1000000 && !row.sampled) {
    console.log(row.chain, row.token, Math.round(row.usd), row.from_label, row.to_label);
  }
});

stream.addEventListener("block", (message) => {
  const block = JSON.parse(message.data);
  console.log("block", block.chain, block.number);
});

stream.onerror = () => console.warn("stream dropped, the browser will reconnect");

/api/live sends access-control-allow-origin: *, so EventSource also works from pages on other origins.

#In Node.js

Node has no built-in EventSource that you can rely on across versions, so this reads the stream with fetch. It reconnects, respects 429, and fills gaps.

JavaScript

// stream.mjs: run with node stream.mjs
const LIVE = "https://farbeamlabs.com/api/live";
const REST = "https://farbeamlabs.com/api/v1/transfers";
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

let lastId = 0;
const seen = new Set();

function handle(row) {
  if (seen.has(row.uid)) return;
  seen.add(row.uid);
  if (seen.size > 5000) seen.delete(seen.values().next().value);
  lastId = Math.max(lastId, row.id ?? 0);
  console.log(row.chain, row.token, Math.round(row.usd), row.uid);
}

async function fillGap() {
  if (!lastId) return;
  const missed = [];
  let before = null;
  while (true) {
    const query = new URLSearchParams({ limit: "100" });
    if (before) query.set("before", String(before));
    const res = await fetch(`${REST}?${query}`);
    if (!res.ok) return;
    const { data, meta } = await res.json();
    for (const row of data) {
      if (row.id <= lastId) { missed.reverse().forEach(handle); return; }
      missed.push(row);
    }
    if (meta.next_cursor === null) break;
    before = meta.next_cursor;
  }
  missed.reverse().forEach(handle);
}

async function readStream() {
  const res = await fetch(LIVE, { headers: { accept: "text/event-stream" } });
  if (res.status === 429) {
    await sleep(Number(res.headers.get("retry-after") || 30) * 1000);
    return;
  }
  if (!res.ok) throw new Error(`HTTP ${res.status}`);
  await fillGap();
  const decoder = new TextDecoder();
  let buffer = "";
  for await (const chunk of res.body) {
    buffer += decoder.decode(chunk, { stream: true });
    let end;
    while ((end = buffer.indexOf("\n\n")) >= 0) {
      const frame = buffer.slice(0, end);
      buffer = buffer.slice(end + 2);
      let event = "message";
      let data = "";
      for (const line of frame.split("\n")) {
        if (line.startsWith("event: ")) event = line.slice(7);
        if (line.startsWith("data: ")) data += line.slice(6);
      }
      if (event === "transfer" && data) handle(JSON.parse(data));
    }
  }
}

while (true) {
  try {
    await readStream();
  } catch (err) {
    console.error("stream error:", err.message);
  }
  await sleep(5000);
}

#In Python

This uses requests with a streaming response. A read timeout of 60 seconds will detect a dead connection, since pings arrive every 25 seconds.

Python

# stream.py: run with python3 stream.py
import json, time
import requests

LIVE = "https://farbeamlabs.com/api/live"
REST = "https://farbeamlabs.com/api/v1/transfers"

last_id = 0
seen = {}

def handle(row):
    global last_id
    if row["uid"] in seen:
        return
    seen[row["uid"]] = True
    if len(seen) > 5000:
        seen.pop(next(iter(seen)))
    last_id = max(last_id, row.get("id") or 0)
    print(row["chain"], row["token"], round(row["usd"]), row["uid"])

def fill_gap():
    if not last_id:
        return
    missed, before = [], None
    while True:
        params = {"limit": 100}
        if before:
            params["before"] = before
        res = requests.get(REST, params=params, timeout=30)
        if not res.ok:
            return
        body = res.json()
        for row in body["data"]:
            if row["id"] <= last_id:
                for item in reversed(missed):
                    handle(item)
                return
            missed.append(row)
        before = body["meta"]["next_cursor"]
        if before is None:
            break
    for item in reversed(missed):
        handle(item)

def read_stream():
    with requests.get(LIVE, stream=True, timeout=(10, 60), headers={"accept": "text/event-stream"}) as res:
        if res.status_code == 429:
            time.sleep(int(res.headers.get("retry-after", 30)))
            return
        res.raise_for_status()
        fill_gap()
        event, data = "message", ""
        for line in res.iter_lines(decode_unicode=True):
            if line == "":
                if event == "transfer" and data:
                    handle(json.loads(data))
                event, data = "message", ""
            elif line.startswith("event: "):
                event = line[7:]
            elif line.startswith("data: "):
                data += line[6:]

while True:
    try:
        read_stream()
    except requests.RequestException as err:
        print("stream error:", err)
    time.sleep(5)

#Reconnecting and filling gaps

The stream has no replay, and it holds nothing while you are gone. The samples above use this method.

Keep the highest id you have seen, and a set of recent uid values.
On reconnect, walk List large transfers from the top, stopping at your id.
Handle those rows oldest first, removing duplicates by uid.
The stream may send a row that the walk also returned, and the uid set handles that.

Do the walk after the stream opens, so no row falls between the two. REST has rate limits, so a client that reconnects often should wait longer between attempts.

Differences to note between surfaces: ts is an integer on the stream and an ISO string on REST. If you store ts, convert first.

#Limits and advice

Keep one stream per service, and fan out inside your own code. Six per network is the limit, and a stuck process that reconnects without closing can use them up.
The stream carries all rows at $100,000 or more. Expect many events per minute. Filter first, then do slow work on a queue.
Do not use it as a record of everything. When the service restarts, the stream drops, and pollers resume from the current head, so a period can be missing for good. See indexing.
For filtered delivery to a server, use webhooks.
WebSockets are Planned, and not built.

PreviousGo client NextRecipe: exchange flows