AutoGen - Maximem Synap

Expose Synap memory to an AutoGen agent as two BaseTool implementations. The agent decides when to call each, and both tools cooperate with AutoGen’s CancellationToken so a long search can be aborted mid-flight.

Requires Python 3.11+.

Overview

This guide shows how to add Synap to an AutoGen application to build agents that:

Recall user-specific facts, preferences, and past conversations
Persist new information surfaced during a conversation
Respect AutoGen’s cooperative cancellation model so memory calls don’t leak past a cancelled task

The Synap AutoGen integration ships two drop-in tool classes, both implementing AutoGen’s BaseTool interface.

Class	AutoGen interface	Purpose
`SynapSearchTool`	`BaseTool`	Searches Synap memory by natural-language query
`SynapStoreTool`	`BaseTool`	Stores a new memory in Synap

Setup

Install the package alongside AutoGen:

pip install maximem-synap-autogen autogen-agentchat autogen-core

uv add maximem-synap-autogen autogen-agentchat autogen-core
# pip-compatible (existing venv): uv pip install maximem-synap-autogen autogen-agentchat autogen-core

The pip package is maximem-synap-autogen, but the import drops the maximem- prefix and uses underscores: from synap_autogen import ....

Configure your API key. Generate one from the Synap Dashboard.

.env

SYNAP_API_KEY=synap_your_key_here
OPENAI_API_KEY=your-openai-api-key

Initialize the SDK once at application startup:

from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK()
await sdk.initialize()

See SDK Initialization for the full lifecycle and configuration options.

Basic integration

The smallest useful integration registers both tools on an AssistantAgent and runs a task. The agent reads tool descriptions and decides whether to search or store:

# pip install maximem-synap-autogen autogen-agentchat autogen-core
from maximem_synap import MaximemSynapSDK
from autogen_agentchat.agents import AssistantAgent
from synap_autogen import SynapSearchTool, SynapStoreTool

sdk = MaximemSynapSDK()
await sdk.initialize()

tools = [
    SynapSearchTool(sdk=sdk, user_id="alice", customer_id="acme"),
    SynapStoreTool(sdk=sdk, user_id="alice", customer_id="acme"),
]

agent = AssistantAgent(
    name="MemoryAgent",
    model_client=your_model_client,
    tools=tools,
    system_message=(
        "Use synap_search to recall user context. "
        "Use synap_store to remember new information."
    ),
)

await agent.run(task="What are my top priorities this week?")

The scoping triple (user_id, optional customer_id) is bound at construction. The model only ever sees query, max_results, and mode, never the user identity. This prevents prompt-injection attempts from spoofing scope.

Core concepts

Search tool

SynapSearchTool is the read side. The model can override max_results and mode per call but cannot reach outside the scope bound at construction.

from synap_autogen import SynapSearchTool

search = SynapSearchTool(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",   # optional, required for B2B instances
)

Tool schema exposed to the model:

{
  "query": "string",
  "max_results": "int (default 5)",
  "mode": "\"fast\" | \"accurate\" (default \"fast\")"
}

Returns a list of memory objects with content, type, and confidence fields. The two retrieval modes trade latency against comprehensiveness:

	`fast`	`accurate`
Search	Vector + graph (no LLM subquery decomposition)	Vector + graph + LLM subquery decomposition + reranking
Best for	Real-time chat	Multi-entity queries

fast is lower-latency and suited to the hot path; accurate adds LLM-driven query decomposition and reranking for relationship-aware queries at a higher latency cost. Search failures degrade gracefully: the tool returns [] and logs an error so the agent can continue.

Store tool

SynapStoreTool is the write side. The model supplies the content and an optional memory type; everything else is fixed at construction.

from synap_autogen import SynapStoreTool

store = SynapStoreTool(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
)

Tool schema exposed to the model:

{
  "content": "string",
  "memory_type": "string (default \"fact\")"
}

Returns {"status": "stored", "id": "..."} on success. Store failures surface explicitly: the tool raises SynapIntegrationError so the agent (and you) know if persistence failed.

Cooperative cancellation

Both tools propagate AutoGen’s CancellationToken. When the token is cancelled, the in-flight Synap call is aborted and the tool returns control to the agent rather than continuing to completion:

from autogen_core import CancellationToken

