CrewAI - Maximem Synap

Make CrewAI’s built-in memory durable and queryable across crews. The integration plugs into CrewAI’s StorageBackend protocol, so existing crews keep working. They just gain a long-term memory that survives restarts.

Requires Python 3.11+.

Overview

This guide shows how to add Synap to a CrewAI application to build crews that:

Persist Memory entries across executions and processes
Retrieve semantically-relevant context from prior crew runs
Scope memories per user or per organization

The Synap CrewAI integration ships a single class: a drop-in replacement for CrewAI’s default storage backend.

Class	CrewAI interface	Purpose
`SynapStorageBackend`	`StorageBackend`	Persistent storage for CrewAI’s unified `Memory` system

Setup

Install the package alongside CrewAI:

pip install maximem-synap-crewai crewai

uv add maximem-synap-crewai crewai
# pip-compatible (existing venv): uv pip install maximem-synap-crewai crewai

The pip package is maximem-synap-crewai, but the import drops the maximem- prefix and uses underscores: from synap_crewai 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 swaps in SynapStorageBackend as the storage for CrewAI’s Memory. Every save call inside the crew is routed to Synap, and search returns ranked, semantically-matched results:

# pip install maximem-synap-crewai crewai
from maximem_synap import MaximemSynapSDK
from crewai import Agent, Crew, Task
from crewai.memory import Memory
from synap_crewai import SynapStorageBackend

sdk = MaximemSynapSDK()
await sdk.initialize()

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

memory = Memory(storage=backend)

crew = Crew(
    agents=[your_agent],
    tasks=[your_task],
    memory=memory,
)

result = crew.kickoff(inputs={"topic": "quarterly planning"})

CrewAI’s memory machinery is now backed by Synap. Search failures degrade gracefully (empty result + log); writes raise on failure so silent data loss is impossible.

Core concepts

Lifecycle methods

SynapStorageBackend implements CrewAI’s StorageBackend protocol by mapping each method to the equivalent Synap operation:

backend = SynapStorageBackend(sdk=sdk, user_id="alice", customer_id="acme")

Method	Behavior
`save(value, metadata)`	Ingests a memory fragment with optional metadata tags
`search(query, limit, score_threshold)`	Semantic search; returns ranked results
`list_records(limit)`	Returns recent memories (uses a broad Synap search)
`count()`	Returns an approximate count via a broad search

CrewAI operations that have no direct Synap equivalent (notably delete) are no-ops with a warning logged. The memory pipeline keeps moving rather than blocking the crew.

Async support

CrewAI 0.100+ supports async task execution. SynapStorageBackend exposes asearch for use in async crews:

results = await backend.asearch("project deadlines", limit=5)

The synchronous search method wraps asearch via an event-loop bridge, so the same backend works in both sync and async crews. You don’t need separate configurations.

Complete example: research crew with shared memory

The following crew chains a researcher and a writer. Both agents share the same Synap-backed memory, so the writer sees what the researcher learned, and the next crew kickoff has access to everything from earlier runs:

from crewai import Agent, Crew, Task, Process
from crewai.memory import Memory
from synap_crewai import SynapStorageBackend


def build_research_crew(sdk, user_id: str, customer_id: str | None = None) -> Crew:
    backend = SynapStorageBackend(sdk=sdk, user_id=user_id, customer_id=customer_id)
    memory = Memory(storage=backend)

    researcher = Agent(
        role="Senior Researcher",
        goal="Find concrete, sourced information about the topic.",
        backstory="Methodical analyst who cites sources.",
    )

    writer = Agent(
        role="Technical Writer",
        goal="Turn research into a clear summary.",
        backstory="Editor with a bias toward clarity.",
    )

    research_task = Task(
        description="Research the topic: {topic}.",
        expected_output="A bullet list of sourced findings.",
        agent=researcher,
    )

    writing_task = Task(
        description="Write a 200-word summary based on the research.",
        expected_output="A polished prose summary.",
        agent=writer,
        context=[research_task],
    )

    return Crew(
        agents=[researcher, writer],
        tasks=[research_task, writing_task],
        process=Process.sequential,
        memory=memory,
    )


# Usage
crew = build_research_crew(sdk, user_id="alice", customer_id="acme")
result = crew.kickoff(inputs={"topic": "agentic memory architectures"})

# Subsequent runs benefit from accumulated memory
crew.kickoff(inputs={"topic": "long-context retrieval"})

Three things to notice in this pattern:

Memory survives the crew. Re-running kickoff with a new topic still benefits from the prior run’s findings.
Scope is fixed at backend construction. All agents in this crew share the same user_id/customer_id. Build a separate crew per user if you need isolation.
CrewAI’s built-in retrieval is unchanged. You don’t restructure tasks or prompts; the memory backend just becomes Synap.

Advanced patterns

Multi-tenant scoping

SynapStorageBackend accepts 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
backend = SynapStorageBackend(sdk=sdk, user_id="alice")

# Organization-scoped
backend = SynapStorageBackend(sdk=sdk, user_id="alice", customer_id="acme")

For multi-tenant services, construct a backend per request; never share a backend whose scope doesn’t match the inbound user.

Sync and async in the same codebase

SynapStorageBackend supports both search (sync) and asearch (async). The sync method bridges to the async path via an event loop, so:

In a sync crew, just call crew.kickoff() as usual; search works transparently.
In an async crew, prefer asearch for direct access; the framework will call it for you when you await crew tasks.

Failure semantics

The integration follows the Synap-wide contract:

search and asearch degrade gracefully: return [] and log an error if Synap is unreachable.
save surfaces failures: raises SynapIntegrationError so the crew (and caller) know if persistence failed.
Unsupported operations (e.g., delete) are no-ops with a warning rather than raising, so they don’t kill the crew.

This is by design: read failures shouldn’t break a crew run mid-flight, 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

AutoGen

BaseTool implementations for AutoGen agents.

Pydantic AI

Deps and tools for Pydantic AI.

Memory Scopes

How user_id and customer_id interact across reads and writes.

Ingestion

Direct ingestion API for pipelines that need finer control than save.

​Overview

​Setup

​Basic integration

​Core concepts

​Lifecycle methods

​Async support

​Complete example: research crew with shared memory

​Advanced patterns

​Multi-tenant scoping

​Sync and async in the same codebase

​Failure semantics

​Going further

​Next steps

AutoGen

Pydantic AI

Memory Scopes

Ingestion

Overview

Setup

Basic integration

Core concepts

Lifecycle methods

Async support

Complete example: research crew with shared memory

Advanced patterns

Multi-tenant scoping

Sync and async in the same codebase

Failure semantics

Going further

Next steps