memoair 0.2.1

Python SDK for MemoAir - Memory as a Service for AI Agents

MemoAir Python SDK

Memory as a Service for AI Agents

MemoAir provides persistent, graph-augmented memory for AI agents. The SDK talks directly to the production MemoAir Go backend for the common add / search loop and exposes optional helpers for OpenAI function calling and LangChain.

Installation

pip install memoair

Quick start

from memoair import MemoAir

client = MemoAir(api_key="memoair_pk_...")  # or set MEMOAIR_API_KEY

# Save a memory
client.add(
    "Alice prefers Python for backend services",
    user_id="alice",
    tags=["preference"],
)

# Recall it
results = client.search_memories(
    "What language does Alice like?",
    user_id="alice",
)

print(results.to_prompt_context())
# 1. (87% relevant) [Bio] Alice prefers Python for backend services [ID: note-...]

The headline API is two methods:

• client.add(content, user_id=..., tags=..., metadata=...) saves a memory via POST /v1/notes/batchUpsert on the Go backend.
• client.search_memories(query, user_id=..., scope="org|user|all") retrieves relevant memories via POST /v1/search and strips memvid metadata from snippets so the result is LLM-ready.

Both have async equivalents on AsyncMemoAir.

Voice Memory Runtime

For voice agents, the SDK exposes one required memory tool backed by a local Rust memoair-runtime process. By default the SDK manages that process: it checks loopback health, downloads and verifies the matching GitHub Release binary on first use, caches it under ~/.cache/memoair/runtime, starts it with your workspace/user/API-key config, then talks to it over 127.0.0.1.

from memoair import MemoAirVoiceMemory

async with MemoAirVoiceMemory(
    api_key="mka_...",
    workspace_id="ws_123",
    user_id="user_42",
) as memory:
    tool = memory.search_memory_tool()
    context = await tool("what timezone does this user prefer?")

Advanced production deployments can run memoair-runtime explicitly as a Docker/systemd/Kubernetes sidecar and pass runtime_url=... or set MEMOAIR_RUNTIME_BINARY to use a preinstalled binary instead of downloading.

Configuration

client = MemoAir(
    api_key="memoair_pk_...",            # or MEMOAIR_API_KEY
    base_url="https://api.memoair.dev",  # Production Go backend (or MEMOAIR_BASE_URL)
    graph_url="https://graph.memoair.dev",  # Legacy graph service (or MEMOAIR_GRAPH_URL)
    workspace_id="ws_default",            # MEMOAIR_WORKSPACE_ID
    user_id="alice",                      # MEMOAIR_USER_ID — default X-User-Id
    timeout=30.0,
    max_retries=3,
)

When workspace_id and/or user_id are set on the client, every call inherits them unless overridden per-call.

Resources

# Production (Go backend)
client.notes.add(content="...", user_id="alice")
client.notes.list(workspace_id="ws_default")
client.notes.delete("note-id")

client.search.search(
    "What does Alice prefer?",
    user_id="alice",
    scope="user",
    tags=["preference"],
)

The legacy graph-service resources (client.memories, client.facts, client.documents, client.ontology, client.communities, client.agent) are still available for advanced use cases. The headline methods (memories.add, memories.get, search.query, search.get_memory) emit DeprecationWarning and will be removed in a future release — migrate to client.add() / client.search_memories().

Async

import asyncio
from memoair import AsyncMemoAir

async def main():
    async with AsyncMemoAir(api_key="memoair_pk_...") as client:
        await client.add("Alice prefers Python", user_id="alice")
        results = await client.search_memories(
            "What does Alice prefer?",
            user_id="alice",
        )
        print(results.to_prompt_context())

asyncio.run(main())

OpenAI tool calling

The SDK ships ready-to-use tools specs and a dispatcher so MemoAir slots into an existing OpenAI agent loop with two lines of glue.

from openai import OpenAI
from memoair import MemoAir

mem = MemoAir(api_key="memoair_pk_...")
llm = OpenAI()
messages = [{"role": "user", "content": "Remember I prefer dark mode."}]

while True:
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=mem.openai_tools(),
    )
    msg = response.choices[0].message
    messages.append(msg.model_dump())

    if not msg.tool_calls:
        break

    for call in msg.tool_calls:
        messages.append(mem.dispatch_tool_call(call, user_id="alice"))

print(messages[-1]["content"])

The two tools are:

• memoair_save_memory(content, tags?, metadata?, graph_target?)
• memoair_search_memories(query, scope?, limit?, tags?)

