Agent Integration

If you are wiring Synap into a live agent (a chat assistant, a copilot, a voice bot, anything with a conversation loop), this is the integration to build. Your agent reports what happened; Synap handles retrieval and memory formation.

Agent integration flow: sdk.initialize() initializes the SDK, then sdk.instance.listen() starts listening once at startup, then a repeating per-turn block of send_message(user_message), fetch() served from cache, your LLM, and send_message(assistant_message), and finally sdk.instance.stop_listening() on shutdown

That is the whole loop. There is no memories.create() in it.

Why there is no ingestion call

Conversation turns you report with send_message() are persisted to conversation history. When the conversation compacts, Synap promotes those raw turns into the long-term ingestion pipeline: the same extraction stages memories.create() runs, producing memories of the same quality. Compaction happens on any of three triggers:

Trigger	Default	Applies to
Token threshold	3,000 tokens	Long conversations
Message count	10 messages	Busy conversations
Idle period	5 minutes of inactivity	Everything else

The token and message thresholds are configurable per Instance. The idle trigger is what makes this complete: a two-turn exchange that never approaches the other thresholds still compacts a few minutes after the user stops, and its turns still become memory.

Because the idle trigger catches whatever the thresholds miss, every conversation reaches long-term memory eventually. Reporting turns with send_message() is sufficient on its own.

The integration

Initialize once, at process startup

from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK(api_key=os.environ["SYNAP_API_KEY"])
await sdk.initialize()

import { SynapClient } from '@maximem/synap-js-sdk';

const sdk = new SynapClient();   // reads SYNAP_API_KEY from the environment
await sdk.initialize();

import { SynapClient } from '@maximem/synap-js-sdk';

const sdk = new SynapClient();   // reads SYNAP_API_KEY from the environment
await sdk.initialize();

Open one stream for the process

Not one per user, not one per session. Scope travels on each call, not on the stream.

await sdk.instance.listen(
    on_reconnect=lambda attempt: log.info("stream reconnected (%d)", attempt),
    on_disconnect=lambda reason: log.warning("stream lost: %s", reason),
)

// Node only, and needs the optional peers:
//   npm install @grpc/grpc-js @grpc/proto-loader
await sdk.instance.listen({
  on_reconnect: (attempt) => console.info('stream reconnected (%d)', attempt),
  on_disconnect: (reason) => console.warn('stream lost: %s', reason),
});

// Node only, and needs the optional peers:
//   npm install @grpc/grpc-js @grpc/proto-loader
await sdk.instance.listen({
  on_reconnect: (attempt: number) => console.info('stream reconnected (%d)', attempt),
  on_disconnect: (reason: string) => console.warn('stream lost: %s', reason),
});

This step is the one JavaScript restriction. Every other call in this loop runs on Edge and Workers; listen() is gRPC, so it needs Node plus @grpc/grpc-js and @grpc/proto-loader. Importing the SDK in an Edge route stays safe, only the stream is unavailable there.

See Real-Time Anticipation in a Server for quotas, reconnects, and the health check worth alerting on.

Report each turn

import uuid

conversation_id = str(uuid.uuid4())   # one per conversation, reused every turn

async def handle_turn(user_text: str, user_id: str, customer_id: str) -> str:
    await sdk.instance.send_message(
        content=user_text,
        role="user",
        event_type="user_message",
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,
    )

    context = await sdk.user.context.fetch(
        user_id=user_id,
        customer_id=customer_id,
        conversation_id=conversation_id,
        search_query=[user_text],
    )

    reply = await your_llm(context, user_text)

    await sdk.instance.send_message(
        content=reply,
        role="assistant",
        event_type="assistant_message",
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,
    )
    return reply

import { randomUUID } from 'node:crypto';

const conversationId = randomUUID();   // one per conversation, reused every turn

async function handleTurn(userText, userId, customerId) {
  await sdk.instance.send_message({
    content: userText,
    role: 'user',
    event_type: 'user_message',
    conversation_id: conversationId,
    user_id: userId,
    customer_id: customerId,
  });

  const context = await sdk.user.context.fetch({
    user_id: userId,
    customer_id: customerId,
    conversation_id: conversationId,
    search_query: [userText],
  });

  const reply = await yourLlm(context, userText);

  await sdk.instance.send_message({
    content: reply,
    role: 'assistant',
    event_type: 'assistant_message',
    conversation_id: conversationId,
    user_id: userId,
    customer_id: customerId,
  });
  return reply;
}

