Voice SDK reference (v0.3)

MemoAirVoiceClient

High-level async client. Construct it once at process boot, pass a per-call user={id,name,metadata} object to search_memory / save_response, and call aclose() on shutdown. All recall calls are local-loopback; create_index and list_indexes hit MemoAir cloud.

from memoair_voice import MemoAirVoiceClient

__init__

MemoAirVoiceClient(
    *,
    api_key: str,
    project_id: str,
    agent_id: str,
    cloud_base_url: str = "https://backend.memoair.space",
    max_concurrent_users: int = 20,
    runtime_idle_ttl_s: int = 300,
    runtime_port_range: tuple[int, int] = (7878, 7977),
) -> MemoAirVoiceClient

Raises ValueError if any of api_key, project_id, or agent_id are empty strings. Construct the client once at boot; pass user={...} per call so the internal runtime pool can fan out across end-users.

search_memory()

Per-turn local recall across up to four lanes. Returns a composed contextText ready to splice into your LLM prompt. Never hits the cloud; never blocks longer than timeout_ms.

async def search_memory(
    self,
    query: str,
    *,
    user: dict,
    user_id: str | None = None,  # compatibility alias for user["id"]
    lanes: list[str] | None = None,
    intent: str = "answer_current_user",
    top_k: dict[str, int] | None = None,
    timeout_ms: int = 250,
) -> SearchResult

Returns: SearchResult (see Result Types below) — always returns; falls back to empty contextText on degraded lanes rather than raising.

result = await client.search_memory(
    "what timezone does the user prefer?",
    user={
        "id": "caller_42",
        "name": "Alex",
        "metadata": {"plan": "premium"},
    },
    lanes=["profile", "working", "permanent", "org"],
    timeout_ms=250,
)
print(result.contextText)
print("hit sources:", result.sources)

save_response()

Persist a completed (user, assistant) turn. Forwards to the local runtime which appends to the session JSONL and enqueues cloud sync. Best-effort: transient runtime errors are swallowed so the audio path is never blocked.

async def save_response(
    self,
    *,
    user_text: str,
    assistant_text: str,
    user: dict,
    user_id: str | None = None,
    turn_id: str | None = None,
    tool_calls: list[dict] | None = None,
    metadata: dict | None = None,
) -> str | None

Returns: the runtime's localEventId on success, or None if the runtime dropped/swallowed the call.

await client.save_response(
    user_text="What time is standup?",
    assistant_text="10:30 IST.",
    user={"id": "caller_42", "name": "Alex"},
    metadata={"framework": "livekit", "interrupted": False},
)

create_index()

Cloud-side: build or append an org index. Build-or-append semantics — the first call with a given name CREATES the index, subsequent calls APPEND documents (or replace by matching id). Refreshes the local runtime's org projection on success.

async def create_index(
    self,
    name: str,
    documents: list[dict],
    *,
    metadata: dict | None = None,
) -> IndexBuildResult

Returns: IndexBuildResult(index_name, chunk_count, version).

Raises: MemoAirVoiceMemoryError with status_code=413 if the payload exceeds the 100-doc / 1 MB cap — use the MemoAir dashboard for bulk ingest beyond that.

result = await client.create_index(
    "agent-memory",
    documents=[
        {"id": "returns", "text": "30-day returns with receipt.", "metadata": {"topic": "returns"}},
        {"id": "shipping", "text": "Express ships in 1-2 days.", "metadata": {"topic": "shipping"}},
    ],
)
print(f"Indexed {result.chunk_count} chunks, version={result.version}")

list_indexes()

Cloud-side: list all org indexes in the current project. No arguments.

async def list_indexes(self) -> list[dict]

Returns: raw cloud response list (each item has name, version, chunkCount, createdAt, …).

session() / aclose()

Use session(user=...) when many calls happen back-to-back for the same end-user and you want to pin the runtime handle. Always call aclose() on process shutdown so the pool flushes and terminates every runtime.

session = await client.session(user={"id": "caller_42", "name": "Alex"})
async with session as s:
    result = await s.search_memory("what did we discuss?")
    await s.save_response(user_text="what did we discuss?", assistant_text="...")
 
await client.aclose()

MemoAirLiveKitAgent

Drop-in livekit.agents.Agent subclass shipped by memoair-livekit. Wires every memory lifecycle hook for you. LiveKit dispatch is 1:1 (one process per call), so this class keeps user_id at construction time and pins its internal pool to that single end-user. See LiveKit integration for an end-to-end agent.

from memoair_livekit import MemoAirLiveKitAgent

__init__

MemoAirLiveKitAgent(
    *,
    api_key: str,
    project_id: str,
    agent_id: str,
    user_id: str,
    instructions: str | None = None,
    search_timeout_ms: int = 250,
    cloud_base_url: str = "https://backend.memoair.space",
    **agent_kwargs,
) -> MemoAirLiveKitAgent