Use AsyncMemoAir.dispatch_tool_call(...) for async loops.

LangChain

Install the optional extra:

pip install "memoair[langchain]"

The integration ships four primitives plus a small helper:

Primitive	LangChain base class	Use it for
MemoAirRetriever	BaseRetriever	Semantic fact search inside chains and agents
MemoAirTripartiteRetriever	BaseRetriever	One-call retrieval across user, organization, and ontology graphs
MemoAirChatMessageHistory	BaseChatMessageHistory	Drops directly into RunnableWithMessageHistory
MemoAirSearchTool, MemoAirSaveTool	BaseTool	Function-calling agents (create_agent(tools=[...]))
MemoAirMemory	helper	Quick retrieve-context + save-turn loop

Retriever

from langchain_openai import ChatOpenAI
from memoair import MemoAir
from memoair.integrations.langchain import MemoAirRetriever

client = MemoAir(api_key="memoair_pk_...")
retriever = MemoAirRetriever(client=client, group_ids=["user:alice"], max_facts=5)
docs = retriever.invoke("What does Alice prefer for backend work?")

Native chat history (RunnableWithMessageHistory)

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from memoair import MemoAir
from memoair.integrations.langchain import MemoAirChatMessageHistory

client = MemoAir(api_key="memoair_pk_...")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant with persistent MemoAir memory."),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)
chain = prompt | ChatOpenAI(model="gpt-4o")

with_history = RunnableWithMessageHistory(
    chain,
    lambda session_id: MemoAirChatMessageHistory(client=client, group_id=session_id),
    input_messages_key="input",
    history_messages_key="history",
)

Function-calling agents (BaseTool)

from langchain.agents import create_agent
from memoair import MemoAir
from memoair.integrations.langchain import MemoAirMemory

client = MemoAir(api_key="memoair_pk_...")
memory = MemoAirMemory(client=client, group_id="user:alice")

agent = create_agent(
    model="openai:gpt-4o",
    tools=memory.as_tools(),  # MemoAirSearchTool + MemoAirSaveTool
    system_prompt="Use MemoAir to recall facts and save anything worth remembering.",
)

The complete agent template at sdk/examples/langchain_agent.py composes all four primitives.

LangGraph checkpointers are intentionally not provided: MemoAir's API is fact-extraction oriented and does not expose the raw key-value state surface that BaseCheckpointSaver requires. Use MemoAirChatMessageHistory (or memory.as_tools()) for memory and a separate LangGraph checkpointer (MemorySaver, SQLite, Postgres, Redis) for graph state.

LangSmith observability

Install the optional extra:

pip install "memoair[langsmith]"

LangChain apps can enable LangSmith tracing with environment variables and keep using MemoAirRetriever, MemoAirChatMessageHistory, or MemoAirSearchTool.

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsv2_...
export LANGSMITH_PROJECT=memoair-agent

For direct SDK apps, wrap retrieval with a LangSmith retriever span:

from memoair import MemoAir
from memoair.integrations.langsmith import trace_memoair_search

client = MemoAir(api_key="memoair_pk_...", workspace_id="ws_default", user_id="alice")

documents = trace_memoair_search(
    client,
    "What does Alice prefer?",
    user_id="alice",
    workspace_id="ws_default",
    scope="all",
    limit=5,
    metadata={"environment": "production"},
)

The helper returns LangSmith-compatible documents with page_content, type="Document", and MemoAir metadata such as memory ID, score, scope, user ID, workspace ID, graph facts, and graph reasoning paths.

Error handling

from memoair import (
    MemoAir,
    MemoAirError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
)

try:
    client.search_memories("hello", user_id="alice")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited; retry after {e.retry_after}s")
except MemoAirError as e:
    print(f"MemoAir error: {e.message}")

Advanced / legacy graph APIs

The legacy graph-service resources remain available. They are deprecated for new code and should not appear in your hot path.

# Episodic graph ingestion (legacy)
from memoair import Message
client.memories.add(
    group_id="user:alice",
    messages=[Message(content="I love Python", role_type="user")],
)

# Tripartite cross-graph search
client.search.tripartite(
    query="What does Alice prefer?",
    user_id="user:alice",
    org_id="org:acme",
)

# Explicit triplets
client.facts.add_triplet(
    group_id="user:alice",
    source_name="Alice",
    target_name="Python",
    edge_name="prefers",
    fact="Alice prefers Python",
)

See DESIGN.md for full architecture and CHANGELOG.md for release notes.

License

MIT License — see LICENSE for details.