From claim to evidence, hands-on — Cordata's OSS projects

Cordata publishes four open-source Python packages, and this post is a tour of all four, run together on one dataset rather than described one README at a time. They come out of the posts on this site: a pipeline executor that emits its own evidence, a small emitter for code that emits no lineage of its own, a tool that turns those events into the records a data protection officer asks for, and a catalog server an AI agent can query. Together they are the practical side of the arc this site has drawn from the pillars through the pipeline series, which part 2 closes as from claim to evidence: here the evidence is run rather than described. Each package is on PyPI, and the walkthrough below starts from pip install with nothing cloned.

The dataset is a table of scored card transactions with an IBAN column — the bank account number, which is personal data under the GDPR, the DSGVO in German-speaking markets. The question is the one a data protection officer actually asks about a column like that: where does this data live, and when somebody asks for it to be erased, did the erasure reach all of it?

The catalog knows the transactions table the IBANs arrive in, the scored table built from it, and every pipeline that declares it reads the source, and it is right about all of that. The lineage finds one more place the IBANs went: a CSV file an export job wrote and never registered anywhere, whose only record is the event the job emitted when it ran. Finding that copy takes both kinds of tool, and that is what the rest of this post shows.

§ 1 — The four packages at a glance

The first three speak OpenLineage, an open standard for the events a job emits when it runs: which job, which datasets it read and wrote, and any number of facets — typed attachments such as a schema or a classification. The fourth speaks the Model Context Protocol, MCP, the protocol an AI assistant uses to call tools on a server it did not build.

The diagram, in words: the solid arrows carry OpenLineage events, and the dashed one does not. The runtime and the export script both emit events into one directory of newline-delimited JSON files, and qedro reads that directory. The catalog server never reads the events. It installs the runtime as a dependency, reads its descriptor model and example catalog, and answers an agent over MCP. What the two sides share is how they name a dataset, which is what § 5 relies on.

Everything below was run on 2 November 2026 in an empty directory with Python 3.14 — the packages need 3.12 or later:

python -m venv .venv && . .venv/bin/activate
pip install 'cordata-pipeline-runtime[run]==0.2.9' art30-emit==0.1.0 \
    qedro==0.10.1 cordata-catalog-mcp==0.3.0

All four install into one environment without a conflict. The output blocks are copied from that run and shortened only by leaving lines out, which … marks.

§ 2 — cordata-pipeline-runtime: three pipelines, one event stream

cordata-pipeline-runtime is the executor from A pipeline is a descriptor, not a program and its second part: a pipeline is a YAML declaration of its source, its SQL steps, its quality expectations and its policy tags, and one generic executor runs every such declaration. pipeline-runtime demo seeds a local example — a DuckDB catalog standing in for Glue, and Parquet files for the warehouse — into the directory it is given, here walk, and runs its three pipelines, which write their OpenLineage events to walk/out:

$ pipeline-runtime demo walk
seed        50,000 transactions at v7, 10,240 CDC claims rows
[7a06] descriptor  transactions-scored-daily (fraud)      apiVersion v1 OK
[7a06] schema      fraud_raw.transactions                 pinned v7, catalog v7 OK
[7a06] read        glue_table                             50,000 rows
[7a06] step        filter_settled        (sql)            35,919 rows
[7a06] step        score                 (sql)            35,919 rows
[7a06] policy      sensitivity=high residency=eu subject_type=customer resolved against ontology
[7a06] expect      transactions_scored                    5 passed, 0 failed
[7a06] write       fraud_curated.transactions_scored      iceberg, 1 partition key
[7a06] emit        COMPLETE → lineage adapter             inputs=1 outputs=1
…
[639b] descriptor  merchant-settlement-daily (finance)    apiVersion v2 OK
[639b] schema      fraud_raw.transactions                 pinned v7, catalog v7 OK
…
[639b] emit        COMPLETE → lineage adapter             inputs=1 outputs=1
 
events      walk/out/lineage.ndjson

Two domains read fraud_raw.transactions: fraud scores it, and finance settles merchants from it. Every event a run emits carries the pipeline's declared purpose and lawful basis as a processing facet; the dataset it wrote carries the resolved policy tags as the standard tags facet, and the run names the application release that published the data it read. None of that is written down separately, so none of it can drift from what ran.

The demo data is synthetic: the runtime generates its IBANs from a fixed random seed, so none of them is a real bank account.

