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.get_activities(), # remember, recall, forget
interceptors=[CortexDBInterceptor(client=client, scope=scope)],
)The bundled remember activity sets no idempotency key, so a retried activity writes again. For
exactly-once capture, write your own activity that calls
client.experience(..., idempotency_key=f"{info.workflow_run_id}-{info.activity_id}") (keys are at most
64 characters).