Python SDK

Reference for the MemoryOS Python SDK, including tenant-scoped memory clients and the universal Memory Passport client.

Install the Python SDK:

pip install memoryos

Clients

• Sync client: Memory
• Async client: AsyncMemory
• Universal cross-agent client: UniversalMemory

from memoryos import AsyncMemory, Memory
from memoryos.universal import UniversalMemory

There is no separate SDK for domain schemas. If your tenant enables EdTech or Support, add() and get() remain the main integration path. Optional domain helper methods expose structured profile data for dashboards.

Tenant-scoped memory

Use Memory from your backend with a tenant API key.

For solo builders: simple mode

Start here if you are building an MVP, small SaaS app, single chatbot, or solo product. You do not need event_id, run_id, service writers, or authority rules. MemoryOS creates safe internal source metadata automatically.

from memoryos import Memory

client = Memory(api_key="mem_live_xxx")

add()

add() queues a conversation for extraction. It returns quickly; extraction runs in the background.

result = client.add(
    messages=[{"role": "user", "content": "I prefer Python examples."}],
    external_user_id="customer-123",
    agent_id="support-bot",
    metadata={"source": "chat"},
)

print(result.status)
print(result.job_id)

For multi-service companies: source-aware mode

Use this when Billing, Support, CRM, Product, or another backend service can write facts about the same user. Source metadata enables auditability, deduplication, conflict handling, and authority rules.

Use Memory.source(...) so your app does not have to manually generate every ID while testing.

result = client.add(
    external_user_id="customer-123",
    messages=[
        {"role": "user", "content": "What plan am I using?"},
        {"role": "assistant", "content": "Your current subscription plan is Growth."},
    ],
    source=Memory.source(
        "billing-service",
        event_id="billing-plan-2026-06-30-001",
        scope={"workspace_id": "ws_123"},
        evidence=[
            {
                "source_type": "subscription_record",
                "reference": "billing/subscriptions/ws_123",
            }
        ],
    ),
)

For testing, one tenant API key plus Memory.source("billing-service") is enough. For production multi-service traffic, register billing-service once from the Tenant Dashboard and bind a dedicated API key/service writer.

Parameters

Parameter	Type	Required	Notes
messages	list[dict[str, str]] | list[ConversationMessage]	Yes	Roles must be user, assistant, or system
external_user_id	str	Yes	End-user identifier inside your tenant
agent_id	str | None	No	Optional agent identifier
metadata	dict[str, Any] | None	No	Optional metadata attached to the ingestion job
idempotency_key	str | None	No	Prevents duplicate jobs when your app retries
source	MemorySource | dict | None	No	Provenance source block for multi-service governance

Return fields

Field	Type	Meaning
job_id	str | None	Extraction job id when queued
status	str	queued, passthrough, L1, L2, L3, L4, or blocked
blocked_reason	str | None	Reason when blocked
nothing_to_extract	bool	Request passed the gate but had no durable memory
retry_after_seconds	int | None	Retry hint for rate-limit blocks
budget_remaining_pct	float | None	Remaining quota estimate
quota_mode	str	Current quota mode
processing_eta_seconds	int | None	Queue ETA when ingestion is delayed
processing_status	str	Background ingestion health
circuit_status	str	Platform dependency status

Check extraction job status

The Python SDK does not yet wrap job polling. Use the REST endpoint with the same API key when you need exact extraction results.

import requests

job = requests.get(
    f"{client.base_url}/v1/memories/jobs/{result.job_id}",
    headers={"Authorization": f"ApiKey {client.api_key}"},
    timeout=20,
).json()["data"]

print(job["status"])
print(job["memories_created"])
print(job["pending_candidates_buffered"])
print(job["extraction_metadata"])

pending_candidates_buffered > 0 means MemoryOS kept a weak signal for reinforcement instead of dropping it.

get()

result = client.get(
    query="How should I answer this user?",
    external_user_id="customer-123",
    limit=5,
    categories=["preference", "goal"],
)

prompt_addition = result.system_prompt_addition if result.has_context else ""

Parameters

Parameter	Type	Required	Notes
query	str	Yes	Natural-language retrieval query
external_user_id	str	Yes	End-user identifier inside your tenant
limit	int	No	Default 10
categories	list[str] | None	No	Optional category filter
agent_id	str | None	No	Optional agent filter
time_filter_days	int | None	No	Return only memories from the last N days
format	"bullets" | "json" | "xml"	No	Format for system_prompt_addition
context_max_tokens	int	No	Prompt context token budget, default 500

Return fields