§ 3 — Wrapping an export script with art30-emit

Pipelines are the easy case, because something already runs them and can be taught to emit. The copies that go missing are made elsewhere: a script somebody wrote to send the high-scoring transactions to the fraud investigators, run from a cron entry, emitting nothing. This one exports every transaction scoring 0.5 or above, IBAN included, to a CSV file:

import os
 
import duckdb
from art30_emit import declare
 
SCORED = "cordata://fraud/fraud_curated.transactions_scored"
EXPORT = "file://localhost/exports/fraud_cases.csv"
 
os.makedirs("exports", exist_ok=True)
 
with declare(
    "cordata.fraud/case-export",
    purpose="fraud-investigation",
    legal_basis="legitimate-interest",
    reads=[SCORED],
    writes=[EXPORT],
    classify={
        EXPORT: {
            "data_category": "financial",
            "subject_type": "customer",
            "residency": "eu",
            "retention": "1y",
        }
    },
):
    duckdb.sql("""
        COPY (
            SELECT tx_id, account_id, iban, amount_eur, fraud_score
            FROM read_parquet(
                'walk/warehouse/cordata-fraud-curated-eu-central-1/transactions_scored/**/*.parquet',
                hive_partitioning = true)
            WHERE fraud_score >= 0.5
        ) TO 'exports/fraud_cases.csv' (HEADER)
    """)

declare(...) from art30-emit is the only addition to an otherwise ordinary script. It emits a START event on entry and a COMPLETE on exit — or FAIL, if the work raises — with the same processing facet the runtime emits and the same standard tags facet. It decides nothing itself: no purpose is inferred from the job name, and a field the caller did not declare is absent from the event rather than empty, because nobody said and it is none are different answers to an auditor.

The dataset it reads is named exactly as the runtime names it, cordata://fraud plus fraud_curated.transactions_scored, so the two event streams join without any mapping. The run writes its events beside the runtime's:

ART30_EMIT_FILE=walk/out/case-export.ndjson python case_export.py

§ 4 — qedro: the Art. 30 record and the erasure walk

qedro reads OpenLineage events and produces the artefacts a data protection officer asks for. The first is the record of processing activities that Art. 30 GDPR requires a controller to keep — who processes which data, for what purpose, under which lawful basis — usually maintained by hand in a spreadsheet that describes the pipelines as somebody remembered them.

That record carries more weight than its format suggests. Art. 30(4) obliges a controller to hand it to the supervisory authority on request, and Art. 5(2) makes the controller answerable for being able to demonstrate compliance, not merely for achieving it. In practice the obligation turns into a handful of recurring problems, and each of them is one qedro command:

• A register that drifts out of date. Kept by hand, the record is correct on the day somebody last updated it and drifts with every deployment after that, and the drift is invisible, because a stale register reads exactly like a current one. qedro ropa regenerates it from what the pipelines actually did, and every line says whether it rests on an emitted fact, on a mapping somebody wrote or on a declaration.
• Audit preparation by hand. When a regulator, an auditor or a customer's data-protection questionnaire asks how personal data is processed, the answer is often assembled from a wiki and people's memory in the week before it is due. Here it is a file generated on demand — text, markdown, JSON or the spreadsheet an auditor asks for — carrying its own statement of what it covers and what it does not.
• Changes that go undocumented. qedro diff compares two records, so running it month on month shows what changed, including a purpose that is still listed but no longer emitted by anything. Given the AI governance team's own register as one side, it reports where that register and the pipelines disagree, and who owns each disagreeing row.
• Erasure requests on a deadline. Art. 12(3) gives a controller one month to act on a request, extendable by two more only for complex or numerous requests, and the hard part is every derived copy somebody forgot. qedro erasure walks the lineage forward from the erased table and lists each descendant, and whether a job has rewritten it since.
• AI Act duties on the way. For an organisation deploying a high-risk AI system, qedro ropa --view deployer reads the deployer's questions off the same record: the purpose and lawful basis, the model versions each run reported, and the inputs it read.

None of this needs Cordata's other packages. qedro reads OpenLineage from dbt, Airflow, Spark, Flink or anything else that emits it, and the richer the facets an emitter sends, the more of the record stands on evidence rather than on the mapping file.

On the demo's events, the first command checks that they are readable at all:

$ qedro events walk/out
  8 events · 7 jobs · 6 datasets · 2 files  ∎
  complete 4, other 3, start 1

