Skip to main content
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.

How domain schemas work with the SDK

Domain schemas are configured on the tenant, not in SDK code. That means this code stays the same:
from memoryos import Memory

client = Memory(api_key="mem_...")

client.add(
    messages=messages,
    external_user_id="customer-123",
)

result = client.get(
    query="How should the assistant answer?",
    external_user_id="customer-123",
)
What changes is the tenant setting:
Tenant settingWhat add() doesWhat get() returns
General EngineGeneric memory extractionGeneric prompt-ready memory
EdTech SchemaGeneric memory + EdTech overlayGeneric memory + tutoring/student context
Customer Support SchemaGeneric memory + Support overlayGeneric memory + support/customer context
For Support, system_prompt_addition can include current issue, support history, resolution preference, sentiment risk, and safety rules. Your own tools still provide live truth such as order status, invoice status, refunds, or ticket updates. Use the same SDK methods for all three modes.

Memory

from memoryos import Memory

client = Memory(api_key="mem_...")

add()

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

Parameters

ParameterTypeRequiredNotes
messages`list[dict[str, str]]list[ConversationMessage]`YesAt least one message; roles must be user, assistant, or system
external_user_idstrYesEnd-user identifier inside your tenant
agent_id`strNone`NoOptional agent identifier
metadata`dict[str, Any]None`NoOptional metadata attached to the ingestion job

Return fields

FieldTypeMeaning
job_id`strNone`Extraction job id when queued
statusstrCurrent add outcome
blocked_reason`strNone`Reason when blocked
retry_after_seconds`intNone`Retry hint for rate-limit blocks
budget_remaining_pct`floatNone`Remaining quota after the request estimate
quota_mode`“FULL""PASSTHROUGH""DEGRADED_RETRIEVE""BLOCKED”`Quota envelope mode from response headers
processing_eta_seconds`intNone`Queue ETA when ingestion is delayed
processing_status`“normal""delayed”`Background ingestion health
circuit_status`“HEALTHY""DEGRADED""CRITICAL”`Platform dependency status

Status outcomes

status valueMeaningTypical action
queuedRequest accepted and extraction job queuedStore job_id if you want to track job state
passthroughTenant is in passthrough mode, so no memory write happensContinue your app without memory persistence
L1Blocked by per-user rate limitRetry after retry_after_seconds
L2Blocked by low-quality inputImprove message quality before retrying
L3Blocked as semantically duplicateAvoid sending repeated near-identical content
L4Blocked by budget governanceUpgrade plan or wait for reset
blockedInternal quality-gate failure fallbackRetry later and monitor logs

get()

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

Parameters

ParameterTypeRequiredNotes
querystrYesNatural-language retrieval query
external_user_idstrYesEnd-user identifier inside your tenant
limitintNoDefault 10
categories`list[str]None`NoOptional category filter
time_filter_days`intNone`NoReturn only memories from the last N days
format`“bullets""json""xml”`NoFormat for system_prompt_addition
context_max_tokensintNoPrompt context token budget, default 500

Return fields

FieldTypeMeaning
itemslist[MemoryResult]Retrieved memories
cachedboolWhether the result came from the hot cache
system_prompt_additionstrPrompt-ready context block
context_token_countintTokens used by system_prompt_addition
memories_from_hot_tierintNumber of returned memories served from hot tier
quota_mode`“FULL""PASSTHROUGH""DEGRADED_RETRIEVE""BLOCKED”`Quota envelope mode
is_passthroughboolTrue when you should skip memory context
is_degradedboolTrue when retrieval is degraded
circuit_status`“HEALTHY""DEGRADED""CRITICAL”`Platform dependency status

Degradation handling pattern

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

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

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. EdTech example:
profile = client.get_edtech_profile("student_123")

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

if profile:
    print(profile.weak_topics[:3])
Async version:
profile = await client.get_edtech_profile("student_123")

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)
Returns a MemoryPage with:
  • items
  • next_cursor
  • limit
  • total

export()

bundle = client.export(external_user_id="customer-123")
export() maps to GET /v1/users/me/export and returns a MemoryExport bundle containing:
  • user
  • memories
  • api_keys
  • agents
export() currently maps to the user export endpoint, not a tenant-wide export endpoint.

get_usage() and delete_user()

The current Python SDK does not expose get_usage() or delete_user() methods. Use the REST API directly for those operations:
  • tenant usage: GET /v1/tenant/usage
  • current user deletion: DELETE /v1/users/me
  • tenant proxy-user deletion: DELETE /v1/users/{external_user_id}

AsyncMemory

AsyncMemory has the same method surface as Memory, but every method is async. Use it when your backend is already async, for example FastAPI, async workers, or high-concurrency chat services. It is not a separate memory engine and it works with General, EdTech, and Support tenants.
import asyncio

from memoryos import AsyncMemory


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

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

        print(add_result.status)
        print(memories.system_prompt_addition)


asyncio.run(main())

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
Import it from the dedicated module:
from memoryos.universal import UniversalMemory

client = UniversalMemory(
    agent_api_key="agent_sk_...",
    uui_token="uui_...",
    base_url="https://api.memoryos.io",
)
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"],
)
categories preselects the memory categories for this consent link. Users can still add or remove categories before approving. If you omit it, MemoryOS uses the global agent’s default categories. Pass a callback only if your app needs to update its own UI automatically after consent:
consent_url = UniversalMemory.consent_url(
    agent_id="your_global_agent_id",
    redirect_uri="https://yourapp.com/integrations/memoryos/callback",
    state="secure_random_state_token",
    categories=["preference", "fact"],
)

add()

from memoryos.universal import UniversalMemory

client = UniversalMemory(
    agent_api_key="agent_sk_...",
    uui_token="uui_...",
)

result = client.add(
    messages=[
        {
            "role": "user",
            "content": "Please remember that I am strongest in FastAPI, PostgreSQL, and retrieval systems."
        }
    ],
    metadata={"source": "consent-demo"},
)

Parameters

ParameterTypeRequiredNotes
messageslist[dict[str, str]] | list[ConversationMessage]YesSame message shape as tenant-scoped add()
metadatadict[str, Any] | NoneNoOptional JSON metadata

Behavior

  • writes to /v1/universal/memories/add
  • uses the active UUI grant to determine whether the agent may write
  • returns 403 UAT_002 if the grant is read-only
  • stores results in the universal memory collection, not the tenant-scoped memory collection

get()

from memoryos.universal import UniversalMemory

client = UniversalMemory(
    agent_api_key="agent_sk_...",
    uui_token="uui_...",
)

result = client.get(
    query="What do you know about my technical strengths?",
    limit=5,
)

print(result.permission_status)
print(result.categories_available)
for item in result.items:
    print(item.content)

Parameters

ParameterTypeRequiredNotes
querystrYesNatural-language retrieval query
limitintNoDefault 10

Additional return fields

In addition to the normal RetrieveResult fields, universal retrieval adds:
FieldTypeMeaning
permission_statusstr | NonePermission-related explanation such as no_grant_for_user
categories_availablelist[str]Categories this agent is currently allowed to read

Empty result behavior

If the user has not granted this agent access, MemoryOS returns:
  • items = []
  • permission_status = "no_grant_for_user"
  • categories_available = []

Lifecycle

from memoryos.universal import UniversalMemory

with UniversalMemory(
    agent_api_key="agent_sk_...",
    uui_token="uui_...",
) as client:
    result = client.get("What do you know about this user?", limit=5)
    print(len(result.items))