Production Checklist

# Good: loaded from secrets manager at runtime
import boto3

def get_bootstrap_token():
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="synap/bootstrap-token")
    return response["SecretString"]

sdk = MaximemSynapSDK(
    instance_id=os.environ["SYNAP_INSTANCE_ID"],
    bootstrap_token=get_bootstrap_token()
)

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,
        stream_idle=60.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
)

result = await admin_client.config.validate(
    instance_id="inst_xxx",
    config_yaml=config_content,
    dry_run=True
)
assert result.valid, f"Config validation failed: {result.errors}"

storage:
  retention:
    max_memory_age_days: 365    # Not 0 (unlimited) unless justified

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
context = await sdk.conversation.context.fetch(
    conversation_id=conv_id,
    search_query=[query],
    mode="fast"           # ~50-100ms
)

# Background analysis: use accurate mode
context = await sdk.conversation.context.fetch(
    conversation_id=conv_id,
    search_query=[query],
    mode="accurate"       # ~200-500ms, better precision
)

# Good: single batch call
await sdk.memories.batch_create(
    documents=[
        {"document": doc1, "document_type": "document", "user_id": "user_1"},
        {"document": doc2, "document_type": "email", "user_id": "user_1"},
        {"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="injection-ready"  # Ready to insert into your LLM prompt
)

stats = await sdk.cache.stats()
print(f"Cache entries: {stats.entry_count}")
print(f"Hit rate: {stats.hit_rate:.1%}")