Field	Type	Meaning
retrieval_id	str | None	Use this when sending feedback
items	list[MemoryResult]	Retrieved memories
cached	bool	Whether the result came from the hot cache
system_prompt_addition	str	Prompt-ready context block
context_token_count	int	Tokens used by system_prompt_addition
memories_from_hot_tier	int	Number of returned memories served from hot tier
quota_mode	str	Quota envelope mode
is_passthrough	bool	True when you should skip memory context
is_degraded	bool	True when retrieval is degraded
circuit_status	str	Platform dependency status

feedback()

Use feedback after retrieval to tell MemoryOS whether the memory helped. This improves lifecycle scoring and can queue retrospective extraction after user corrections.

result = client.get(
    query="What language should I use for this user?",
    external_user_id="customer-123",
)

# Call your LLM here using result.system_prompt_addition.

if result.retrieval_id:
    feedback = client.feedback(
        retrieval_id=result.retrieval_id,
        outcome="used_successfully",
        used_memory_ids=[item.id for item in result.items],
        agent_confidence=0.86,
        metadata={"agent_id": "support-bot"},
    )

    print(feedback.feedback_id)

When a user corrects the answer, send the correction text:

feedback = client.feedback(
    retrieval_id=result.retrieval_id,
    outcome="user_corrected",
    used_memory_ids=[item.id for item in result.items],
    correction="Actually I prefer Hindi explanations for billing questions, not English.",
    agent_confidence=0.2,
    metadata={"agent_id": "support-bot"},
)

print(feedback.correction_job_id)

If correction_job_id is present, MemoryOS queued an async retrospective extraction pass. Do not block your user flow while that job runs.

Domain schemas

Domain schemas are configured on the tenant, not in SDK code.

Tenant setting	What add() does	What get() returns
General Engine	Generic memory extraction	Generic prompt-ready memory
EdTech Schema	Generic memory + EdTech overlay	Generic memory + tutoring/student context
Customer Support Schema	Generic memory + Support overlay	Generic memory + support/customer context

Your code stays the same across domain modes. For Support, your own backend tools still provide live truth such as order status, invoice status, refunds, or ticket updates.

Domain profile helpers

For domain-aware tenants, normal get() already includes domain-aware context. Use profile helpers only when your product needs structured UI data.

profile = client.get_edtech_profile("student_123")

if profile and profile.has_exam_context:
    print(profile.exam_name, profile.exam_date)

AsyncMemory

AsyncMemory has the same method surface as Memory, but every method is async.

import asyncio
from memoryos import AsyncMemory

async def main() -> None:
    async with AsyncMemory(api_key="mem_live_xxx") as client:
        add_result = await client.add(
            messages=[{"role": "user", "content": "I prefer weekly summaries."}],
            external_user_id="customer-123",
        )

        result = await client.get(
            query="What does this user prefer?",
            external_user_id="customer-123",
            limit=5,
        )

        if result.retrieval_id:
            await client.feedback(
                retrieval_id=result.retrieval_id,
                outcome="used_successfully",
                used_memory_ids=[item.id for item in result.items],
            )

        print(add_result.status)
        print(result.system_prompt_addition)

asyncio.run(main())

Other tenant methods

delete()

deleted = client.delete(
    memory_id="1c5d5ab6-73c8-4b12-90e9-0f7f9db8db4f",
    external_user_id="customer-123",
    hard_delete=False,
)

Archives by default. Set hard_delete=True to permanently delete.

list()

page = client.list(external_user_id="customer-123", limit=50)

for memory in page.items:
    print(memory.content)

export()

bundle = client.export(external_user_id="customer-123")

export() maps to GET /v1/users/me/export and returns a MemoryExport bundle.

UniversalMemory

UniversalMemory is the cross-agent client for the Memory Passport flow. It uses:

• an agent API key (agent_sk_...)
• a user UUI token (uui_...) for the approved Memory Passport user

from memoryos.universal import UniversalMemory

client = UniversalMemory(
    agent_api_key="agent_sk_live_xxx",
    uui_token="uui_live_xxx",
    base_url="https://api.memoryo.dev",
)

Generate the user consent URL from your tenant app. If you omit redirect_uri, MemoryOS shows a hosted completion page after approval.

consent_url = UniversalMemory.consent_url(
    agent_id="your_global_agent_id",
    state="user_session_id",
    categories=["preference", "goal"],
)

Users can add or remove categories before approving.

Related pages

• POST /v1/memories/add
• POST /v1/memories/retrieve
• Extraction Quality Loop
• Memory Provenance
• Memory Passport
• Cross-agent memory sharing