token = CancellationToken()
result = await search.run({"query": "project deadlines"}, cancellation_token=token)

# From elsewhere in the same asyncio task group:
token.cancel()

This matters in long-running team conversations where one agent’s tool call should be cancellable by an orchestrator.

Complete example: assistant team with shared memory

The following team has two agents sharing the same Synap-backed memory: a Researcher that searches and stores, and a Planner that searches but never stores. Both run inside an AutoGen RoundRobinGroupChat:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from synap_autogen import SynapSearchTool, SynapStoreTool


def build_team(sdk, model_client, user_id: str, customer_id: str | None = None):
    # Researcher gets both tools: it can recall and remember
    researcher = AssistantAgent(
        name="Researcher",
        model_client=model_client,
        tools=[
            SynapSearchTool(sdk=sdk, user_id=user_id, customer_id=customer_id),
            SynapStoreTool(sdk=sdk, user_id=user_id, customer_id=customer_id),
        ],
        system_message=(
            "You gather facts about the user. "
            "Always call synap_search first; if you learn something new, "
            "call synap_store before responding."
        ),
    )

    # Planner only reads: it shouldn't pollute memory with plans
    planner = AssistantAgent(
        name="Planner",
        model_client=model_client,
        tools=[SynapSearchTool(sdk=sdk, user_id=user_id, customer_id=customer_id)],
        system_message=(
            "You produce action plans based on what Researcher has found. "
            "Use synap_search to verify facts. End your turn with TERMINATE."
        ),
    )

    return RoundRobinGroupChat(
        [researcher, planner],
        termination_condition=TextMentionTermination("TERMINATE"),
    )


# Usage
team = build_team(sdk, model_client, user_id="alice", customer_id="acme")
await team.run(task="Plan my work for next week.")

Three things to notice in this pattern:

Scope is per-tool, per-agent. Both agents see the same user’s memory, but only the Researcher can write. This is a common safety pattern: minimize the surface that can mutate memory.
Memory is shared across agents. Anything Researcher stores during this run is immediately available to Planner on the next turn.
CancellationToken flows automatically through AutoGen’s task group, so cancelling the team also cancels any in-flight Synap calls.

Advanced patterns

Multi-tenant scoping

Both tools accept the standard scoping triple: user_id (required), optional customer_id, optional conversation_id. customer_id is required on B2B Synap instances and ignored on single-tenant ones. See Memory Scopes.

# User-scoped only
search = SynapSearchTool(sdk=sdk, user_id="alice")

# Organization-scoped (user sees org-shared memories too)
search = SynapSearchTool(sdk=sdk, user_id="alice", customer_id="acme-corp")

For multi-tenant services, construct tools per request rather than caching them globally; each task should have its scope baked in.

Tuning retrieval mode

The model can pass mode: "accurate" for higher-recall, slower lookups. For a global default, configure your system prompt to specify the mode the agent should prefer; the tool will respect whatever the model passes.

Failure semantics

The integration follows the Synap-wide contract:

SynapSearchTool degrades gracefully: returns [] and logs an error
SynapStoreTool surfaces failures: raises SynapIntegrationError so the agent and caller know persistence failed
Cancellation is honored: both tools abort cleanly when CancellationToken.cancel() fires

This is by design: read failures shouldn’t break a team conversation, but silent write failures would corrupt the memory pool.

Going further

Patterns overview: reusable memory patterns across frameworks.
Cookbook overview: end-to-end worked examples.

Next steps

OpenAI Agents

Function tools for the OpenAI Agents SDK.

CrewAI

Storage backend for CrewAI crews.

Context Fetch

The retrieval API behind SynapSearchTool: modes, scopes, and response shapes.

Memory Scopes

How user_id and customer_id interact across reads and writes.

​Overview

​Setup

​Basic integration

​Core concepts

​Search tool

​Store tool

​Cooperative cancellation

​Complete example: assistant team with shared memory

​Advanced patterns

​Multi-tenant scoping

​Tuning retrieval mode

​Failure semantics

​Going further

​Next steps

OpenAI Agents

CrewAI

Context Fetch

Memory Scopes

Overview

Setup

Basic integration

Core concepts

Search tool

Store tool

Cooperative cancellation

Complete example: assistant team with shared memory

Advanced patterns

Multi-tenant scoping

Tuning retrieval mode

Failure semantics

Going further

Next steps