> ## Documentation Index
> Fetch the complete documentation index at: https://docs.maximem.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Agent Integration
> The integration built for agents: open one real-time stream, report each turn as it happens, and let Synap form long-term memory on its own. No per-turn ingestion calls.
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.
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
```python theme={null}
from maximem_synap import MaximemSynapSDK
sdk = MaximemSynapSDK(api_key=os.environ["SYNAP_API_KEY"])
await sdk.initialize()
```
Not one per user, not one per session. Scope travels on each call, not on the stream.
```python theme={null}
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](/patterns/real-time-anticipation-server) for quotas, reconnects, and the health check worth alerting on.
```python theme={null}
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.
```python theme={null}
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.
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.
Stream quotas are per Instance and per client. Opening one per user session exhausts them under real concurrency.
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:
A fact the very next turn depends on, or a live demo showing memory forming. Waiting for compaction is not an option.
Promotion ingests conversation turns as conversation content. To set `document_type`, `mode`, custom metadata, or to write at customer or client scope, ingest explicitly.
Product docs, support tickets, CRM records, and backfills belong in [ingestion](/concepts/how-ingestion-works), 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](/integrations/strands-agents) via `SynapStreamHook`, and the [Vercel AI SDK](/integrations/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.
The stream in depth: event types, cache behavior, and failure modes.
Quotas, reconnects, and the silent failure to alert on.