Advanced: for real-time integrations. Most applications only need
fetch and 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.listen() opens a bidirectional gRPC stream between the SDK and the Synap platform. Once the stream is active, you can call 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 for the full model.
Connection targets come from SDKConfig: grpc_host, grpc_port, and grpc_use_tls.
Events the SDK sends for you
Once the stream is live, the SDK instrumentsfetch() automatically. You do not write this code, but these events flow on your stream:
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 theuser_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.
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.
Parameters
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.
Callable[[str], None]
Callback invoked when the stream disconnects. Receives the disconnect reason as a string.
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.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.Returns
ReturnsNone. The coroutine resolves once the stream is established; the stream itself stays open until you call instance.stop_listening.
Example
Raises
SDKNotInitializedError: wheninitialize()has not been called.AuthenticationError: when credentials are rejected by the platform.
See also
- Real-Time Anticipation: what the stream does and what it deliberately does not do.
- Real-Time Anticipation in a Server: running one shared stream in a multi-tenant process.
- instance.send_message: push agent activity over the active stream.
- instance.stop_listening: close the stream and release resources.
- fetch: request-response context retrieval (no streaming required).