cortexdbai 0.2.5

The Long-Term Memory Layer for AI Systems

CortexDB Python SDK

The Long-Term Memory Layer for AI Systems.

CortexDB gives your AI agents persistent, searchable memory. Store conversations, decisions, incidents, and domain knowledge as structured episodes. Retrieve relevant context with a single call.

Installation

pip install cortexdbai

For HTTP/2 support (recommended for async workloads):

pip install cortexdbai[async]

Quick Start

from cortexdb import Cortex

# Connect to CortexDB
cortex = Cortex("http://localhost:3141", api_key="your-api-key")

# Remember something
cortex.remember("The payments service was migrated to Kubernetes on March 15th.")

# Recall relevant memories
result = cortex.recall("When did we migrate payments?")
for match in result.matches:
    print(f"[{match.confidence:.0%}] {match.content}")

# Forget when needed (GDPR, cleanup, etc.)
cortex.forget(query="old deploy logs", reason="quarterly cleanup")

# Search with filters
results = cortex.search(
    "deploy failure",
    source="github",
    time_range="7d",
    limit=10,
)
print(results.context)

Async Usage

Every method has an async counterpart via AsyncCortex:

import asyncio
from cortexdb import AsyncCortex

async def main():
    async with AsyncCortex("http://localhost:3141", api_key="your-api-key") as cortex:
        await cortex.remember("Async memory storage works great.")

        result = await cortex.recall("How does async work?")
        for match in result.matches:
            print(match.content)

asyncio.run(main())

Episode Ingestion

For structured data, use episodes with full metadata:

from cortexdb import Cortex, IngestEpisodeRequest, Actor, Source, ActorType, EpisodeType

cortex = Cortex("http://localhost:3141")

# Ingest a structured episode
episode = IngestEpisodeRequest(
    content="PR #412: Add retry logic to payment processor",
    episode_type=EpisodeType.CODE_CHANGE,
    actor=Actor(id="ci-bot", name="CI Bot", actor_type=ActorType.SERVICE),
    source=Source(connector="github", external_id="pr-412", url="https://github.com/org/repo/pull/412"),
    metadata={"branch": "main", "files_changed": 3},
)
response = cortex.ingest_episode(episode)
print(f"Episode {response.episode_id} ingested.")

# Or use the smart store shorthand
cortex.store("Plain text goes through remember().")
cortex.store({
    "content": "Dict data goes through ingest_episode().",
    "type": "deployment",
    "source": "slack",
})

Entity Lookup

Explore entities and their relationships in the knowledge graph:

cortex = Cortex("http://localhost:3141")

# Look up an entity
entity = cortex.entity("payments-service")
print(f"{entity.name} ({entity.entity_type})")
print(f"  Episodes: {entity.episode_count}")
print(f"  First seen: {entity.first_seen}")

# Browse relationships
for rel in entity.relationships:
    print(f"  {rel.relationship} -> {rel.target_entity_id} ({rel.confidence:.0%})")

# Create explicit relationships
cortex.link(
    source_entity_id="payments-service",
    target_entity_id="auth-service",
    relationship="DEPENDS_ON",
    fact="Payments calls auth for token validation",
    confidence=0.95,
)

Error Handling

The SDK provides a structured exception hierarchy:

from cortexdb import Cortex
from cortexdb.exceptions import (
    CortexError,          # Base exception
    CortexAPIError,       # HTTP 4xx/5xx errors
    CortexAuthError,      # HTTP 401/403
    CortexRateLimitError, # HTTP 429 (includes retry_after)
    CortexConnectionError,# Network failures
    CortexTimeoutError,   # Request timeouts
)

cortex = Cortex("http://localhost:3141")

try:
    cortex.remember("important data")
except CortexRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except CortexAuthError as e:
    print(f"Authentication failed: {e.message}")
except CortexConnectionError:
    print("Cannot reach CortexDB server")
except CortexAPIError as e:
    print(f"API error [{e.status_code}]: {e.message}")

Transient errors (429, 502, 503, 504) and connection failures are automatically retried with exponential backoff. Configure retry behavior at client initialization:

cortex = Cortex(
    "http://localhost:3141",
    max_retries=5,   # default: 3
    timeout=60.0,    # default: 30.0 seconds
)

Configuration

Constructor Parameters

Parameter	Type	Default	Description
base_url	str	http://localhost:3141	CortexDB server URL
api_key	str	None	Bearer token for authentication
timeout	float	30.0	Request timeout in seconds
max_retries	int	3	Max retries for transient failures

Environment Variables

Variable	Description
CORTEXDB_API_KEY	Fallback API key when none is passed directly

Convenience Methods

Method	Description
cortex.memory(q)	Alias for recall() -- the GTM one-liner
cortex.store(data)	Smart dispatch: str -> remember(), dict -> ingest_episode()
cortex.context(q)	Recall with higher token budget (8192) for LLM agents

Links

• Documentation
• GitHub Repository
• Issue Tracker
• Changelog

License

Apache-2.0