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
- 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.The JavaScript SDK is a wrapper around the Python SDK and requires a Python 3.11+ runtime on the host. Pin Next.js route handlers to the Node.js runtime. Edge Runtime, Cloudflare Workers, Bun, Deno Deploy, and AWS Lambda Node-only runtimes are not supported. See Installation → JavaScript / TypeScript SDK for the full supported-platforms list.
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(); // open real-time anticipation stream
}
}
export function getSynap() {
if (!_synap) throw new Error("Synap not initialized");
return _synap;
}
// app/api/chat/route.ts
import { streamText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { getSynap } from "@/instrumentation";
export const runtime = "nodejs"; // required: the SDK spawns Python and cannot run in Edge Runtime
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
The Synap SDK is async-native. If your application uses a synchronous framework, you need to bridge the async calls. Here are the recommended patterns: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
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory.content}" 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
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
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory.content}" 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
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
memories = (
context.facts + context.preferences + context.episodes
+ context.emotions + context.temporal_events
)
memory_block = "\n".join([
f"- {memory.content}" 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
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.The JavaScript SDK is a wrapper around the Python SDK and requires a Python 3.11+ runtime on the host. Edge Runtime, Cloudflare Workers, Bun, Deno Deploy, and AWS Lambda Node-only runtimes are not supported. See Installation → JavaScript / TypeScript SDK.
// 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();
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(...)
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)
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)
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=...)
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.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
Always wrap Synap SDK calls in try/except blocks in production. 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.