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-airflowA capture task
from airflow.decorators import dag, task
from cortexdb.v1 import V1Client
import hashlib, 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"],
# idempotent across retries; hashed because keys are capped at 64 characters
idempotency_key=hashlib.sha256(f"{ctx['run_id']}-{row['id']}".encode()).hexdigest()[:48],
)
ingest()
cortexdb_ingest()Keying idempotency_key off the Airflow run_id makes a retried task replay rather than duplicate.