import { randomUUID } from 'node:crypto';
import type { RawContext } from '@maximem/synap-js-sdk';

declare function yourLlm(context: RawContext, userText: string): Promise<string>;

const conversationId = randomUUID();   // one per conversation, reused every turn

async function handleTurn(
  userText: string,
  userId: string,
  customerId: string,
): Promise<string> {
  await sdk.instance.send_message({
    content: userText,
    role: 'user',
    event_type: 'user_message',
    conversation_id: conversationId,
    user_id: userId,
    customer_id: customerId,
  });

  const context = await sdk.user.context.fetch({
    user_id: userId,
    customer_id: customerId,
    conversation_id: conversationId,
    search_query: [userText],
  });

  const reply = await yourLlm(context, userText);

  await sdk.instance.send_message({
    content: reply,
    role: 'assistant',
    event_type: 'assistant_message',
    conversation_id: conversationId,
    user_id: userId,
    customer_id: customerId,
  });
  return reply;
}

Emit assistant_message after the reply. Anticipation runs between turns, so this event is what pre-warms the next one.

Close on shutdown

await sdk.instance.stop_listening()
await sdk.shutdown()

await sdk.instance.stop_listening();
await sdk.shutdown();

await sdk.instance.stop_listening();
await sdk.shutdown();

Report the whole turn, not just the two ends

Most integrations report the user’s message and the assistant’s reply and stop there. That works, and it leaves most of the value on the table. Synap then sees the question and the answer and nothing about the work in between, which is where the next question is usually decided. Five things happen in a turn. Report all five.

What happened	Call	What you lose by skipping it
The user spoke	`send_message(event_type="user_message")`	Nothing is recorded. No memory forms from this conversation at all.
Your agent reasoned	`record_thinking(...)`	Anticipation predicts the next turn from the observable edges only, and misses what the agent was actually working towards.
It called a tool	`record_tool_call(...)`	Anticipation cannot tell a lookup from a booking, so it cannot prefetch for what the tool is about to return.
The tool answered	`record_tool_result(...)`	The turn’s outcome is invisible. A failed lookup and a successful one look identical.
Your agent replied	`send_message(event_type="assistant_message")`	Anticipation never runs. This is the event it acts on: nothing is prefetched, and every following turn is a cold retrieval.

If you implement only one of these beyond the user turn, implement assistant_message. Anticipation fires on it. An integration that reports tool calls and reasoning but never the reply gets no prefetching at all, and looks from the outside exactly like one that is working.

Where each call goes

async with sdk_session():                      # your request handler
    # 1. the user's turn, BEFORE you retrieve
    await sdk.instance.send_message(
        content=user_text, role="user", event_type="user_message",
        conversation_id=conv_id, user_id=user_id, customer_id=customer_id,
    )

    context = await sdk.fetch(...)             # served from cache if anticipated

    # 2. each reasoning step, as your agent produces it
    await sdk.instance.record_thinking(
        content=thought, conversation_id=conv_id,
        user_id=user_id, customer_id=customer_id,
    )

    # 3. and 4. around every tool call
    await sdk.instance.record_tool_call(
        tool_name, tool_args, tool_call_id=call_id,
        conversation_id=conv_id, user_id=user_id, customer_id=customer_id,
    )
    result = await run_tool(...)
    await sdk.instance.record_tool_result(
        result, tool_name=tool_name, tool_call_id=call_id,
        conversation_id=conv_id, user_id=user_id, customer_id=customer_id,
    )

    # 5. the reply, AFTER your agent produces it
    await sdk.instance.send_message(
        content=reply, role="assistant", event_type="assistant_message",
        conversation_id=conv_id, user_id=user_id, customer_id=customer_id,
    )

# when the conversation ends and your process keeps running
await sdk.instance.end_session(conv_id)

No session call appears in that loop. The SDK opens one on the first event and closes it when you stop listening. end_session is only for a conversation that ends while the process lives on: a call hangs up, a chat window closes.

