Python SDK

pip install memoryo-sdk

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

import os
from memoryos import Memory

client = Memory(api_key=os.environ["MEMORYOS_API_KEY"])

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

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

result = client.add(
    external_user_id="customer-123",
    conversation_id="support-884",
    messages=[
        {
            "role": "assistant",
            "content": (
                "I can remember this. Proposed memory: "
                "User prefers a one-line diagnosis before numbered troubleshooting steps."
            ),
            "external_turn_id": "support-884:turn-13",
            "source_kind": "assistant_output",
            "is_memory_proposal": True,
            "proposed_memory": {
                "content": "User prefers a one-line diagnosis before numbered troubleshooting steps.",
                "category": "preference",
            },
        }
    ],
)
if result.proposal_ids:
    print("Registered proposal:", result.proposal_ids[0])
else:
    print("No proposal registered:", result.status, result.blocked_reason)

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",
            }
        ],
    ),
)

job = client.wait_for_job(result.job_id, timeout=120, poll_interval=1)

print(job.status)
print(job.memories_created)
print(job.created_memory_ids)
print(job.pending_candidates_buffered)
print(job.extraction_metadata)

result = client.get(
    query="How should I answer this user?",
    external_user_id="customer-123",
    limit=5,
    categories=["preference", "goal"],
    as_of=None,  # or a timezone-aware datetime for historical retrieval
)

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

if result.clarification:
    # Render result.clarification.question in your existing chat and map the
    # user's selection to one of: A, B, both, neither.
    resolution = client.answer_clarification(
        result.clarification.id,
        external_user_id="customer-123",
        answer="B",
        free_text="The Scale plan is current.",
    )

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)

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)

profile = client.get_edtech_profile("student_123")

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

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())

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

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

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

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

import os
from memoryos.universal import UniversalMemory

client = UniversalMemory(
    agent_api_key=os.environ["MEMORYOS_AGENT_API_KEY"],
    uui_token=os.environ["MEMORYOS_UUI_TOKEN"],
)

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

universal.add(
    [{"role": "user", "content": "I prefer short technical answers."}],
    metadata={"source": "chat"},
    idempotency_key="conversation-event-123",
)

result = universal.get(
    "How should I personalize this answer?",
    limit=5,
    format="json",
    context_max_tokens=900,
)