CortexDB Docs
Integrations

Temporal

Capture and recall CortexDB memory from Temporal workflows and activities.

CortexDB integrates with Temporal as activities (durable capture/recall steps) and an interceptor that records workflow events automatically.

Install

pip install cortexdbai[temporal]

Activities + interceptor

from cortexdb import Cortex
from cortexdb.integrations.temporal import CortexDBActivities, CortexDBInterceptor

client = Cortex(api_url="https://api-v1.cortexdb.ai", actor="service:worker",
                bearer=os.environ["CORTEX_TOKEN"])
scope = "org:acme/service:worker"

activities = CortexDBActivities(client=client, scope=scope)

worker = Worker(
    client=temporal_client,
    task_queue="memory",
    activities=[activities.experience, activities.recall],
    interceptors=[CortexDBInterceptor(client=client, scope=scope)],
)

Activities keep capture idempotent across retries — use the Temporal run_id or activity id as the idempotency_key.

See also

On this page