CrewAI

StorageBackend implementation that routes CrewAI’s unified memory through 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.

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

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:

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

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:

1. Memory survives the crew. Re-running kickoff with a new topic still benefits from the prior run’s findings.
2. 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.
3. 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.

---

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.