Python SDK

The official Python client for MemoryRelay. Provides both synchronous and asynchronous clients with full type safety via Pydantic models.

Installation

pip install memoryrelay

tip

The package has minimal dependencies. It uses httpx for HTTP and pydantic for data validation.

Quick Start (Sync)

from memoryrelay import MemoryRelay

client = MemoryRelay(api_key="mem_your_api_key")

# Create an agent
agent = client.agents.create(name="my-agent", description="My first agent")

# Store a memory
memory = client.memories.create(
    agent_id=agent.id,
    content="The project uses PostgreSQL 16 with pgvector for semantic search.",
    metadata={"source": "architecture-review", "confidence": 0.95}
)

# Search by meaning
results = client.memories.search(
    agent_id=agent.id,
    query="What database does the project use?",
    limit=5
)

for result in results:
    print(f"[{result.score:.2f}] {result.content}")

Quick Start (Async)

import asyncio
from memoryrelay import AsyncMemoryRelay

async def main():
    async with AsyncMemoryRelay(api_key="mem_your_api_key") as client:
        agent = await client.agents.create(
            name="my-agent",
            description="Async agent"
        )

        memory = await client.memories.create(
            agent_id=agent.id,
            content="FastAPI handles async requests natively."
        )

        results = await client.memories.search(
            agent_id=agent.id,
            query="How are requests handled?",
            limit=5
        )

        for result in results:
            print(f"[{result.score:.2f}] {result.content}")

asyncio.run(main())

Authentication

The client accepts an API key in two ways:

Explicit parameter -- pass api_key when initializing the client.
Environment variable -- set MEMORYRELAY_API_KEY and omit the parameter.

import os

# Option 1: Explicit
client = MemoryRelay(api_key="mem_your_api_key")

# Option 2: Environment variable
os.environ["MEMORYRELAY_API_KEY"] = "mem_your_api_key"
client = MemoryRelay()  # picks up from environment automatically

warning

Never hard-code API keys in source files. Use environment variables or a secrets manager.

Context Manager

Both clients support context managers for automatic resource cleanup:

# Sync
with MemoryRelay(api_key="mem_your_api_key") as client:
    results = client.memories.search(agent_id=agent_id, query="deployment")

# Async
async with AsyncMemoryRelay(api_key="mem_your_api_key") as client:
    results = await client.memories.search(agent_id=agent_id, query="deployment")

Custom Base URL

For self-hosted deployments, point the client at your own instance:

client = MemoryRelay(
    api_key="mem_your_api_key",
    base_url="https://memory.your-domain.com"
)

Resources Reference

`client.memories`

Core memory storage and retrieval operations.

Method	Description
`create(agent_id, content, metadata=None)`	Store a new memory
`search(agent_id, query, limit=10)`	Semantic search across memories
`list(agent_id, limit=50, offset=0)`	List memories with pagination
`get(memory_id)`	Retrieve a specific memory
`update(memory_id, content=None, metadata=None)`	Update memory content or metadata
`delete(memory_id)`	Soft-delete a memory
`batch_create(agent_id, memories)`	Store multiple memories in one call

# Batch create
memories = client.memories.batch_create(
    agent_id=agent.id,
    memories=[
        {"content": "Python 3.11 is the minimum version."},
        {"content": "Tests use pytest with async fixtures."},
        {"content": "Redis is used for rate limiting."},
    ]
)

`client.agents`

Manage agent namespaces that isolate memory collections.

Method	Description
`create(name, description=None)`	Create a new agent
`list(limit=50, offset=0)`	List all agents
`get(agent_id)`	Get agent details
`update(agent_id, name=None, description=None)`	Update an agent
`delete(agent_id)`	Delete an agent and its memories
`stats(agent_id)`	Get memory count and usage statistics

`client.entities`

Knowledge graph nodes -- people, organizations, technologies, and concepts extracted from memories.

Method	Description
`create(agent_id, name, entity_type, metadata=None)`	Create an entity
`list(agent_id, entity_type=None, limit=50, offset=0)`	List entities with optional type filter
`get(entity_id)`	Get entity details
`update(entity_id, name=None, metadata=None)`	Update an entity
`delete(entity_id)`	Delete an entity
`link(entity_id, memory_id, relationship=None)`	Link an entity to a memory

# Create and link an entity
entity = client.entities.create(
    agent_id=agent.id,
    name="PostgreSQL",
    entity_type="technology",
    metadata={"version": "16"}
)

