Context, End to End

record_message / memories.create
        │  ingest
        ▼
  multi-stage pipeline ──► vector store + graph store
        │  extract                 │  store
        ▼                          ▼
   structured memories      context.fetch() ──► ranked context for the next turn
                                   │  retrieve
                                   ▼
                          your agent's prompt

import uuid
from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK(api_key="synap_your_key_here")
await sdk.initialize()

# One UUID per conversation, reused across every turn
conversation_id = str(uuid.uuid4())

await sdk.conversation.record_message(
    conversation_id=conversation_id,
    role="user",
    content="I prefer dark mode and concise answers.",
    user_id="user_alice",
    # customer_id="customer_acme",  # required on B2B; omit on B2C (auto-resolved)
)

await sdk.conversation.record_message(
    conversation_id=conversation_id,
    role="assistant",
    content="Got it — I'll keep answers short and assume dark mode.",
    user_id="user_alice",
)

Turn 1:  User: "What's our current API rate limit?"
         Assistant: "Your current rate limit is 1,000 requests per minute."

Turn 2:  User: "Can we increase that for our enterprise plan?"
         Assistant: "Yes, enterprise plans support up to 10,000 req/min..."

Turn 3:  User: "What about burst handling?"
         Assistant: "Burst allowances provide a 2x multiplier..."
  ...

context = await sdk.conversation.context.fetch(
    conversation_id=conversation_id,
    search_query=["migration timeline"],
)
# context.facts / context.preferences / context.episodes hold the
# long-term memories retrieved for this turn; the short-term transcript
# supplies the in-session continuity.

USER scope     →  Alice's personal memories (highest priority)
CUSTOMER scope →  Acme Corp's shared knowledge
CLIENT scope   →  your application's product knowledge
WORLD scope    →  global domain knowledge (lowest priority)

context = await sdk.user.context.fetch(
    user_id="user_alice",
    # customer_id="customer_acme",  # B2B only
    search_query=["project timeline", "Q2 deliverables"],
    types=["facts", "temporal_events"],  # restrict to specific memory types
)

# context.facts → "Q2 roadmap includes API v3 launch and dashboard redesign"
# context.temporal_events → "API v3 launch deadline: June 15"

await sdk.memories.create(
    document="""
    Acme Corp Engineering Handbook
    - All services must use Python 3.11 or later
    - Production deployments: Tuesdays and Thursdays, 10am-2pm PT
    - Hotfix deployments require VP Engineering approval
    """,
    document_type="document",
    customer_id="customer_acme",
    # No user_id — shared across all users at this customer
)

from maximem_synap import CreateMemoryRequest

# Single doc — no user_id or customer_id = CLIENT scope
await sdk.memories.create(
    document="Our standard SLA guarantees 99.9% uptime...",
    document_type="document",
    document_id="doc_sla_v2",
)

# Bulk load product documentation
documents = [
    CreateMemoryRequest(document=open("docs/api-reference.md").read(), document_type="document"),
    CreateMemoryRequest(document=open("docs/changelog-v3.md").read(), document_type="document"),
]
await sdk.memories.batch_create(documents=documents)

User Query: "What is the refund policy?"
         │
         ▼
  1. USER scope     → "Alice has a VIP 60-day return window"   ← highest priority
  2. CUSTOMER scope → "Acme Corp negotiated 45-day returns"    ← medium priority
  3. CLIENT scope   → "Standard refund policy: 30 days"        ← lowest priority
         │
         ▼
  Agent receives all three, ranked. A well-designed prompt prefers
  the most specific (user) answer over the general (org) baseline.

import asyncio
import uuid

conversation_id = str(uuid.uuid4())  # reuse this conversation's UUID

# Kick off compaction (fire-and-forget — returns a trigger handle)
trigger = await sdk.conversation.context.compact(
    conversation_id=conversation_id,
    strategy="adaptive",
    target_tokens=2000,
)
print(f"Compaction {trigger.compaction_id} status: {trigger.status}")

# Poll until the run completes
while True:
    status = await sdk.conversation.context.get_compaction_status(
        conversation_id=conversation_id,
    )
    if status.status in ("completed", "failed"):
        break
    await asyncio.sleep(2)

# Read the compacted result
result = await sdk.conversation.context.get_compacted(conversation_id=conversation_id)
print(f"Compressed {result.original_token_count} -> {result.compacted_token_count} tokens")
print(f"Quality: {result.validation_score:.2f}, passed: {result.validation_passed}")
print(f"Facts: {result.facts}")
print(f"Current state: {result.current_state}")