Skip to main content

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.

await sdk.instance.listen(
    on_reconnect=None,
    on_disconnect=None,
    on_context=None,
)
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.

​
Parameters

​
on_reconnect
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.
​
on_disconnect
Callable[[str], None]
Callback invoked when the stream disconnects. Receives the disconnect reason as a string.
​
on_context
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.

​
Returns

Returns None. The coroutine resolves once the stream is established; the stream itself stays open until you call instance.stop_listening.

​
Example

import asyncio
from maximem_synap import MaximemSynapSDK

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

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="conv_abc",
    )
    # ... your agent loop continues here, calling send_message
    # for each user turn, tool call, and context request.
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 for the full SDK exception hierarchy.

​
See also