client.entities.link(
    entity_id=entity.id,
    memory_id=memory.id,
    relationship="uses"
)

`client.sessions`

Group related memories into bounded interaction periods.

Method	Description
`create(agent_id, metadata=None)`	Start a new session
`get_or_create(agent_id, session_key)`	Resume or start a session by key
`list(agent_id, limit=50, offset=0)`	List sessions
`get(session_id)`	Get session details and memories
`end(session_id)`	Mark a session as ended
`summarize(session_id)`	Generate a summary of the session
`delete(session_id)`	Delete a session

# Start or resume a session
session = client.sessions.get_or_create(
    agent_id=agent.id,
    session_key="project-review-2026-03-17"
)

# Store memories within the session context
client.memories.create(
    agent_id=agent.id,
    content="Reviewed the authentication module.",
    metadata={"session_id": str(session.id)}
)

# Summarize when done
summary = client.sessions.summarize(session_id=session.id)
client.sessions.end(session_id=session.id)

`client.decisions`

Track architectural and design decisions with lifecycle management.

Method	Description
`create(agent_id, title, content, metadata=None)`	Record a new decision
`list(agent_id, status=None, limit=50, offset=0)`	List decisions
`get(decision_id)`	Get decision details
`update(decision_id, title=None, content=None)`	Update a decision
`delete(decision_id)`	Delete a decision
`check(agent_id, query)`	Check if a query conflicts with existing decisions
`supersede(decision_id, new_decision_id, reason=None)`	Mark a decision as superseded

# Record a decision
decision = client.decisions.create(
    agent_id=agent.id,
    title="Use pgvector for vector search",
    content="Chose pgvector over a separate vector database to reduce operational complexity.",
    metadata={"category": "architecture"}
)

# Later, supersede it
new_decision = client.decisions.create(
    agent_id=agent.id,
    title="Migrate to OpenAI embeddings",
    content="Switching from local sentence-transformers to OpenAI for higher quality embeddings."
)

client.decisions.supersede(
    decision_id=decision.id,
    new_decision_id=new_decision.id,
    reason="OpenAI embeddings improved search relevance by 23%."
)

`client.projects`

Multi-project context management with dependency tracking.

Method	Description
`create(agent_id, name, description=None, metadata=None)`	Register a project
`list(agent_id, limit=50, offset=0)`	List projects
`get(project_id)`	Get project details
`update(project_id, name=None, description=None)`	Update a project
`delete(project_id)`	Delete a project
`context(project_id)`	Get full project context (memories, decisions, patterns)

`client.patterns`

Discover and share reusable patterns across projects.

Method	Description
`create(agent_id, name, content, metadata=None)`	Create a pattern
`search(agent_id, query, limit=10)`	Search patterns by meaning
`list(agent_id, limit=50, offset=0)`	List patterns
`get(pattern_id)`	Get pattern details
`update(pattern_id, name=None, content=None)`	Update a pattern
`delete(pattern_id)`	Delete a pattern
`adopt(pattern_id, project_id)`	Adopt a pattern for a project
`unadopt(pattern_id, project_id)`	Remove a pattern adoption

Error Handling

The SDK raises typed exceptions for different failure modes:

from memoryrelay import MemoryRelay
from memoryrelay.exceptions import (
    MemoryRelayError,      # Base exception for all SDK errors
    AuthenticationError,   # Invalid or expired API key (401)
    NotFoundError,         # Resource not found (404)
    ValidationError,       # Invalid request parameters (422)
    RateLimitError,        # Too many requests (429)
    APIError,              # Other API errors (5xx, network issues)
)

client = MemoryRelay(api_key="mem_your_api_key")

try:
    memory = client.memories.get(memory_id="nonexistent-id")
except NotFoundError:
    print("Memory does not exist.")
except AuthenticationError:
    print("Check your API key.")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds.")
except ValidationError as e:
    print(f"Invalid request: {e.details}")
except MemoryRelayError as e:
    print(f"Unexpected error: {e}")

info

All exceptions inherit from MemoryRelayError, so you can catch it as a fallback for any SDK error.

Installation​

Quick Start (Sync)​

Quick Start (Async)​

Authentication​

Context Manager​

Custom Base URL​

Resources Reference​

client.memories​

client.agents​

client.entities​

client.sessions​

client.decisions​

client.projects​

client.patterns​

Error Handling​