Python SDK

The official Python SDK for CortexDB. Supports Python 3.9+ with both synchronous and asynchronous clients.

Installation

pip install cortexdb

Client Initialization

Synchronous Client

from cortexdb import Cortex

client = Cortex(
    base_url="http://localhost:3141",     # CortexDB server URL
    api_key="your-api-key",               # API key for authentication
    timeout=30.0,                          # Request timeout in seconds
    max_retries=3,                         # Retry count for transient errors
)

Async Client

from cortexdb import AsyncCortex

client = AsyncCortex(
    base_url="http://localhost:3141",
    api_key="your-api-key",
)

# Use as async context manager
async with AsyncCortex(api_key="your-api-key") as client:
    results = await client.recall(query="test")

Environment Variables

The client reads the CORTEXDB_API_KEY environment variable as a fallback when no api_key is provided:

import os
os.environ["CORTEXDB_API_KEY"] = "your-api-key"

client = Cortex()  # Reads from env

Methods Reference

client.remember()

Store a piece of information in CortexDB.

result = client.remember(
    content="We decided to migrate to CockroachDB for payments.",
    tenant_id="default",        # Optional. Defaults to "default".
    scope="engineering",         # Optional. Scope qualifier.
    metadata={"team": "infra"},  # Optional. Arbitrary key-value metadata.
)

Returns: RememberResponse

class RememberResponse:
    event_id: str        # Unique event ID assigned by the server

client.recall()

Retrieve relevant memories matching a natural-language query.

results = client.recall(
    query="What database are we using for payments?",
    tenant_id="default",          # Optional. Defaults to "default".
    max_tokens=4096,               # Optional. Token budget for response.
    min_confidence=0.0,            # Optional. Minimum confidence threshold.
)

Returns: RecallResponse

class RecallResponse:
    context: str          # Synthesized context from matching memories
    confidence: float     # Confidence score
    latency_ms: int       # Server-side latency in milliseconds

client.forget()

Delete memories matching a query.

result = client.forget(
    query="payments migration",    # Optional. Pattern to match.
    reason="Decision reversed",    # Optional. Justification for deletion.
    tenant_id="default",           # Optional. Defaults to "default".
)

Returns: ForgetResponse

class ForgetResponse:
    forgotten_entities: int    # Number of entities forgotten
    forgotten_edges: int       # Number of edges forgotten
    audit_id: str              # Audit trail identifier

client.search()

Search across memories and episodes with filters.

results = client.search(
    query="authentication issues",
    source="slack",                 # Optional. Filter by source connector.
    episode_type="message",         # Optional. Filter by episode type.
    time_range="7d",                # Optional. Time range (e.g. "1h", "24h", "7d", "30d").
    namespace="engineering",        # Optional. Filter by namespace.
    limit=20,                       # Optional. Max results. Default 20.
    offset=0,                       # Optional. Pagination offset.
    tenant_id="default",            # Optional. Defaults to "default".
)

Returns: SearchResponse

class SearchResponse:
    context: str                    # Synthesized context
    confidence: float               # Confidence score
    latency_ms: int                 # Server-side latency
    episodes: List[EpisodeBrief]    # Matching episodes
    total_episodes: int             # Total matches

client.entity()

Get detailed information about a specific entity.

entity = client.entity(
    entity_id="ent_abc123",
    tenant_id="default",            # Optional. Defaults to "default".
)

Returns: EntityResponse

class EntityResponse:
    id: str
    name: str
    entity_type: str
    first_seen: str
    last_seen: str
    episode_count: int
    relationships: List[EntityRelationship]
    recent_episodes: List[EpisodeBrief]

client.link()

Create an explicit relationship between two entities.

result = client.link(
    source_entity_id="ent_123",
    target_entity_id="ent_456",
    relationship="DEPENDS_ON",
    fact="Payments service depends on Redis for caching",  # Optional.
    confidence=0.9,                  # Optional. Default 0.9.
    tenant_id="default",             # Optional. Defaults to "default".
)

Returns: LinkResponse

class LinkResponse:
    id: str               # Created edge ID
    created_at: str        # Timestamp

client.ingest_episode()

Ingest a structured episode.

from cortexdb import IngestEpisodeRequest, Actor, ActorType, Source

episode = IngestEpisodeRequest(
    content="Deploy succeeded at 14:32 UTC.",
    type="event",
    tenant_id="default",
    actor=Actor(id="ci-bot", name="CI Bot", type=ActorType.SERVICE),
    source=Source(system="github-actions"),
    metadata={"commit": "abc123"},
)

result = client.ingest_episode(episode)

You can also use the from_dict convenience method:

episode = IngestEpisodeRequest.from_dict({
    "content": "Deploy succeeded at 14:32 UTC.",
    "type": "event",
    "source": "github-actions",
    "actor": "ci-bot",
})
result = client.ingest_episode(episode)

Returns: IngestEpisodeResponse

class IngestEpisodeResponse:
    episode_id: str
    event_id: str
    ingested_at: str

client.list_episodes()

List episodes with optional filtering.

result = client.list_episodes(
    tenant_id="default",       # Optional. Defaults to "default".
    offset=0,                   # Optional. Pagination offset.
    limit=50,                   # Optional. Max results. Default 50.
)

Returns: ListEpisodesResponse

class ListEpisodesResponse:
    episodes: List[Episode]
    total: int
    offset: int
    limit: int

client.health()

Check server health.

health = client.health()
print(health.status)    # "ok"
print(health.version)   # Server version

client.metrics()

Retrieve server metrics.

metrics = client.metrics()

Convenience Methods

client.memory()

One-liner alias for recall():

result = client.memory("why did checkout fail yesterday")

client.store()

Smart store that accepts a string or a dict:

# String — delegates to remember()
client.store("The deploy succeeded at 14:32 UTC.")

# Dict — converts to IngestEpisodeRequest and delegates to ingest_episode()
client.store({
    "content": "Deploy ok",
    "type": "event",
    "source": "ci",
})

client.context()

Retrieve enriched context for an LLM agent with a higher default token budget (8192):

result = client.context(
    query="current status of payments",
    agent_id="incident-bot",      # Optional. Agent identifier.
    tenant_id="default",           # Optional.
    max_tokens=8192,               # Optional. Default 8192.
)

Error Handling

from cortexdb import (
    CortexError,           # Base exception
    CortexAPIError,        # HTTP error response (has status_code, error_code)
    CortexAuthError,       # HTTP 401 or 403
    CortexRateLimitError,  # HTTP 429 (has retry_after)
    CortexConnectionError, # Cannot connect to server
    CortexTimeoutError,    # Request timed out
)

try:
    client.recall(query="test")
except CortexRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except CortexAuthError as e:
    print("Check your API key")
except CortexAPIError as e:
    print(f"[HTTP {e.status_code}] {e.message}")

The client automatically retries on transient errors (429, 502, 503, 504) with exponential backoff.

Next Steps