Looking for SegmentStream Marketing Analytics? It moved to measure.segmentstream.com →
segmentstream-pipeline (Python) · pipeline/

Pipelines your agent can write and you can read.

Connections, ingestion, models, checks, schedules and AI evaluation in plain Python, in the pipeline/ and connections/ folders. The platform validates the graph, runs it on managed compute and shows the lineage of every table.

Get started

pipeline/pyproject.toml
# pipeline/pyproject.tomldependencies = ["segmentstream-pipeline[bigquery]"]

At a glance

  • Python, with Ibis models compiled to warehouse SQL
  • Bronze, silver and gold tiers; daily partitions; backfills
  • Assets, checks, jobs and schedules in one definitions file
  • 24 ready-made connections, plus any API with a key, OAuth 2.0 or client credentials

What it does

01

Connections

Declare connections once; pipelines, ontology actions and the company agent share them. OAuth tokens and keys stay with the platform.

connections/connections.py
from segmentstream.connections import GoogleAdsConnection, HubSpotConnectionconnections = (    GoogleAdsConnection(key="google_ads"),    HubSpotConnection(key="hubspot"),)
02

Ingestion

Load any source by day, with backfills. Re-running a day replaces it exactly.

pipeline/ingestion/raw_orders.py
@s.ingestion(    tier=s.Tier.Bronze,    partitioning=s.Partitioning.Daily(column="order_date", start_date="2026-01-01"),    schema={"order_id": "string", "order_date": "date", "amount": "float64"},    primary_key=("order_id",),)def raw_orders(context: s.Context) -> Iterator[s.IngestionRow]:    for day in context.partition_keys:        yield from fetch_orders(day)
03

Models

Models are dataframe expressions compiled to warehouse SQL, with declared schemas and inputs resolved by name.

pipeline/models/revenue_summary.py
@s.model(    tier=s.Tier.Gold,    partitioning=s.Partitioning.Daily(column="order_date", start_date="2026-01-01"),    schema={"order_date": "date", "customer_count": "int64", "total_revenue": "float64"},)def revenue_summary(customer_revenue: ir.Table) -> ir.Table:    return customer_revenue.group_by("order_date").aggregate(        customer_count=customer_revenue.count(),        total_revenue=customer_revenue.revenue.sum(),    )
04

Checks

Checks evaluate a table and record the result next to it, where lineage and agents can see it.

pipeline/models/raw_orders_checks.py
@s.asset_check(asset=raw_orders)def positive_order_amounts(raw_orders: ir.Table, warehouse: s.WarehouseResource) -> s.CheckResult:    invalid = int(warehouse.execute(raw_orders.filter(raw_orders.amount <= 0).count()))    return s.CheckResult(passed=invalid == 0, metadata={"invalid_order_count": invalid})
05

Jobs and schedules

Jobs select assets; schedules run them from code, in a timezone, over the partitions you name.

pipeline/definitions.py
revenue = s.job("revenue_pipeline", assets=[revenue_summary], include_upstream=True)defs = s.definitions(    assets=[raw_orders, revenue_summary],    checks=[positive_order_amounts],    jobs=[revenue],    schedules=[        s.schedule(            "revenue_morning",            job=revenue,            cron="0 6 * * *",            timezone="Europe/London",            partitions=s.LatestPartitions(3),        ),    ],)
06

AI evaluation

Ask a decision model typed questions about any record and store the answers, with their probabilities, next to it.

pipeline/ingestion/lead_evaluations.py
client = s.evaluation_client("evaluation")result = client.evaluate(    state={"job_title": "Head of Learning & Development", "company": "Acme"},    questions={        "relevant": {            "type": "boolean",            "instructions": "Would sales accept this demo request as an opportunity?",        },    },)relevant = result["answers"]["relevant"]["probability"]
Inside SegmentStream

What the Pipeline SDK builds.

Ingestion and models from pipeline/, as a map of what feeds what, with the state of every table's checks.

segmentstream.com
The console's table map for web analytics: reference tables for click ids, IP ranges and agents feed hits and leads, which feed sessions, agent and rule hits, and the leads, invalid-traffic and marketing reports, each marked with the state of its checks
What feeds what, from raw hits to the reports people read, with the state of each table’s checks.

The rest of the surface

Build your company’s own software. Let agents run it.

$ segmentstream workspace create acme --region eu --subdomain acme

▸ Created workspace acme · database, warehouse and a private git repository

$ segmentstream workspace WORKSPACE-ID clone acme && cd acme

▸ your coding agent builds what your team asked for and runs the checks

$ git add -A && git commit -m "First version"

$ segmentstream deploy

▸ app, company agent, API and MCP tools live for your invited members

Start a pilot

Our engineers build the first version on your data in two weeks, at a fixed price.

Or build it yourself with your coding agent:

curl -fsSL https://segmentstream.com/install.sh | sh

We’ll email you product updates, tips and onboarding help. Unsubscribe anytime.