The ∎ is the end-of-proof mark, and qedro prints it only when the output stands on its own evidence. For the record, the controller has to be named — that is an assertion about an organisation rather than something lineage can carry — so it goes into a qedro.yaml of four lines:

controller:
  name: Example Finanz GmbH
  contact: [email protected]
 
domains: [fraud, policy, finance]

Everything else comes off the wire:

$ qedro ropa walk/out --config qedro.yaml
Record of processing activities — Example Finanz GmbH
  contact: [email protected]
…
  cordata.fraud/case-export
    purpose       fraud-investigation
    legal basis   legitimate-interest
    categories    financial
    subjects      customer
    residency     eu
    retention     1y
    reads         cordata://fraud/fraud_curated.transactions_scored
    writes        file://localhost/exports/fraud_cases.csv
    runs          1 in window, last 2026-11-02T07:41:19.615221+00:00
 
  cordata.fraud/transactions-scored-daily
    purpose       fraud-detection
    legal basis   legitimate-interest
    subjects      customer
    residency     eu
    unclassified  1 of 2 datasets carries no classification:
                  cordata://fraud/fraud_raw.transactions
    reads         cordata://fraud/fraud_raw.transactions
    writes        cordata://fraud/fraud_curated.transactions_scored
    runs          1 in window, last 2026-11-02T07:41:18.132260+00:00
…
  Scope of this record
    source        walk/out
…
    provenance    7 evidenced, 0 from the mapping file, 0 undeclared
…
  every activity stands on emitted evidence   ∎

Seven activities, every purpose and lawful basis taken from an emitted facet, and none from the mapping file qedro would otherwise fall back on. The export script is in the record because it emitted; a script that emits nothing would appear only if somebody declared it, marked as an assertion.

The second artefact answers the erasure half of the question. A request under Art. 17 GDPR means deleting a person's data from the source and from everything derived from it, and every way that goes wrong is a derived copy somebody forgot. qedro erasure walks the lineage forward from the erased dataset:

$ qedro erasure walk/out --dataset fraud_raw.transactions --since 2026-11-02T00:00:00+00:00
Erasure of cordata://fraud/fraud_raw.transactions
 
    tombstone     2026-11-02T00:00:00+00:00 — from --since, not emitted
 
    cordata://finance/finance_curated.merchant_settlement_daily
      derived by    cordata.finance/merchant-settlement-daily
      state         rewritten 2026-11-02T07:41:18.689888+00:00 — 1 run since the tombstone
 
    cordata://fraud/fraud_curated.transactions_scored
      derived by    cordata.fraud/transactions-scored-daily
      state         rewritten 2026-11-02T07:41:18.132260+00:00 — 1 run since the tombstone
 
      file://localhost/exports/fraud_cases.csv
        derived by    cordata.fraud/case-export
        state         rewritten 2026-11-02T07:41:19.615221+00:00 — 1 run since the tombstone
…
    propagation   3 of 3 rewritten since the tombstone, 0 not
…
  this record does not claim to be a proof:
    ! the erasure instant came from --since rather than from an emitted lifecycle event, so every rewrite below is measured against a date somebody typed

Three descendants, two hops deep on one branch, and the CSV file is the second hop. The walk reaches it only because the export script emitted an event naming its input the way the runtime names it. The finance branch is there because the runtime names the table finance reads by the domain that owns it rather than by the domain reading it — until version 0.2.2 it did not, and the walk missed finance's table entirely (pipeline-runtime#7, found while building this demo).

The mark is withheld here, correctly. The runtime emits no event marking the erasure itself, so the instant is the one typed after --since, and qedro says so instead of measuring rewrites against a date as though somebody had proved it. It also says, in every run, that it reports datasets rather than rows: a rewritten descendant is evidence the erasure was propagated, not proof that one person's rows are gone.

§ 5 — Asking cordata-catalog-mcp where the IBANs are