Lifecycle hooks

Called automatically by LiveKit's pipeline. You don't invoke these yourself — they're documented so you know what the subclass does.

• on_enter() — opens the MemoAir runtime session (load_index equivalent).
• on_user_turn_completed(turn_ctx, new_message) — runs search_memory with a timeout, then splices a SYSTEM message into turn_ctx.items before the user's message (LiveKit #5053 ordering workaround). Search failures are logged at WARNING and swallowed — the LLM never blocks on memory I/O.
• on_agent_response_completed(chat_ctx, new_message) — persists the completed turn via save_response. Required wiring: LiveKit's base Agent does not call this; forward LiveKit's conversation_item_added event for assistant items into this method (see the LiveKit page for the exact handler).
• on_exit() — idempotent close of the runtime session.

MemoAirVoiceMemory (low-level)

Raw async wrapper around the loopback runtime. Use when you want explicit control over session lifecycle, or when building a framework adapter MemoAir doesn't ship yet. Most code should use MemoAirVoiceClient instead.

from memoair_voice import MemoAirVoiceMemory

Constructor

MemoAirVoiceMemory(
    *,
    project_id: str,
    user_id: str,
    api_key: str | None = None,
    internal_token: str | None = None,
    runtime_url: str = "http://127.0.0.1:7878",
    runtime_token: str | None = None,
    session_id: str | None = None,
    satham: dict | None = None,
    auto_start_runtime: bool = True,
    runtime_binary: str | None = None,
    cloud_base_url: str = "https://backend.memoair.space",
    storage_root: str | None = None,
    # plus runtime_version, runtime_repo, runtime_cache_dir, client, runtime_manager
)

Either api_key or internal_token must be set if the runtime needs to bootstrap from cloud. The runtime binary itself is auto-resolved (downloaded from ghcr.io/memoair/memoair-runtime on first use) when auto_start_runtime=True.

Methods

• await memory.start_session(metadata=None, timeout_s=5.0) -> dict — open a voice session against the runtime. Returns the raw runtime body ({ sessionId, bootstrap, paths }).
• await memory.end_session(reason="completed", timeout_s=5.0) -> dict — close the session and flush.
• memory.search_memory_tool(*, search_memory_timeout_ms=None) -> MemoryTool — return a registerable LLM tool.
• await memory.after_turn(user_text, assistant_text, turn_id, tool_calls=None, metadata=None, timeout_s=1.0) -> str | None — persist a completed turn. Note this is the raw form MemoAirVoiceClient.save_response wraps; here turn_id is required.
• memory.to_openai_function() -> dict — OpenAI Chat Completions / Responses tool schema.
• memory.to_livekit_tool() -> FunctionTool — requires livekit-agents installed.
• memory.to_pipecat_tool() -> dict — { name, schema, handler }. Requires pipecat-ai installed.

MemoAirVoiceMemorySync (sync mirror)

Synchronous variant of MemoAirVoiceMemory with identical method names — drop the async/await keywords. Use with non-async agent frameworks. Same to_openai_function / to_livekit_tool / to_pipecat_tool adapters.

MemoryTool

Async callable returned by search_memory_tool(). Calling it forwards the query to the local runtime and returns the composed contextText string. Register tool.schema() with your LLM framework, or call await tool(query) directly.

SearchResult

@dataclass(frozen=True)
class SearchResult:
    contextText: str             # composed system-message string ready to splice
    profile: dict | None         # current profile snapshot, or None
    working: list[dict]          # working-brief hits
    permanent: list[dict]        # per-user permanent hits
    org: list[dict]              # workspace-shared org hits
    sources: list[dict]          # unioned source metadata for citations
    trace: dict                  # per-lane latency + degraded flag

IndexBuildResult

@dataclass(frozen=True)
class IndexBuildResult:
    index_name: str
    chunk_count: int       # number of embedding chunks created/appended
    version: int | None    # monotonic version, bumped per call

IndexLoadResult

@dataclass(frozen=True)
class IndexLoadResult:
    session_id: str
    profile_version: int | None
    permanent_manifest_version: int | None
    org_manifest_version: int | None
    scopes_loaded: list[str]

MemoAirVoiceMemoryError

Raised on hard runtime errors (non-transient 4xx, timeouts, unparseable responses) and on cloud-side index failures. The runtime's { "error": { code, message, details } } envelope is parsed into the exception attributes.

class MemoAirVoiceMemoryError(Exception):
    code: str                       # e.g. "runtime.identity_mismatch"
    message: str
    status_code: int | None         # HTTP status if applicable
    details: dict                   # extra context from the envelope

Common codes: runtime.bad_request, runtime.session_not_found, runtime.bootstrap_failed, runtime.projection_unavailable, runtime.identity_mismatch, runtime.timeout, runtime.internal.