Check your own work

An integration is finished when all of these are true. If you are an agent wiring this up, check them before you report back.

listen() is called once at startup, and stop_listening() on shutdown.
Every turn reports a user_message and an assistant_message.
Tool calls and results are reported, and share a tool_call_id.
Reasoning is reported where the framework exposes it.
Every event carries conversation_id and user_id (plus customer_id on B2B).
No send_message call sets role by hand for a tool or reasoning event. Use the typed methods; the role and the event type have to agree, and getting that wrong fails silently.
end_session is called when a conversation ends and the process keeps running.

Using one of the supported frameworks? Do not write any of this by hand. The integration packages for OpenAI Agents, Google ADK, the Claude Agent SDK and the Vercel AI SDK report these events for you from the framework’s own hooks. See Framework integrations.

Requirements

Every event needs both user_id and customer_id. A conversation event missing either is discarded server-side with no client-visible error. The turn is never persisted, so it never becomes memory. On a B2C instance send user_id only: customer_id is not accepted there, and an event without one is persisted normally.

required

The stream must stay open across turns. This integration suits servers, workers, and voice sessions, not per-request serverless functions, which cannot hold a connection.

required

Stream quotas are per Instance and per client. Opening one per user session exhausts them under real concurrency.

required

Reused across every turn of a conversation. send_message() does not validate it, but fetch() and every other call do.

What still uses REST

The stream is not the entire transport, and it is not meant to be. Two things go over HTTP:

initialize(): resolves your client and instance from the API key
fetch() on an anticipation miss: pushed bundles cover what Synap predicted; anything it did not predict is fetched normally

You do not code either differently. fetch() is one call whether it resolves from the local anticipation cache in about a millisecond or falls through to the network.

When to still call `memories.create()`

Promotion is automatic but not instant. A turn becomes retrievable when its conversation compacts, which for a quiet conversation means about five minutes. Reach for explicit ingestion in three cases:

use memories.create

A fact the very next turn depends on, or a live demo showing memory forming. Waiting for compaction is not an option.

use memories.create

Promotion ingests conversation turns as conversation content. To set document_type, mode, custom metadata, or to write at customer or client scope, ingest explicitly.

use memories.create

Product docs, support tickets, CRM records, and backfills belong in ingestion, not on the stream.

Do not do both on the same text. If you report a turn with send_message() and ingest that same turn with memories.create(), the content is extracted twice, once immediately and once at compaction, costing double and producing overlapping memories the deduplication stage then reconciles.Pick one owner per piece of content: stream conversation turns, ingest everything else.

Framework integrations

Five packages drive the stream for you, from the framework’s own hooks. With any other framework, or none, use the loop above directly: it is plain SDK calls and composes with anything.

Framework	What it reports for you	What is still yours
OpenAI Agents	Tool calls, tool results, model reasoning, and the reply	The user’s turn, before the run
Google ADK	Tool calls, tool results, and the reply	The user’s turn, before the run
Claude Agent SDK	The user’s turn, tool calls, tool results, and the session end	The reply: no hook carries the final assistant text, so call `report_assistant_turn` once your run returns
Vercel AI SDK	Turns, tool calls, and reasoning, from the model middleware	Nothing
Strands Agents	Turns and tool intent	Tool results and reasoning

Each package is silent unless listen() is running, so adding one changes nothing until you open a stream.

How anticipation works

The stream in depth: event types, cache behavior, and failure modes.

Running it in a server

Quotas, reconnects, and the silent failure to alert on.

Setup

SDK Guide

Why there is no ingestion call

The integration

Report the whole turn, not just the two ends

Where each call goes

Check your own work

Requirements

What still uses REST

When to still call `memories.create()`

Framework integrations

How anticipation works

Running it in a server

​Why there is no ingestion call

​The integration

​Report the whole turn, not just the two ends

​Where each call goes

​Check your own work

​Requirements

​What still uses REST

​When to still call memories.create()

​Framework integrations

How anticipation works

Running it in a server

Why there is no ingestion call

The integration

Report the whole turn, not just the two ends

Where each call goes

Check your own work

Requirements

What still uses REST

When to still call `memories.create()`

Framework integrations