> ## 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.

<Frame>
  <img src="https://mintcdn.com/maximemai/vw5cRE_0HUi1NbIA/images/agent-integration-flow.png?fit=max&auto=format&n=vw5cRE_0HUi1NbIA&q=85&s=854cf10ae1ee105c7708c0871a8b6ce2" alt="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" width="1536" height="1024" data-path="images/agent-integration-flow.png" />
</Frame>

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.

<Note>
  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.
</Note>

## The integration

<Steps>
  <Step title="Initialize once, at process startup">
    <CodeGroup>
      ```python Python theme={null}
      from maximem_synap import MaximemSynapSDK

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

      ```javascript JavaScript theme={null}
      import { SynapClient } from '@maximem/synap-js-sdk';

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

      ```typescript TypeScript theme={null}
      import { SynapClient } from '@maximem/synap-js-sdk';

      const sdk = new SynapClient();   // reads SYNAP_API_KEY from the environment
      await sdk.initialize();
      ```
    </CodeGroup>
  </Step>

  <Step title="Open one stream for the process">
    Not one per user, not one per session. Scope travels on each call, not on the stream.

    <CodeGroup>
      ```python 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),
      )
      ```

      ```javascript JavaScript theme={null}
      // 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),
      });
      ```

      ```typescript TypeScript theme={null}
      // 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),
      });
      ```
    </CodeGroup>

    <Note>
      **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.
    </Note>

    See [Real-Time Anticipation in a Server](/patterns/real-time-anticipation-server) for quotas, reconnects, and the health check worth alerting on.
  </Step>

  <Step title="Report each turn">
    <CodeGroup>
      ```python 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
      ```

      ```javascript JavaScript theme={null}
      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;
      }
      ```

      ```typescript TypeScript theme={null}
      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;
      }
      ```
    </CodeGroup>

    Emit `assistant_message` **after** the reply. Anticipation runs between turns, so this event is what pre-warms the next one.
  </Step>

  <Step title="Close on shutdown">
    <CodeGroup>
      ```python Python theme={null}
      await sdk.instance.stop_listening()
      await sdk.shutdown()
      ```

      ```javascript JavaScript theme={null}
      await sdk.instance.stop_listening();
      await sdk.shutdown();
      ```

      ```typescript TypeScript theme={null}
      await sdk.instance.stop_listening();
      await sdk.shutdown();
      ```
    </CodeGroup>
  </Step>
</Steps>

## Requirements

<Warning>
  **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.
</Warning>

<ParamField path="A long-lived process" type="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.
</ParamField>

<ParamField path="One stream per process" type="required">
  Stream quotas are per Instance and per client. Opening one per user session exhausts them under real concurrency.
</ParamField>

<ParamField path="A UUID conversation_id" type="required">
  Reused across every turn of a conversation. `send_message()` does not validate it, but `fetch()` and every other call do.
</ParamField>

## 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:

<ParamField path="You need it retrievable immediately" type="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.
</ParamField>

<ParamField path="You need control over typing or scope" type="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.
</ParamField>

<ParamField path="The content is not a conversation" type="use memories.create">
  Product docs, support tickets, CRM records, and backfills belong in [ingestion](/concepts/how-ingestion-works), not on the stream.
</ParamField>

<Warning>
  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.
</Warning>

## 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.

<CardGroup cols={2}>
  <Card title="How anticipation works" icon="bolt" href="/concepts/real-time-anticipation">
    The stream in depth: event types, cache behavior, and failure modes.
  </Card>

  <Card title="Running it in a server" icon="server" href="/patterns/real-time-anticipation-server">
    Quotas, reconnects, and the silent failure to alert on.
  </Card>
</CardGroup>