cordata-catalog-mcp is the server designed in The catalog is the API: six read-only tools over the catalog — ownership, schema, LF-tags (Lake Formation's key–value classification labels), grant reach and declared readers — for an AI agent connected over MCP. It reads the runtime's descriptor model and never a row of data. A short client asks it the same questions:

import asyncio
import json
import sys
 
from mcp import Client
from mcp.client.stdio import StdioServerParameters
 
ANALYST = {
    "domain": ["fraud"],
    "sensitivity": ["public", "internal", "confidential"],
    "residency": ["eu"],
}
 
 
async def main():
    # The same server cordata-catalog-mcp starts, launched from this venv.
    server = StdioServerParameters(command=sys.executable, args=["-m", "catalog_mcp"])
    async with Client(server) as catalog:
        found = await catalog.call_tool("find_datasets", {"tags": {"subject_type": "customer"}})
        for match in found.structured_content["datasets"]:
            print("customer data in", f"{match['domain']}/{match['table']}")
 
        scored = await catalog.call_tool(
            "get_dataset", {"domain": "fraud", "table": "transactions_scored"}
        )
        dataset = scored.structured_content
        iban = next(c for c in dataset["columns"] if c["name"] == "iban")
        print("get_dataset owner", dataset["metadata"]["owner"])
        print("get_dataset iban", json.dumps(iban))
        print("get_dataset lineage", json.dumps(dataset["lineage"]))
 
        grant = await catalog.call_tool(
            "check_grant_expression",
            {"domain": "fraud", "table": "transactions_scored", "expression": ANALYST},
        )
        print("check_grant_expression", json.dumps(grant.structured_content))
 
        readers = await catalog.call_tool(
            "get_consumers", {"domain": "fraud", "table": "transactions"}
        )
        for consumer in readers.structured_content["consumers"]:
            print("declared reader of fraud/transactions:", consumer["pipeline"],
                  f"({consumer['domain']}, pins v{consumer['pins']})")
 
 
asyncio.run(main())

customer data in fraud/transactions_scored
customer data in policy/claims_deduped
get_dataset owner [email protected]
get_dataset iban {"name": "iban", "type": "VARCHAR", "lf_tags": {"domain": "fraud", "sensitivity": "high", "residency": "eu", "subject_type": "customer"}}
get_dataset lineage {"namespace": "cordata://fraud", "name": "fraud_curated.transactions_scored"}
check_grant_expression {"reaches": false, "columns": []}
declared reader of fraud/transactions: merchant-settlement-daily (finance, pins v7)
declared reader of fraud/transactions: transactions-scored-daily (fraud, pins v7)

The catalog knows who owns the scored table, that its IBAN column is classified high, and that the analyst role — granted every sensitivity except high — reaches none of it. That is what a catalog is for, and nothing in the event stream could answer it: lineage says what happened, and a grant is about who may.

Asked who reads the transactions table the IBANs arrive in, it names fraud's scoring pipeline and finance's settlement pipeline, with the schema version each pins. That list is the runtime's own: get_consumers calls the same function the runtime's producer-side check uses to fail an application's build when a schema change would break a pinned reader, so the catalog and that check cannot disagree about who depends on the table.

That check is the Data Contracts pillar's argument in one command: each consumer declares the version it reads, the list of consumers is generated from those declarations rather than kept by hand, and the producer's build fails when a proposed change would break one of them. The runtime ships two proposed releases of the application behind the transactions table with its example. Pointed at the demo's catalog, the check reads the proposal that renames a column both pipelines use:

EXAMPLE=$(python -c 'import os, pipeline_runtime.example as e; print(os.path.relpath(e.__path__[0]))')
CORDATA_CATALOG=walk/catalog.duckdb python -m pipeline_runtime.consumers \
    "$EXAMPLE/proposals/card-ledger-v5.0.0.yml" "$EXAMPLE/domains"

card-ledger release v5.0.0 proposes fraud_raw.transactions v8; the catalog is at v7
2 descriptors under .venv/lib/python3.14/site-packages/pipeline_runtime/example/domains read fraud_raw.transactions
 
BREAKS      merchant-settlement-daily (finance)  pins v7  owner [email protected]
            .venv/lib/python3.14/site-packages/pipeline_runtime/example/domains/finance/pipelines/merchant_settlement.yml
            + merchant_ref (VARCHAR, not null)
            - merchant_id (VARCHAR, not null)   <- breaks
 
BREAKS      transactions-scored-daily (fraud)  pins v7  owner [email protected]
            .venv/lib/python3.14/site-packages/pipeline_runtime/example/domains/fraud/pipelines/transactions_scored.yml
            + merchant_ref (VARCHAR, not null)
            - merchant_id (VARCHAR, not null)   <- breaks
 
2 of 2 consumers fail this check

It exits 1, which is what turns this into a failed build in the application's CI, before the release ships rather than after a pipeline breaks. The other proposal, v4.12.0, only adds a nullable column: the same check exits 0, and marks both pipelines BUMP — nothing they read changes, but each stops at its next run until its pin moves to v8, so no pipeline ever reads a version it was not written against.

The export script is on neither list. It declares nothing to the catalog — no descriptor, no registered output — so the catalog has no record of the job or of the CSV file it writes, and an agent asking where is this data gets a complete answer about everything it was told. One more consequence of where classification lives: the source table itself is marked unclassified, because classifying an application's table is its owner's decision, so find_datasets with subject_type=customer does not return it even though it holds the IBANs.

The catalog is right about every dataset that was registered with it. The copy that matters for an erasure is one that never was, and its only record is the event the job emitted when it ran.

The lineage field get_dataset returned is how the two answers meet. The catalog names the dataset fraud/transactions_scored; the events name it cordata://fraud plus fraud_curated.transactions_scored, because OpenLineage identifies a dataset by a namespace and a name. Since version 0.2.0 the catalog returns the lineage form beside its own, so an agent connected to both can take a dataset the catalog knows and look up what was derived from it — which is how the CSV file turns up — without either server reading the other.

§ 6 — Shared vocabulary, and one deliberate dependency

Only one of the four imports another. qedro and art30-emit install without the runtime, and qedro in particular must not depend on anything AWS-shaped: it reads classification from the standard tags facet whatever produced it, so LF-tags are one possible source beside Purview classifications or Google Cloud policy tags, and none of them is required.

The catalog server is the one that does, deliberately. It installs the runtime's base package — the descriptor models, the policy check against the ontology and, since 0.3.0, the module that reads pipelines' pins — and none of its execution stack. A vocabulary and a naming scheme can be copied between packages and checked for drift; a pin parser cannot, because two copies would drift, and a pin read differently is a list of readers that disagrees with the producer's own check.

What all four share is vocabulary, and the share is partial in a way worth stating precisely:

The processing facet is the strongest link. Its JSON schema is generated in qedro's repository from the runtime's own Processing model, with a test that fails when the two diverge, and art30-emit implements the same published schema — two independent emitters of one facet. The runtime's Metadata and Contract models are also what the catalog server returns, so its output schema is derived from the model rather than restated.

The table's fourth row is the gap. The runtime's ontology classifies by sensitivity, and qedro reports the categories of personal data Art. 30(1)(c) asks for from data_category, which the runtime never emits. So the export script's activity reports categories financial and the pipelines' activities report none.

Keeping the rest of the code apart has a side effect worth having: each package is checked by the others in a way its own tests do not check it. Three of the runtime's fixes in October were found by reading its events with qedro rather than by a test in the runtime — validation jobs that dropped the purpose and lawful basis (pipeline-runtime#9), a dataset classified in one event and not in the next (#10), and a signature verdict reported against a commit nothing named (#11).

§ 7 — Known limits in these versions

Each limit below is true of the versions above:

• Categories of personal data are blank for the pipelines' datasets, for the reason in § 6. Adding data_category to the runtime's ontology is a decision about the ontology, not a fix.
• The erasure instant is typed, not emitted. Nothing in the runtime emits OpenLineage's lifecycleStateChange facet for a deletion, so the erasure check needs --since and withholds its mark.
• A run from an installed copy names no commit. An installed package sits in no work tree, so its runs report no commit, no signature verdict and no standard sourceCodeLocation facet. That is deliberate: until 0.2.5 such a run reported descriptor_git_commit_signed: false while naming no commit, which reads as a commit reported unsigned, and pipeline-runtime#11 made both fields absent instead, because missing provenance is not bad provenance. descriptor_sha256 is unaffected, so the record still proves which descriptor bytes ran and stops one step short of the commit they came from. From a checkout — git clone --branch v0.2.9 and pip install -e — the same demo emits the commit and the standard facet: the facet since 0.2.8, and the demo has run in a clone at all only since 0.2.7.
• The catalog's column lists are a stand-in. A deployment reads them from each domain's Glue catalog; here a JSON file plays that part, and CI checks it against the Parquet the pipelines write.

If you run a catalog and are weighing whether lineage justifies its cost — or think this argument has a hole in it — I would like to hear it at cordata.tech/contact. Related reading: The catalog is the API for the design behind the catalog server, part 2 of the pipeline series for the events the runtime emits, and the AI Readiness pillar for why records derived from those events matter beyond the GDPR.