This page is the framework-and-provider cookbook: drop-in snippets for FastAPI, Flask, Next.js, Django, OpenAI, Anthropic, Google Gemini (free tier), and the Vercel AI SDK. It assumes you’ve already finished the Quickstart and have an API key.If you want a fully-worked walkthrough that builds one application end-to-end with conversation routing, error handling, and graceful degradation, follow First Integration instead, then come back here when you need to swap frameworks or LLM providers.
Overview
This guide walks you through integrating Synap into your application. You will learn how to initialize the SDK in popular Python web frameworks, bridge async/sync execution models, and wire Synap into your LLM provider’s generation pipeline. By the end of this page, your application will have a working memory-augmented agent pattern: retrieve context from Synap, inject it into the LLM prompt, generate a response, and ingest the conversation back into Synap.Framework Integration
- FastAPI
- Flask
- Express (Node)
- Next.js (Vercel AI SDK)
- Django
FastAPI is the most common framework for Synap integrations. Use the
lifespan event for initialization and shutdown, and access the SDK instance from your route handlers.from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request
from maximem_synap import MaximemSynapSDK
@asynccontextmanager
async def lifespan(app: FastAPI):
sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
await sdk.initialize()
app.state.sdk = sdk
try:
yield
finally:
await sdk.shutdown()
app = FastAPI(lifespan=lifespan)
def get_sdk(request: Request) -> MaximemSynapSDK:
"""Dependency that provides the initialized SDK from app.state."""
return request.app.state.sdk
@app.post("/chat")
async def chat(
message: str,
user_id: str,
customer_id: str,
conversation_id: str,
synap: MaximemSynapSDK = Depends(get_sdk)
):
# 1. Retrieve relevant context
context = await synap.conversation.context.fetch(
conversation_id=conversation_id,
search_query=[message]
)
# 2. Build prompt with retrieved memories
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
system_prompt = build_system_prompt(memories)
# 3. Call your LLM (see LLM integration below)
response = await generate_response(system_prompt, message)
# 4. Ingest the turn for long-term memory
await synap.memories.create(
document=f"User: {message}\nAssistant: {response}",
document_type="ai-chat-conversation",
user_id=user_id,
customer_id=customer_id,
metadata={"conversation_id": conversation_id}
)
return {"response": response}
Storing the SDK on
app.state (instead of a module-level global) keeps it scoped to the FastAPI app instance, which makes integration tests trivial: TestClient(app) creates a fresh app with its own SDK, and you can replace app.state.sdk with a mock before each test. The Depends(get_sdk) indirection then lets you override the SDK per-test via app.dependency_overrides[get_sdk] = lambda: mock_sdk.Flask is synchronous by default, so you need to bridge to Synap’s async SDK. Use the app factory pattern and initialize the SDK at startup.
import asyncio
from flask import Flask, request, g
from maximem_synap import MaximemSynapSDK
sdk = None
loop = None
def create_app():
global sdk, loop
app = Flask(__name__)
# Create a dedicated event loop for async operations
loop = asyncio.new_event_loop()
# Initialize SDK synchronously at startup
sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
loop.run_until_complete(sdk.initialize())
@app.teardown_appcontext
def shutdown_sdk(exception=None):
pass # Handled by atexit
import atexit
@atexit.register
def cleanup():
if sdk and loop:
loop.run_until_complete(sdk.shutdown())
loop.close()
@app.route("/chat", methods=["POST"])
def chat():
data = request.json
message = data["message"]
user_id = data["user_id"]
customer_id = data["customer_id"]
conversation_id = data["conversation_id"]
# Bridge async SDK calls to sync Flask handlers
context = loop.run_until_complete(
sdk.conversation.context.fetch(
conversation_id=conversation_id,
search_query=[message]
)
)
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
system_prompt = build_system_prompt(memories)
response = generate_response(system_prompt, message)
loop.run_until_complete(
sdk.memories.create(
document=f"User: {message}\nAssistant: {response}",
document_type="ai-chat-conversation",
user_id=user_id,
customer_id=customer_id,
metadata={"conversation_id": conversation_id}
)
)
return {"response": response}
return app
Using
loop.run_until_complete() in Flask blocks the worker thread during async operations. For production Flask deployments with high concurrency, consider migrating to FastAPI or running the async SDK calls in a thread pool.Initialize once at process startup and share the client across requests. The
client is safe to reuse: constructing a second one on the same identity
returns the first, so an accidental second
new SynapClient() does not open
a second cache.import express from 'express';
import { SynapClient, isSynapError } from '@maximem/synap-js-sdk';
const app = express();
app.use(express.json());
const sdk = new SynapClient(); // reads SYNAP_API_KEY from the environment
await sdk.initialize();
app.post('/chat', async (req, res) => {
const { message, userId, customerId, conversationId } = req.body;
let memories = [];
try {
const context = await sdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [message],
});
memories = context.facts ?? [];
} catch (e) {
if (!isSynapError(e)) throw e; // only degrade on Synap's own errors
req.log?.warn(`Synap unavailable: ${e.code}`);
}
const reply = await yourLlm(memories, message);
await sdk.memories.create({
document: `User: ${message}\nAssistant: ${reply}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: { conversation_id: conversationId },
});
res.json({ reply });
});
// Flush in-flight work before the process exits.
process.on('SIGTERM', async () => {
await sdk.shutdown();
process.exit(0);
});
app.listen(3000);
The SDK does not register its own
process.on('exit') handler: it would
leak a listener per client, never fire in Lambda, and process does not
exist in Workers. Call shutdown() yourself, as above.The Synap SDK’s HTTP paths run anywhere, Edge and Workers included. The
real-time anticipation stream is gRPC and needs Node, so pin any route
that calls
listen() to the Node.js runtime. See
Where it runs.instrumentation.ts to initialize the provider once at server startup. The synap.wrap() call in route handlers is lightweight; the provider instance is shared across requests.// instrumentation.ts
import { createSynap } from "@maximem/synap-vercel-adk";
let _synap: Awaited<ReturnType<typeof createSynap>> | null = null;
export async function register() {
if (process.env.NEXT_RUNTIME === "nodejs") {
_synap = await createSynap({ apiKey: process.env.SYNAP_API_KEY });
await _synap.listen(); // optional: open real-time anticipation stream
}
}
export function getSynap() {
if (!_synap) throw new Error("Synap not initialized");
return _synap;
}
listen() warms the anticipation cache so retrieval resolves locally. Your memory writes come from synap.wrap(), which ingests each turn automatically; listen() is not a substitute for it. Note that the middleware also streams each turn as a conversation event, so the same content is extracted again when the conversation compacts. See Real-Time Anticipation.// app/api/chat/route.ts
import { streamText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { getSynap } from "@/instrumentation";
export const runtime = "nodejs"; // required only because instrumentation.ts calls listen()
export async function POST(req: Request) {
const { messages, userId, conversationId } = await req.json();
const model = getSynap().wrap(anthropic("claude-sonnet-4-6"), {
userId,
conversationId,
});
const result = streamText({ model, messages });
return result.toDataStreamResponse();
}
In Django, initialize the SDK in your For Django with ASGI (async views):
AppConfig.ready() method. Use asgiref.sync_to_async for bridging in async views, or asyncio.run() for synchronous views.# myapp/apps.py
import asyncio
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = "myapp"
def ready(self):
from maximem_synap import MaximemSynapSDK
# Store SDK instance on the app config
self.sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
loop = asyncio.new_event_loop()
loop.run_until_complete(self.sdk.initialize())
self._loop = loop
# myapp/views.py
from django.apps import apps
from django.http import JsonResponse
from django.views.decorators.http import require_POST
import json
def get_sdk():
return apps.get_app_config("myapp").sdk
def get_loop():
return apps.get_app_config("myapp")._loop
@require_POST
def chat(request):
data = json.loads(request.body)
sdk = get_sdk()
loop = get_loop()
context = loop.run_until_complete(
sdk.conversation.context.fetch(
conversation_id=data["conversation_id"],
search_query=[data["message"]]
)
)
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
system_prompt = build_system_prompt(memories)
response = generate_response(system_prompt, data["message"])
loop.run_until_complete(
sdk.memories.create(
document=f"User: {data['message']}\nAssistant: {response}",
document_type="ai-chat-conversation",
user_id=data["user_id"],
customer_id=data["customer_id"],
metadata={"conversation_id": data["conversation_id"]}
)
)
return JsonResponse({"response": response})
# myapp/views.py (async version)
from django.http import JsonResponse
import json
async def chat(request):
data = json.loads(request.body)
sdk = get_sdk()
# No bridging needed: SDK is natively async
context = await sdk.conversation.context.fetch(
conversation_id=data["conversation_id"],
search_query=[data["message"]]
)
# ... rest of the handler
Using
loop.run_until_complete() in Django blocks the worker thread during async operations. For production Django deployments with high concurrency, prefer the ASGI/async views pattern shown above, or run the async SDK calls in a thread pool. Be aware that long-lived event loops can interact poorly with dev-server auto-reload and ASGI worker lifecycles.Async/Sync Bridging
Python only. The JavaScript SDK returns promises throughout and there is no
synchronous surface to bridge.
asyncio.run(): Simplest approach
asyncio.run(): Simplest approach
Use for scripts, CLI tools, and simple synchronous applications. Creates a new event loop for each call.
import asyncio
from maximem_synap import MaximemSynapSDK
sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
asyncio.run(sdk.initialize())
# Later, in synchronous code:
context = asyncio.run(sdk.conversation.context.fetch(...))
asyncio.run() creates a new event loop each time. This is fine for low-frequency calls but adds overhead for high-throughput applications.Dedicated event loop: Best for web frameworks
Dedicated event loop: Best for web frameworks
Create a single event loop at startup and reuse it for all SDK calls. This is the pattern shown in the Flask and Django examples above.
import asyncio
# At startup
loop = asyncio.new_event_loop()
# For each call
result = loop.run_until_complete(sdk.some_async_method())
asgiref.sync_to_async: Django-specific
asgiref.sync_to_async: Django-specific
If you are using Django’s async views alongside sync code,
asgiref provides utilities for bridging:from asgiref.sync import async_to_sync
# Wrap an async SDK method for use in sync code
fetch_context = async_to_sync(sdk.conversation.context.fetch)
context = fetch_context(conversation_id="3f6b1a2c-4d5e-6f7a-8b9c-0d1e2f3a4b5c", ...) # conversation_id must be a valid UUID
LLM Provider Integration
- OpenAI
- Anthropic
- Google Gemini (free)
- Vercel AI SDK
Full example of a memory-augmented agent using OpenAI’s
gpt-4o:from openai import AsyncOpenAI
from maximem_synap import MaximemSynapSDK
openai_client = AsyncOpenAI(api_key="sk-...")
synap_sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
async def memory_augmented_chat(
user_message: str,
user_id: str,
customer_id: str,
conversation_id: str
) -> str:
# Step 1: Retrieve relevant context from Synap
context = await synap_sdk.conversation.context.fetch(
conversation_id=conversation_id,
search_query=[user_message]
)
# Step 2: Build the system prompt with retrieved memories
# Each memory type names its text field differently: facts, preferences
# and temporal events use `content`, episodes use `summary`, emotions
# use `context`. Reading `.content` across all five raises
# AttributeError on the first episode.
def memory_text(item) -> str:
return getattr(item, "content", None) or getattr(item, "summary", None) \
or getattr(item, "context", "")
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory_text(memory)}" for memory in memories
])
system_prompt = f"""You are a helpful assistant with access to the following
information about the user and their organization:
{memory_block}
Use this information to provide personalized, contextual responses.
If the user's question relates to something in the memories above,
reference it naturally in your response."""
# Step 3: Call OpenAI with the enriched prompt
response = await openai_client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
temperature=0.7
)
assistant_message = response.choices[0].message.content
# Step 4: Ingest the conversation turn for long-term memory
await synap_sdk.memories.create(
document=f"User: {user_message}\nAssistant: {assistant_message}",
document_type="ai-chat-conversation",
user_id=user_id,
customer_id=customer_id,
metadata={
"conversation_id": conversation_id,
"model": "gpt-4o",
"source": "chat"
}
)
return assistant_message
import OpenAI from 'openai';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const openaiClient = new OpenAI({ apiKey: 'sk-...' });
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(userMessage, userId, customerId, conversationId) {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories
const memories = flattenContextItems(context);
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = memories.map((m) => `- ${m.memory}`).join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.
If the user's question relates to something in the memories above,
reference it naturally in your response.`;
// Step 3: Call OpenAI with the enriched prompt
const response = await openaiClient.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: systemPrompt },
{ role: 'user', content: userMessage },
],
temperature: 0.7,
});
const assistantMessage = response.choices[0]?.message.content ?? '';
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'gpt-4o',
source: 'chat',
},
});
return assistantMessage;
}
import OpenAI from 'openai';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const openaiClient = new OpenAI({ apiKey: 'sk-...' });
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(
userMessage: string,
userId: string,
customerId: string,
conversationId: string,
): Promise<string> {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories
const memories = flattenContextItems(context);
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = memories.map((m) => `- ${m.memory}`).join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.
If the user's question relates to something in the memories above,
reference it naturally in your response.`;
// Step 3: Call OpenAI with the enriched prompt
const response = await openaiClient.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: systemPrompt },
{ role: 'user', content: userMessage },
],
temperature: 0.7,
});
const assistantMessage = response.choices[0]?.message.content ?? '';
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'gpt-4o',
source: 'chat',
},
});
return assistantMessage;
}
Full example of a memory-augmented agent using Anthropic’s Claude:
from anthropic import AsyncAnthropic
from maximem_synap import MaximemSynapSDK
anthropic_client = AsyncAnthropic(api_key="sk-ant-...")
synap_sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
async def memory_augmented_chat(
user_message: str,
user_id: str,
customer_id: str,
conversation_id: str
) -> str:
# Step 1: Retrieve relevant context from Synap
context = await synap_sdk.conversation.context.fetch(
conversation_id=conversation_id,
search_query=[user_message]
)
# Step 2: Build the system prompt with retrieved memories
# Each memory type names its text field differently: facts, preferences
# and temporal events use `content`, episodes use `summary`, emotions
# use `context`. Reading `.content` across all five raises
# AttributeError on the first episode.
def memory_text(item) -> str:
return getattr(item, "content", None) or getattr(item, "summary", None) \
or getattr(item, "context", "")
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory_text(memory)}" for memory in memories
])
system_prompt = f"""You are a helpful assistant with access to the following
information about the user and their organization:
{memory_block}
Use this information to provide personalized, contextual responses.
Reference relevant memories naturally when they apply to the user's question."""
# Step 3: Call Claude with the enriched prompt
response = await anthropic_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=system_prompt,
messages=[
{"role": "user", "content": user_message}
]
)
assistant_message = response.content[0].text
# Step 4: Ingest the conversation turn for long-term memory
await synap_sdk.memories.create(
document=f"User: {user_message}\nAssistant: {assistant_message}",
document_type="ai-chat-conversation",
user_id=user_id,
customer_id=customer_id,
metadata={
"conversation_id": conversation_id,
"model": "claude-sonnet-4-6",
"source": "chat"
}
)
return assistant_message
import Anthropic from '@anthropic-ai/sdk';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const anthropicClient = new Anthropic({ apiKey: 'sk-ant-...' });
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(userMessage, userId, customerId, conversationId) {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories.
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = flattenContextItems(context)
.map((m) => `- ${m.memory}`)
.join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.`;
// Step 3: Call the model with the enriched prompt
const response = await anthropicClient.messages.create({
model: 'claude-sonnet-4-6',
max_tokens: 1024,
system: systemPrompt,
messages: [{ role: 'user', content: userMessage }],
});
const assistantMessage = response.content[0]?.type === 'text' ? response.content[0].text : '';
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'claude-sonnet-4-6',
source: 'chat',
},
});
return assistantMessage;
}
import Anthropic from '@anthropic-ai/sdk';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const anthropicClient = new Anthropic({ apiKey: 'sk-ant-...' });
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(
userMessage: string,
userId: string,
customerId: string,
conversationId: string,
): Promise<string> {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories.
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = flattenContextItems(context)
.map((m) => `- ${m.memory}`)
.join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.`;
// Step 3: Call the model with the enriched prompt
const response = await anthropicClient.messages.create({
model: 'claude-sonnet-4-6',
max_tokens: 1024,
system: systemPrompt,
messages: [{ role: 'user', content: userMessage }],
});
const assistantMessage = response.content[0]?.type === 'text' ? response.content[0].text : '';
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'claude-sonnet-4-6',
source: 'chat',
},
});
return assistantMessage;
}
Gemini has a free API tier: generate a key at Google AI Studio with no billing setup. This lets you complete the Quickstart and test memory-augmented chat without an OpenAI or Anthropic account.
gemini-2.0-flash:from google import genai
from google.genai import types
from maximem_synap import MaximemSynapSDK
gemini_client = genai.Client(api_key="...") # from Google AI Studio
synap_sdk = MaximemSynapSDK() # reads SYNAP_API_KEY from env
async def memory_augmented_chat(
user_message: str,
user_id: str,
customer_id: str,
conversation_id: str
) -> str:
# Step 1: Retrieve relevant context from Synap
context = await synap_sdk.conversation.context.fetch(
conversation_id=conversation_id,
search_query=[user_message]
)
# Step 2: Build the system prompt with retrieved memories
# Each memory type names its text field differently: facts, preferences
# and temporal events use `content`, episodes use `summary`, emotions
# use `context`. Reading `.content` across all five raises
# AttributeError on the first episode.
def memory_text(item) -> str:
return getattr(item, "content", None) or getattr(item, "summary", None) \
or getattr(item, "context", "")
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory_text(memory)}" for memory in memories
])
system_prompt = f"""You are a helpful assistant with access to the following
information about the user and their organization:
{memory_block}
Use this information to provide personalized, contextual responses.
Reference relevant memories naturally when they apply to the user's question."""
# Step 3: Call Gemini with the enriched prompt
response = await gemini_client.aio.models.generate_content(
model="gemini-2.0-flash",
contents=user_message,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
temperature=0.7
)
)
assistant_message = response.text
# Step 4: Ingest the conversation turn for long-term memory
await synap_sdk.memories.create(
document=f"User: {user_message}\nAssistant: {assistant_message}",
document_type="ai-chat-conversation",
user_id=user_id,
customer_id=customer_id,
metadata={
"conversation_id": conversation_id,
"model": "gemini-2.0-flash",
"source": "chat"
}
)
return assistant_message
import { GoogleGenerativeAI } from '@google/generative-ai';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const genai = new GoogleGenerativeAI(process.env.GOOGLE_API_KEY ?? '');
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(userMessage, userId, customerId, conversationId) {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories.
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = flattenContextItems(context)
.map((m) => `- ${m.memory}`)
.join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.`;
// Step 3: Call the model with the enriched prompt
const model = genai.getGenerativeModel({
model: 'gemini-2.0-flash',
systemInstruction: systemPrompt,
});
const response = await model.generateContent(userMessage);
const assistantMessage = response.response.text();
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'gemini-2.0-flash',
source: 'chat',
},
});
return assistantMessage;
}
import { GoogleGenerativeAI } from '@google/generative-ai';
import { SynapClient, flattenContextItems } from '@maximem/synap-js-sdk';
const genai = new GoogleGenerativeAI(process.env.GOOGLE_API_KEY ?? '');
const synapSdk = new SynapClient(); // reads SYNAP_API_KEY from env
async function memoryAugmentedChat(
userMessage: string,
userId: string,
customerId: string,
conversationId: string,
): Promise<string> {
// Step 1: Retrieve relevant context from Synap
const context = await synapSdk.conversation.context.fetch({
conversation_id: conversationId,
search_query: [userMessage],
});
// Step 2: Build the system prompt with retrieved memories.
// flattenContextItems normalises each type's text field into `.memory`,
// so episodes and emotions are not silently blank.
const memoryBlock = flattenContextItems(context)
.map((m) => `- ${m.memory}`)
.join('\n');
const systemPrompt = `You are a helpful assistant with access to the following
information about the user and their organization:
${memoryBlock}
Use this information to provide personalized, contextual responses.`;
// Step 3: Call the model with the enriched prompt
const model = genai.getGenerativeModel({
model: 'gemini-2.0-flash',
systemInstruction: systemPrompt,
});
const response = await model.generateContent(userMessage);
const assistantMessage = response.response.text();
// Step 4: Ingest the conversation turn for long-term memory
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${assistantMessage}`,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
metadata: {
conversation_id: conversationId,
model: 'gemini-2.0-flash',
source: 'chat',
},
});
return assistantMessage;
}
If your application uses the Vercel AI SDK, Setup (run once at startup)Usage in a route handler or server actionNext.js App Router (streaming)
@maximem/synap-vercel-adk wraps any LanguageModelV1 model as a middleware. Context retrieval and memory writes happen automatically, with no manual fetch/inject/ingest loop needed.Retrieval and ingestion run anywhere, Edge and Workers included. Only
listen() is Node-only, because the anticipation stream is gRPC. See
Where it runs.// instrumentation.ts (Next.js) or top-level module
import { createSynap } from "@maximem/synap-vercel-adk";
export const synap = await createSynap({
apiKey: process.env.SYNAP_API_KEY,
});
// Optional: open real-time anticipation cache updates
await synap.listen();
listen() warms the anticipation cache; memories are written by synap.wrap(), which ingests each turn automatically. The middleware also streams each turn as a conversation event, and those turns are extracted again when the conversation compacts. See Real-Time Anticipation.import { generateText, streamText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { synap } from "@/lib/synap";
// conversationId must be a valid UUID; generate one with crypto.randomUUID()
const model = synap.wrap(anthropic("claude-sonnet-4-6"), {
userId: "user-123",
conversationId: crypto.randomUUID(),
});
// generateText: context injected, memory written automatically
const { text } = await generateText({
model,
messages: [{ role: "user", content: userMessage }],
});
// streamText works identically
const result = streamText({ model, messages });
// app/api/chat/route.ts
import { streamText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { synap } from "@/lib/synap";
export async function POST(req: Request) {
const { messages, userId, conversationId } = await req.json();
const model = synap.wrap(anthropic("claude-sonnet-4-6"), {
userId,
conversationId,
});
const result = streamText({ model, messages });
return result.toDataStreamResponse();
}
synap.wrap() returns a standard LanguageModelV1; it is compatible with all Vercel AI SDK functions (generateText, streamText, generateObject, etc.) and any framework built on top of the AI SDK.For the
writeMemory option (default true), the middleware fires a background memory write after each generation so it never adds latency to the response. Set writeMemory: false to disable.The Memory-Augmented Agent Pattern
Every Synap integration follows the same fundamental pattern:1
Retrieve
Fetch relevant memories from Synap based on the user’s message, identity, and conversation history.
context = await sdk.conversation.context.fetch(...)
const context = await sdk.conversation.context.fetch({ /* ... */ });
const context = await sdk.conversation.context.fetch({ /* ... */ });
2
Build Prompt
Inject retrieved memories into the system prompt or message history. This gives the LLM access to personalized, contextual information.
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
system_prompt = build_prompt_with_memories(memories)
import { flattenContextItems } from '@maximem/synap-js-sdk';
// Concatenates every collection in the same order as the Python example.
const memories = flattenContextItems(context);
const systemPrompt = buildPromptWithMemories(memories);
import { flattenContextItems } from '@maximem/synap-js-sdk';
// Concatenates every collection in the same order as the Python example.
const memories = flattenContextItems(context);
const systemPrompt = buildPromptWithMemories(memories);
3
Generate
Call your LLM provider with the enriched prompt. The model generates a response informed by the user’s history, preferences, and organizational context.
response = await llm_client.generate(system_prompt, user_message)
const response = await llmClient.generate(systemPrompt, userMessage);
const response = await llmClient.generate(systemPrompt, userMessage);
4
Ingest
Send the conversation turn back to Synap for processing and long-term storage. This closes the loop: today’s conversation becomes tomorrow’s context.
await sdk.memories.create(document=turn_content, document_type="ai-chat-conversation", user_id=..., customer_id=...)
await sdk.memories.create({
document: turnContent,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
});
await sdk.memories.create({
document: turnContent,
document_type: 'ai-chat-conversation',
user_id: userId,
customer_id: customerId,
});
Error Handling and Graceful Degradation
Synap should enhance your application, not be a single point of failure. Design your integration to degrade gracefully when Synap is unavailable.In JavaScript, branch on
error.code, not instanceof. With a dual
ESM/CJS dependency graph a consumer can hold two copies of the same error
class, and instanceof then fails against the copy it was not built from.
isSynapError(e) narrows to Synap’s own taxonomy, so a bug in your code
rethrows instead of being swallowed as “Synap was down”. See
How it differs from the Python SDK.async def chat_with_graceful_degradation(user_message: str, **kwargs) -> str:
memories = []
# Attempt to retrieve context, but don't fail if Synap is down
try:
context = await synap_sdk.conversation.context.fetch(
conversation_id=kwargs["conversation_id"],
search_query=[user_message]
)
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
except Exception as e:
logger.warning(f"Synap retrieval failed, proceeding without context: {e}")
# Generate response (with or without memories)
system_prompt = build_system_prompt(memories) # handles empty list gracefully
response = await generate_response(system_prompt, user_message)
# Attempt to ingest, but don't fail if Synap is down
try:
await synap_sdk.memories.create(
document=f"User: {user_message}\nAssistant: {response}",
document_type="ai-chat-conversation",
user_id=kwargs["user_id"],
customer_id=kwargs["customer_id"],
metadata={"conversation_id": kwargs["conversation_id"]}
)
except Exception as e:
logger.warning(f"Synap ingestion failed: {e}")
return response
import { isSynapError, flattenContextItems } from '@maximem/synap-js-sdk';
async function chatWithGracefulDegradation(userMessage, opts) {
let memories = [];
// Attempt to retrieve context, but don't fail if Synap is down
try {
const context = await synapSdk.conversation.context.fetch({
conversation_id: opts.conversationId,
search_query: [userMessage],
});
memories = flattenContextItems(context);
} catch (e) {
// isSynapError narrows to the SDK's own taxonomy; anything else rethrows,
// so a bug in your own code is not swallowed as "Synap was down".
if (!isSynapError(e)) throw e;
logger.warn(`Synap retrieval failed, proceeding without context: ${e.code}`);
}
// Generate response (with or without memories)
const systemPrompt = buildSystemPrompt(memories); // handles empty list gracefully
const response = await generateResponse(systemPrompt, userMessage);
// Attempt to ingest, but don't fail if Synap is down
try {
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${response}`,
document_type: 'ai-chat-conversation',
user_id: opts.userId,
customer_id: opts.customerId,
metadata: { conversation_id: opts.conversationId },
});
} catch (e) {
if (!isSynapError(e)) throw e;
logger.warn(`Synap ingestion failed: ${e.code}`);
}
return response;
}
import { isSynapError, flattenContextItems } from '@maximem/synap-js-sdk';
import type { FlatMemory } from '@maximem/synap-js-sdk';
interface ChatOptions {
conversationId: string;
userId: string;
customerId: string;
}
async function chatWithGracefulDegradation(
userMessage: string,
opts: ChatOptions,
): Promise<string> {
let memories: FlatMemory[] = [];
// Attempt to retrieve context, but don't fail if Synap is down
try {
const context = await synapSdk.conversation.context.fetch({
conversation_id: opts.conversationId,
search_query: [userMessage],
});
memories = flattenContextItems(context);
} catch (e) {
// isSynapError narrows to the SDK's own taxonomy; anything else rethrows,
// so a bug in your own code is not swallowed as "Synap was down".
if (!isSynapError(e)) throw e;
logger.warn(`Synap retrieval failed, proceeding without context: ${e.code}`);
}
// Generate response (with or without memories)
const systemPrompt = buildSystemPrompt(memories); // handles empty list gracefully
const response = await generateResponse(systemPrompt, userMessage);
// Attempt to ingest, but don't fail if Synap is down
try {
await synapSdk.memories.create({
document: `User: ${userMessage}\nAssistant: ${response}`,
document_type: 'ai-chat-conversation',
user_id: opts.userId,
customer_id: opts.customerId,
metadata: { conversation_id: opts.conversationId },
});
} catch (e) {
if (!isSynapError(e)) throw e;
logger.warn(`Synap ingestion failed: ${e.code}`);
}
return response;
}
Always wrap Synap SDK calls in
try blocks in production (try/except in Python, try/catch in JavaScript). Network issues, rate limits, or service disruptions should not prevent your application from responding to users. The LLM can still generate useful responses without memory context; it just won’t be personalized.Next Steps
Authentication
Configure API keys for secure communication.
SDK Initialization
Detailed SDK configuration options.
First Integration Guide
Step-by-step tutorial for your first Synap integration.