Production Checklist

# Good: loaded from secrets manager at runtime
import boto3

def get_api_key():
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="synap/api-key")
    return response["SecretString"]

sdk = MaximemSynapSDK(
    api_key=get_api_key()
)

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

config = SDKConfig(
    log_level="WARNING"    # Not "DEBUG" or "INFO" in production
)

config = SDKConfig(
    timeouts=TimeoutConfig(
        connect=5.0,
        read=30.0,
        write=10.0
    )
)

config = SDKConfig(
    retry_policy=RetryPolicy(
        max_attempts=3,
        backoff_base=1.0,
        backoff_max=10.0,
        backoff_jitter=True     # Always enable jitter in production
    )
)

config = SDKConfig(
    cache_backend="sqlite"     # Not None
)

from maximem_synap.errors import (
    SynapError,
    NetworkTimeoutError,
    RateLimitError,
    ServiceUnavailableError,
    InvalidInputError,
    AuthenticationError,
)

try:
    context = await sdk.conversation.context.fetch(
        conversation_id=conv_id,
        search_query=[query],
        mode="fast"
    )
except (NetworkTimeoutError, ServiceUnavailableError) as e:
    # Transient: retry or fall back to no-memory mode
    logger.warning(
        "Synap unavailable (transient), proceeding without memory: %s "
        "(correlation_id=%s)", e, e.correlation_id
    )
    context = None
except RateLimitError as e:
    # Transient: respect retry_after
    logger.warning(
        "Rate limited, retry after %s seconds (correlation_id=%s)",
        e.retry_after_seconds, e.correlation_id
    )
    context = None
except InvalidInputError as e:
    # Permanent: fix the request
    logger.error("Invalid request to Synap: %s", e)
    raise
except AuthenticationError as e:
    # Permanent: credentials issue
    logger.critical("Synap auth failed: %s", e)
    raise

except SynapError as e:
    logger.error(
        "Synap error: %s (correlation_id=%s)",
        e, e.correlation_id
    )

async def get_memory_context(sdk, conversation_id, query):
    """Retrieve memory context, returning None if unavailable."""
    try:
        return await sdk.conversation.context.fetch(
            conversation_id=conversation_id,
            search_query=[query],
            max_results=5,
            mode="fast"
        )
    except SynapError as e:
        logger.warning(
            "Memory retrieval failed, proceeding without context: %s",
            e
        )
        return None

# In your chat handler:
context = await get_memory_context(sdk, conv_id, user_message)

if context and context.facts:
    # Build enriched prompt with memories
    system_prompt = build_prompt_with_memories(context)
else:
    # Fall back to generic prompt, your app still works
    system_prompt = build_generic_prompt()

except RateLimitError as e:
    await asyncio.sleep(e.retry_after_seconds)
    # Retry the operation

# Real-time chat: use fast mode
# fast = single-pass vector + graph search
context = await sdk.conversation.context.fetch(
    conversation_id=conv_id,
    search_query=[query],
    mode="fast"
)

# Background analysis: use accurate mode
# accurate = same vector + graph search, plus LLM subquery decomposition + reranking
context = await sdk.conversation.context.fetch(
    conversation_id=conv_id,
    search_query=[query],
    mode="accurate"
)

from maximem_synap import CreateMemoryRequest

# Good: single batch call
await sdk.memories.batch_create(
    documents=[
        CreateMemoryRequest(document=doc1, document_type="document", user_id="user_1"),
        CreateMemoryRequest(document=doc2, document_type="email", user_id="user_1"),
        CreateMemoryRequest(document=doc3, document_type="pdf", user_id="user_2"),
    ],
    fail_fast=False    # Continue processing even if one document fails
)

# Avoid: N sequential calls
for doc in documents:
    await sdk.memories.create(document=doc, ...)  # Slower, more API calls

result = await sdk.conversation.context.compact(
    conversation_id=conv_id,
    strategy="adaptive",      # Automatically adjusts compression level
    target_tokens=2000
)

compacted = await sdk.conversation.context.get_compacted(
    conversation_id=conv_id,
    format="structured"       # Also valid: "narrative", "bullet_points"
)

stats = sdk.cache.stats()
print(f"Cache entries: {stats['total_entries']}")
print(f"Cache size: {stats['total_bytes']} bytes")
print(f"Backends: {stats['backends']}")