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()

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),
)

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

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()

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 B2C, pass your user identifier as both.

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

Two packages drive the stream for you: Strands Agents via SynapStreamHook, and the Vercel AI SDK via its model middleware. With any other framework, or none, use the loop above directly. It is plain SDK calls and composes with anything.

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

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

​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

Requirements

What still uses REST

When to still call `memories.create()`

Framework integrations