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

# instance.listen

> Start a bidirectional gRPC stream that delivers real-time anticipated context bundles to your agent.

```python theme={null}
await sdk.instance.listen(
    on_reconnect=None,
    on_disconnect=None,
    on_context=None,
)
```

<Note>
  Advanced: for real-time integrations. Most applications only need [`fetch`](/sdk-reference/context/fetch) and [`memories.create`](/sdk-reference/memories/create). Use `listen` when you want Synap to proactively push anticipated context to your agent over a long-lived stream, instead of waiting for each `fetch()` round-trip.
</Note>

`listen()` opens a bidirectional gRPC stream between the SDK and the Synap platform. Once the stream is active, you can call [`instance.send_message`](/sdk-reference/instance/send-message) to broadcast agent activity (user messages, tool calls, context requests), and the platform streams back anticipated context bundles. Each incoming bundle is automatically stored in the SDK's anticipation cache so subsequent `fetch()` calls return instantly, and your `on_context` callback fires for any custom handling you need.

The stream is a latency optimization layered on top of the normal request-response API: it never changes what `fetch()` returns, only how fast it returns. It is not memory-neutral, though: conversation turns sent over it are persisted and are promoted into long-term memory when the conversation compacts. See [Real-Time Anticipation](/concepts/real-time-anticipation) for the full model.

Connection targets come from [`SDKConfig`](/sdk/configuration): `grpc_host`, `grpc_port`, and `grpc_use_tls`.

### Events the SDK sends for you

Once the stream is live, the SDK instruments `fetch()` automatically. You do not write this code, but these events flow on your stream:

| Event               | Emitted when                                       |
| ------------------- | -------------------------------------------------- |
| `context_fetch`     | A retrieval is requested                           |
| `context_used`      | A retrieval was served from the anticipation cache |
| `context_assembled` | The SDK finalizes what it composed for the model   |

They drive the platform's learning loop (prefetch scoring, per-pattern hit rates, and the Requests page audit trail) and carry ids and counts only, never raw prompt content.

### One stream per instance

In a long-lived or multi-tenant server, open **one** stream for the process and distinguish users via the `user_id` / `customer_id` on each `send_message` and `fetch`. Do not open a stream per user session: the platform enforces per-instance and per-client stream quotas (currently 100 concurrent streams per instance, 200 per client), and a stream-per-session design saturates them under real concurrency, sending surplus connections into a `RESOURCE_EXHAUSTED` reconnect loop.

<Note>
  Streams have a maximum lifetime of one hour, after which the server closes them and the SDK reconnects with exponential backoff (10 attempts; the counter resets on each successful connect). Periodic reconnects are expected; they are not an error condition.
</Note>

### Parameters

<ParamField path="on_reconnect" type="Callable[[int], None]">
  Callback invoked when the underlying stream reconnects after a transient failure. Receives the attempt count as its only argument. Useful for logging or surfacing connection health in your UI.
</ParamField>

<ParamField path="on_disconnect" type="Callable[[str], None]">
  Callback invoked when the stream disconnects. Receives the disconnect reason as a string.
</ParamField>

<ParamField path="on_context" type="Callable[[dict], None] | Callable[[dict], Awaitable[None]]">
  Callback invoked each time the platform pushes an anticipated context bundle. The bundle dict contains keys like `items_by_type`, `retrieval_mode`, and `bundle_id`. Bundles are also written to the SDK's anticipation cache automatically. You only need this callback if you want to react to bundles directly (e.g., to prefetch UI state). Sync and async callables are both supported.
</ParamField>

<Note>
  All three callbacks are optional, and `on_context` is optional even if you rely on anticipation: bundles reach the cache whether or not you supply it. Note the arity: `on_reconnect` takes the attempt count and `on_disconnect` takes the reason. A zero-argument callback raises `TypeError` at the moment the stream drops.
</Note>

### Returns

Returns `None`. The coroutine resolves once the stream is established; the stream itself stays open until you call [`instance.stop_listening`](/sdk-reference/instance/stop-listening).

### Example

```python theme={null}
import asyncio
import uuid
from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK(api_key="synap_your_key_here")
await sdk.initialize()

# conversation_id must be a valid UUID
conversation_id = str(uuid.uuid4())

def on_context(bundle):
    print(f"Got anticipated bundle {bundle.get('bundle_id')}")

# Run listen() in a background task so the main loop can send messages.
listen_task = asyncio.create_task(
    sdk.instance.listen(on_context=on_context)
)

try:
    await sdk.instance.send_message(
        content="What's my account balance?",
        user_id="user_789",
        customer_id="cust_456",
        conversation_id=conversation_id,
    )
    # ... your agent loop continues here, calling send_message
    # for each user turn, tool call, and context request.
    #
    # Streamed turns reach long-term memory only when this conversation
    # compacts. Call memories.create() for anything that must be
    # retrievable before then.
finally:
    await sdk.instance.stop_listening()
    listen_task.cancel()
```

### Raises

* `SDKNotInitializedError`: when `initialize()` has not been called.
* `AuthenticationError`: when credentials are rejected by the platform.

See [Error Codes](/sdk-reference/errors) for the full SDK exception hierarchy.

### See also

* [Real-Time Anticipation](/concepts/real-time-anticipation): what the stream does and what it deliberately does not do.
* [Real-Time Anticipation in a Server](/patterns/real-time-anticipation-server): running one shared stream in a multi-tenant process.
* [instance.send\_message](/sdk-reference/instance/send-message): push agent activity over the active stream.
* [instance.stop\_listening](/sdk-reference/instance/stop-listening): close the stream and release resources.
* [fetch](/sdk-reference/context/fetch): request-response context retrieval (no streaming required).