Build Your First Integration

SYNAP_INSTANCE_ID=inst_a1b2c3d4e5f67890
SYNAP_BOOTSTRAP_TOKEN=your-bootstrap-token-here
OPENAI_API_KEY=sk-your-openai-key-here

from maximem_synap import MaximemSynapSDK, SDKConfig
import os

sdk = MaximemSynapSDK(
    instance_id=os.environ["SYNAP_INSTANCE_ID"],
    bootstrap_token=os.environ["SYNAP_BOOTSTRAP_TOKEN"],
    config=SDKConfig(
        cache_backend="sqlite",
        log_level="INFO"
    )
)

async def init():
    """Bootstrap credentials and establish connection to Synap Cloud."""
    await sdk.initialize()

async def cleanup():
    """Flush pending operations and close connections."""
    await sdk.shutdown()

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel
from openai import AsyncOpenAI
from startup import sdk, init, cleanup

# --- Lifespan Management ---

@asynccontextmanager
async def lifespan(app):
    """Initialize Synap SDK on startup, shut down on exit."""
    await init()
    print("Synap SDK initialized. Ready to serve requests.")
    yield
    await cleanup()
    print("Synap SDK shut down cleanly.")

app = FastAPI(
    title="Synap Chatbot",
    description="A memory-enabled chatbot powered by Synap",
    lifespan=lifespan
)
openai_client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# --- Request/Response Models ---

class ChatRequest(BaseModel):
    message: str
    conversation_id: str
    user_id: str
    customer_id: str | None = None

class ChatResponse(BaseModel):
    response: str
    memories_used: int

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    # -------------------------------------------------------
    # Step 1: Retrieve relevant memories for this conversation
    # -------------------------------------------------------
    context = await sdk.conversation.context.fetch(
        conversation_id=req.conversation_id,
        search_query=[req.message],
        max_results=5,
        types=["facts", "preferences"],
        mode="fast"
    )

    # -------------------------------------------------------
    # Step 2: Build system prompt with memory context
    # -------------------------------------------------------
    memory_lines = []
    for fact in context.facts:
        memory_lines.append(
            f"- {fact.content} (confidence: {fact.confidence:.0%})"
        )
    for pref in context.preferences:
        memory_lines.append(f"- User preference: {pref.content}")

    memory_block = "\n".join(memory_lines) if memory_lines else (
        "No prior context available."
    )

    system_prompt = f"""You are a helpful assistant with memory.

Known information about this user:
{memory_block}

Use this context naturally in your responses. Do not explicitly mention
that you are reading from a memory system -- just be naturally informed."""

    # -------------------------------------------------------
    # Step 3: Call the LLM
    # -------------------------------------------------------
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": req.message}
        ],
        temperature=0.7,
        max_tokens=1024
    )
    assistant_message = response.choices[0].message.content

    # -------------------------------------------------------
    # Step 4: Ingest this conversation turn for future memory
    # -------------------------------------------------------
    await sdk.memories.create(
        document=f"User: {req.message}\nAssistant: {assistant_message}",
        document_type="ai-chat-conversation",
        user_id=req.user_id,
        customer_id=req.customer_id,
        mode="fast",
        metadata={
            "conversation_id": req.conversation_id,
            "source": "chatbot-tutorial"
        }
    )

    return ChatResponse(
        response=assistant_message,
        memories_used=len(memory_lines)
    )

@app.get("/health")
async def health():
    try:
        stats = await sdk.cache.stats()
        return {
            "status": "healthy",
            "synap_connected": True,
            "cache_entries": stats.entry_count
        }
    except Exception as e:
        return {
            "status": "degraded",
            "synap_connected": False,
            "error": str(e)
        }