Build lineage-aware canonical models with agents
16,000+ companies already ingest data with dlt in production; dltHub Transformations picks up right where that pipeline left off, drafting the taxonomy, ontology, and CDM your business runs on from the schemas and data you loaded with dlt.
dltHub Transformations
Your analytics engineer co-pilot. Paste this prompt into Claude/Codex/Cursor:
Run uvx dlthub-start then /annotate-sources to build my first canonical data model from data I've already loaded with dlt
"dltHub and Snowflake deliver a simple, end-to-end pathway for financial institutions to transform raw data into governed analytics and AI-ready datasets without needing a full engineering team. Whether you're an engineer stepping in to answer business questions, an analyst building your own pipelines, or someone fluent in AI with basic Python skills, you can pull data from core banking systems, market feeds, and APIs directly into Snowflake. You deliver outcomes that once required specialised data engineering resources."
Suraj Rajan
Field CTO, Financial Services, Snowflake
Suraj Rajan
Field CTO, Financial Services, Snowflake
THE AGENTIC ANALYTICS ENGINEERING LIFECYCLE ON DLTHUB
From raw tables to business models your agents can reason over
Prototype, explore and validate your data against DuckDB on your laptop with marimo notebooks. Same transformation code ships to production when you're ready.
11 · TAXONOMY
Collapse source tables into the business concepts they represent
The toolkit reads schemas from pipelines you've already run. It finds raw tables that represent the same entity in your business: a HubSpot contact, a Luma guest, and an event attendee all become one Person. Then it proposes the natural key that links their records together, like an email address or external ID.
Outcome: annotated source schemas + taxonomy.json.
uv run dlthub ai toolkit install transformations
Opus 5.0 · dltHub · ~/agent-observability
import dlt
dataset = dlt.attach(pipeline_name="hubspot").dataset()
columns = dataset["contacts"].columns2 · ONTOLOGY
Resolve conflicts, then connect the concepts
Once Person spans two sources, the toolkit surfaces a question a script can't answer alone: when HubSpot and Luma both have a record for the same person and their data conflicts, which one wins — and do people who exist in only one source still make the cut? Differently-named attributes for the same thing (phone vs. phone_number) get unified into one canonical name, and relationships get drawn from natural-key matches and from foreign keys already in the schema. Anything a use case needs but no source table can supply is flagged as a gap, not silently dropped.
Outcome: ontology.ison — tabular Graph ISON, not JSON — plus a human-readable ontology.md for review.
uv run dlthub ai toolkit install transformationsOpus 5.0 · dltHub · ~/agent-observability
dataset = dlt.attach(pipeline_name="gtm_pipeline").dataset()
contacts = dataset["contacts"].to_ibis().select("email")
new_leads = dataset["leads"].to_ibis().select("email")
overlap = contacts.join(new_leads, "email").distinct().count().execute()
print(f"{overlap} people appear in both sources — stitched by email, HubSpot wins on conflict")3 · CDM
Build dimensional models
Transactional events become fact tables, stable business objects (Person, Company, Product) become dimension tables shared across facts. Agents draft the model, you fix it, before a single line of transformation code gets written. Each proposed table is grounded against real rows first, locally in DuckDB, so mistakes surface in development not production.
Outcome: CDM.dbml.
uv run dlthub ai toolkit install transformationsOpus 5.0 · dltHub · ~/agent-observability
dataset = dlt.attach(pipeline_name="gtm_pipeline").dataset()
stage_history = dataset["hubspot__deal_stage_history"].to_ibis().select("deal_id", "stage")
grain = stage_history.group_by(["deal_id", "stage"]).agg(n=stage_history.count())
violations = grain.filter(grain.n > 1).count().execute()
print(f"grain violated — {violations} deals logged the same stage twice" if violations else "grain confirmed")4 · TRANSFORMATIONS
Write one function per CDM entity
Each CDM table becomes one typed, testable function, and the toolkit figures out the dependency order for you.
Write it as Ibis, which compiles to SQL and runs natively inside your destination — no data pulled into a Python process, no separate compute layer to size or pay for. The warehouse's own query planner picks the join order and pushes filters down, the same as it would for a hand-written query.
Test it against local DuckDB, then promote it to production as-is. Outcome: @dlt.hub.transformation functions in transformations/<dataset>_to_cdm.py.
uv run dlthub ai toolkit install transformationsOpus 5.0 · dltHub · ~/agent-observability
@dlt.hub.transformation(write_disposition="replace")
def customer_orders(dataset: dlt.Dataset) -> typing.Iterator[ir.Table]:
orders = dataset.table("orders").to_ibis()
yield orders.group_by("customer_id").aggregate(
first_order=orders.ordered_at.min(),
most_recent_order=orders.ordered_at.max(),
number_of_orders=orders.id.count(),
)
@dlt.source
def customers_metrics(raw_dataset: dlt.Dataset) -> list:
return [customer_orders(raw_dataset)]5 · INCREMENTAL
Every dlt resource decorator argument, for free
@dlt.hub.transformation is a dlt.resource under the hood, so it inherits the same decorator surface you already use on extraction: primary_key and merge_key for deduping, columns for hints and PII tags, table_name to fan one function out to many tables, parallelized to shard a big rebuild across workers. The first run replaces each table wholesale — there's no prior state to diff against. Once the transformation is on a schedule, write_disposition switches to merge or append and takes an incremental cursor: a date column you own, or _dlt_loads.inserted_at for whatever landed since last time. dlt joins _dlt_loads for you.
Outcome: the same function, same decorator, now touching only new or changed rows.
uv run dlthub ai toolkit install transformationsOpus 5.0 · dltHub · ~/agent-observability
6 · VERIFY
Portable across destinations, lineage-aware end to end
Lineage isn't a bolted-on feature — the transformation runs in the same connector, same runtime, that did the ingestion, so column-level hints (PII tags, types, precision) and the _dlt_load_id propagate automatically through every join. "Which ingest produced this row three tables downstream" is a column lookup, not a metadata-service call.
Before a transformation moves from a local DuckDB dev loop to production, a static checker reads the SQL directly and reports, function by function, whether it survives the dialect change — OK, WARN, or ERROR — so dialect drift surfaces at review time, not after deploy.
uv run dlthub ai toolkit install transformationsOpus 5.0 · dltHub · ~/agent-observability
8 of 8
Every check and chart still traces back to the load that produced it
The lifecycle closes where operations begin. The same CDM tables plug straight into dq.checks — is_unique, is_not_null, is_in — for continuous validation: no new connection, because it's the same dataset object the transformation just wrote to. If a check fails, _dlt_load_id says exactly which load produced the row.
| Run | Started | Duration | Rows | Status |
|---|---|---|---|---|
| #4128 | 13:42 · today | 4.21s | 2,214 | Running |
| #4127 | 13:12 · today | 4.04s | 2,189 | Success |
| #4126 | 12:42 · today | 4.17s | 2,202 | Success |
| #4125 | 12:12 · today | 4.31s | 2,176 | Success |
| #4124 | 11:42 · today | 42.5s | 0 | Failed |
| #4123 | 11:12 · today | 4.08s | 2,164 | Success |
import dlt
from dlt.hub import data_quality as dq
pipeline = dlt.attach(pipeline_name="person_interactions_to_cdm")
dataset = pipeline.dataset() # same dataset object the transformation just wrote to
checks = {
"fact_event_attendance": [
dq.checks.is_unique("attendance_sk"),
dq.checks.is_not_null("person_sk"),
dq.checks.is_in("status", ["registered", "attended", "no_show"]),
]
}
dq.CheckSuite(dataset, checks=checks).checksComplete agentic workflows for every phase of data engineering
Not autocomplete, not a chatbot on a dashboard. A guided sequence of skills, commands, rules, and MCP - with guardrails agents can't skip. Maintained by dltHub, controlling the infrastructure agents and pipelines operate on.
Discover individual skills per agentic workflow
See how each workflow guides your agent - step by step, from first prompt to production deployment.
The guided entry point. Names a use case, checks the workspace, and hands off to the right toolkit in a few prompts.
Opus 5.0 · Quick Start · ~/pipelines
Take me through the full workflow with the GitHub API
The guided entry point. Names a use case, checks the workspace, and hands off to the right toolkit in a few prompts.
Opus 5.0 · Quick Start · ~/pipelines
Take me through the full workflow with the GitHub API
Blueprints for data workflows
dltHub is a composable data platform. Blueprints are its ready-made builds: each one dltHub assembled for a specific use case, end to end, from the sources you already use to a production dashboard or API.
Your coding agent
Claude Code, Cursor or Codex
the agentic layerdltHub AI harness
Agentic primitives to build, run and fix pipelines
dltHub context catalog
Lineage, schema, data quality, governance, run state
Pydantic Logfire
Arize
Langfuse
LangChain
the managed infrastructure layerIngest and standardize traces into the OpenAI messages format as a training-ready dataset.
OpenAI
distil labs
Fine-tune a specialist model, served as a drop-in replacement via API to distil labs customers.
Frequently Asked Questions
What is dlt?
dlt (data load tool) is an open-source Python library for building data pipelines. It handles schema inference, incremental loading, nested data normalization, and works with 10,100+ sources. Apache 2.0 licensed and always free to use.
What is dltHub?
dltHub is the managed agentic platform for running dlt pipelines in production. It bundles a managed runtime (deploy with one command, no infra to patch), Python and SQL transformations orchestrated inside your pipeline, data quality checks that fail fast with actionable errors, a managed Iceberg lakehouse with the option to bring your own storage, and an MCP server so agents can analyze pipelines and datasets directly. The outcome: teams ship trustworthy data faster, without owning the infrastructure. See the full feature list in the dltHub docs.
How is dltHub different from a Claude skill or tools like Replit?
Tools like Claude skills or Replit are great for writing and running code. But they are not built for data engineering workflows end to end. dltHub gives your team complete agentic workflows that cover every phase: coding, running, deploying, and debugging pipelines, on infrastructure you control.
How is dlt different from Fivetran or a Python script that uses the request library?
dlt is the perfect match between standardization and customization. You get the automation that matters: schema inference, incremental state, normalization, and loading, while keeping the full flexibility and portability of plain Python. And with agentic dltHub workflows, your team can code, run, deploy, and debug pipelines faster.
How do I get access to dltHub?
dltHub is available now. Book a demo with our team to get set up, or see our pricing page for plans and what's included.