CortexDB Docs
Integrations

Apache Airflow

Capture and recall CortexDB memory from Airflow DAGs using the Python SDK.

Airflow has no dedicated adapter — use the Python SDK directly inside a PythonOperator or @task.

Install

pip install cortexdbai apache-airflow

A capture task

from airflow.decorators import dag, task
from cortexdb.v1 import V1Client
import os, pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1), catchup=False)
def cortexdb_ingest():
    @task
    def ingest(**ctx):
        client = V1Client(api_url="https://api-v1.cortexdb.ai", actor="service:etl",
                          bearer=os.environ["CORTEX_TOKEN"])
        for row in fetch_rows():
            client.experience(
                scope="org:acme/service:etl", text=row["text"], role="user",
                observed_at=row["ts"],
                idempotency_key=f"{ctx['run_id']}-{row['id']}",   # idempotent across retries
            )
    ingest()

cortexdb_ingest()

Keying idempotency_key off the Airflow run_id makes a retried task replay rather than duplicate.

See also

On this page