Python

There is no official Python SDK yet. All integration is via the REST API using httpx (recommended) or requests.

Installation

pip install httpx

httpx supports both sync and async usage, which makes it a good fit for most Python environments. If you prefer requests, every sync example below translates directly.

Quick start

Store and retrieve a memory:

import httpx

BASE = "http://localhost:4545"
HEADERS = {"Authorization": "Bearer YOUR_TOKEN"}  # omit if auth is disabled

def store_memory(content: str, memory_type: str = "short_term") -> dict:
    resp = httpx.post(
        f"{BASE}/api/v1/memories",
        json={"content": content, "memory_type": memory_type},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()

def search_memories(query: str, limit: int = 10) -> list:
    resp = httpx.post(
        f"{BASE}/api/v1/memories/search",
        json={"query": query, "search_type": "hybrid", "limit": limit},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["results"]  # {"results": [{"memory": ..., "score": ...}], "total": n}

# Usage
store_memory("User prefers dark mode", "long_term")
results = search_memories("dark mode preferences")

Async usage

For async frameworks (FastAPI, asyncio, etc.) use httpx.AsyncClient:

import httpx

BASE = "http://localhost:4545"
HEADERS = {"Authorization": "Bearer YOUR_TOKEN"}

async def store_memory(content: str, memory_type: str = "short_term") -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE}/api/v1/memories",
            json={"content": content, "memory_type": memory_type},
            headers=HEADERS,
        )
        resp.raise_for_status()
        return resp.json()

async def search_memories(query: str, limit: int = 10) -> list:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE}/api/v1/memories/search",
            json={"query": query, "search_type": "hybrid", "limit": limit},
            headers=HEADERS,
        )
        resp.raise_for_status()
        return resp.json()["results"]

For long-lived services, create the AsyncClient once and share it rather than opening a new client on every call.

LangChain integration pattern

You can wrap the REST API in a custom LangChain memory class to give any chain persistent memory:

from langchain.memory.chat_memory import BaseChatMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage
import httpx

class RememMemory(BaseChatMemory):
    """LangChain memory backed by Remem REST API."""

    base_url: str = "http://localhost:4545"
    token: str = ""
    search_limit: int = 5

    @property
    def _headers(self) -> dict:
        h = {"Content-Type": "application/json"}
        if self.token:
            h["Authorization"] = f"Bearer {self.token}"
        return h

    @property
    def memory_variables(self) -> list[str]:
        return ["history"]

    def load_memory_variables(self, inputs: dict) -> dict:
        query = inputs.get("input", "")
        if not query:
            return {"history": []}
        resp = httpx.post(
            f"{self.base_url}/api/v1/memories/search",
            json={"query": query, "search_type": "hybrid", "limit": self.search_limit},
            headers=self._headers,
        )
        resp.raise_for_status()
        results = resp.json()["results"]  # [{"memory": Memory, "score": float}, ...]
        return {"history": [r["memory"]["content"] for r in results]}

    def save_context(self, inputs: dict, outputs: dict) -> None:
        for role, text in [("human", inputs.get("input", "")), ("ai", outputs.get("output", ""))]:
            if text:
                httpx.post(
                    f"{self.base_url}/api/v1/memories",
                    json={"content": f"{role}: {text}", "memory_type": "long_term"},
                    headers=self._headers,
                ).raise_for_status()

    def clear(self) -> None:
        pass  # optional: implement bulk delete if needed

Usage:

from langchain.chains import ConversationChain
from langchain.chat_models import ChatOpenAI

memory = RememMemory(base_url="http://localhost:4545", token="YOUR_TOKEN")
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)
response = chain.predict(input="What did we discuss about the UI?")

Error handling

raise_for_status() converts HTTP error responses into httpx.HTTPStatusError. Catch it alongside connection errors for robust handling:

import httpx

def safe_store(content: str, memory_type: str = "short_term") -> dict | None:
    try:
        resp = httpx.post(
            f"{BASE}/api/v1/memories",
            json={"content": content, "memory_type": memory_type},
            headers=HEADERS,
            timeout=10.0,
        )
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPStatusError as exc:
        status = exc.response.status_code
        if status == 401:
            print("Authentication failed — check your token.")
        elif status == 422:
            print(f"Validation error: {exc.response.json()}")
        else:
            print(f"Server error {status}: {exc.response.text}")
        return None
    except httpx.ConnectError:
        print("Could not reach Remem — is the server running?")
        return None
    except httpx.TimeoutException:
        print("Request timed out.")
        return None

Next steps

• Memories API reference — full endpoint documentation including filters and bulk operations
• MCP Overview — use the MCP interface instead of REST for agent frameworks that support it