dltComing soon

Load the live web data SpecStream finds into any dlt destination: your lists, their records, and one row per value with the page it was read from and when. The pipeline is a single Python file built on dlt's REST API source.

SpecStream for dltData & warehouses

What you need

A SpecStream account, free to open.
Python with dlt. The example loads into DuckDB, which needs no server.
A SpecStream API key, which step 1 creates.

1
1. Create an API key
In the SpecStream workspace, go to Agents & API and choose API keys, then Create API key. The pipeline never spends, so read permission and a $0 spending limit are enough.
The key is shown once, so copy it before you move on.

2. Install dlt

Install dlt with the DuckDB extra. For BigQuery, Snowflake, Postgres, or another destination, install the extra for that destination instead.

Terminal
pip install "dlt[duckdb]"

3. Save the pipeline

lists pages through your lists, newest first, and records reads each list's rows. shape_record turns each row into a record with one cell per value.

Save the file below as specstream_pipeline.py. It's also in the dlt example (opens in a new tab) in specstream-tools.

specstream_pipeline.py
import dlt
from dlt.sources.rest_api import rest_api_source

# Workspace buttons, not facts.
BUTTON_COLUMNS = {"matches", "open"}


def plain_cell(column_id: str, cell: dict) -> dict:
    """One value with the page it was read from. A value SpecStream couldn't read is None."""
    provenance = cell.get("provenance") or {}
    return {
        "column_id": column_id,
        "label": cell.get("label") or column_id,
        "value": None if cell.get("absence") else cell.get("display"),
        "source_url": provenance.get("source_url"),
        "observed_at": provenance.get("observed_at"),
    }


def shape_record(row: dict) -> dict:
    """Turn a row of a list into a record with its list, a record ID and plain cells.

    Records SpecStream found have an `ordinal` and `cells`; rows of a list you imported have an `id` and the
    columns you `provided`; rows of a combined list have an `id` and each column's values in `field_values`,
    the value the list shows first.
    """
    if "provided" in row:
        cells = [
            {
                "column_id": key,
                "label": key,
                "value": None if value is None else str(value),
                "source_url": None,
                "observed_at": None,
            }
            for key, value in (row.get("provided") or {}).items()
        ]
    elif "field_values" in row:
        cells = [
            plain_cell(key, values[0])
            for key, values in (row.get("field_values") or {}).items()
            if values and key not in BUTTON_COLUMNS
        ]
    else:
        cells = [
            plain_cell(cell["column_id"], cell)
            for cell in row.get("cells") or []
            if cell.get("column_id") and cell["column_id"] not in BUTTON_COLUMNS
        ]
    entity_id = row.get("entity_id") or next(
        (s["row"]["entity_id"] for s in row.get("sources") or [] if (s.get("row") or {}).get("entity_id")),
        None,
    )
    return {
        "list_id": row["_lists_id"],
        "list_created_at": row["_lists_created_at"],
        "record_id": str(row["ordinal"] if row.get("ordinal") is not None else row["id"]),
        "entity_id": entity_id,
        "cells": cells,
    }


@dlt.source(name="specstream", max_table_nesting=1)
def specstream_source(api_key: str = dlt.secrets.value, start_date: str = "2020-01-01T00:00:00+00:00"):
    return rest_api_source(
        {
            "client": {
                "base_url": "https://api.specstream.ai",
                "auth": {"type": "bearer", "token": api_key},
            },
            "resources": [
                {
                    "name": "lists",
                    "primary_key": "id",
                    "write_disposition": "merge",
                    "endpoint": {
                        "path": "/v1/workspace/collections",
                        "params": {"limit": 100, "sort": "recent"},
                        "data_selector": "data.collections",
                        "paginator": {
                            "type": "cursor",
                            "cursor_path": "data.next_after",
                            "cursor_param": "after",
                        },
                        "incremental": {"cursor_path": "created_at", "initial_value": start_date},
                    },
                },
                {
                    "name": "records",
                    "primary_key": ["list_id", "record_id"],
                    "write_disposition": "merge",
                    "include_from_parent": ["id", "created_at"],
                    "endpoint": {
                        "path": "/v1/workspace/collections/{resources.lists.id}",
                        "data_selector": "data.rows",
                        "paginator": "single_page",
                    },
                    "processing_steps": [{"map": shape_record}],
                },
            ],
        }
    )


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="specstream", destination="duckdb", dataset_name="specstream_data")
    print(pipeline.run(specstream_source()))

4. Set your key and run it

dlt reads the key from the SOURCES__SPECSTREAM__API_KEY environment variable. When the run finishes, the script prints the load summary.

Terminal
export SOURCES__SPECSTREAM__API_KEY="your-api-key"
python specstream_pipeline.py

5. Query the values

Each value is a row in records__cells. Join it to its record with _dlt_parent_id, which matches the record's _dlt_id. The example's dataset is specstream_data.

SQL
select r.list_id, r.entity_id, c.label, c.value, c.source_url, c.observed_at
from specstream_data.records r
join specstream_data.records__cells c on c._dlt_parent_id = r._dlt_id

Tables

These tables land in your dataset, and each run adds to them.

Table	One row per	Columns
`lists`	List	`id`, `name`, `kind`, `created_at`, `row_count`
`records`	Record	`list_id`, `record_id`, `list_created_at`, `entity_id`
`records__cells`	Value	`column_id`, `label`, `value`, `source_url`, `observed_at`

Values and sources

entity_id is shared by every record about the same company, place, person, or posting, which lets you join between lists. A value SpecStream couldn't read loads as null.

Rows of a list you imported carry the columns you provided and no source page. A combined list loads the value it shows.

Later runs

lists is incremental on created_at, so each run loads only the lists created since the last run. Records a SpecStream schedule refreshes come in as a new list, so the next run picks them up. Both lists and records merge on a primary key, so running the pipeline again doesn't duplicate rows.

The first run starts at 2020-01-01T00:00:00+00:00. To begin later, pass start_date, for example specstream_source(start_date="2026-01-01T00:00:00+00:00"). Run the script on whatever scheduler you already use for your other pipelines.

Load somewhere else

Change destination in dlt.pipeline(...) to "bigquery", "snowflake", "postgres", or any other destination dlt supports, and add that destination's credentials the way dlt's docs (opens in a new tab) describe. The key stays the only SpecStream setting.

Cost

Running the pipeline doesn't touch your SpecStream balance. It reads lists that already exist and starts no request. The balance pays when you ask SpecStream for data, per result that matches. Prices and quotes are covered in balance, quotes, and billing.

Disconnect

Stop running the script and remove the environment variable. Revoke the key under Agents & API to end access immediately. Tables already in your destination stay as they are. Questions go to [email protected], and Connect your tools lists the other connections.

SpecStream for dlt covers what it does in dlt on one page. Every other tool has its guide in Connect your tools. You see the most a request can cost before it starts, and only the matches come out of your balance. Billing has the details. For help, email [email protected].