Use with CrewAI

Retrieve Neo4j memory for a CrewAI task, run the crew, and persist its result with a verified task outcome. The selected pattern keeps asynchronous memory operations on their owning event loop and runs CrewAI’s synchronous kickoff() in a worker thread.

1. Prepare the selected environment

Use Python 3.10+ and a POSIX shell. The recipe uses BoltSettings to select Neo4j explicitly and disables entity extraction so that message persistence can be checked independently. Use a dedicated AuraDB instance with vector-index support and no incompatible existing vectors. Follow the Aura connection setup and copy its connection values into the exports below. These recipes use the published Python 0.7.0 package; provider access and database execution must be verified in your environment.

Create a local folder and virtual environment for the examples on this page. The commands reuse an existing environment without changing its files:

mkdir -p ~/agent-memory-tutorials
cd ~/agent-memory-tutorials
if [ -e .venv ]; then
  printf '%s\n' 'Using the existing virtual environment.'
else
  python3 -m venv .venv
fi
source .venv/bin/activate

Expected: ~/agent-memory-tutorials is your working directory and its virtual environment is active. Install the published SDK with the command below.

Each complete code block labelled Save as names a file to create in this folder using your editor. Copy the entire block, including imports and the entry point. Expand each helper disclosure and use Copy code to copy its full source. Keep all files together so their imports resolve.

When continuing from another tutorial or guide, retain the existing environment, configuration, session files, and .tutorial-state/. Reuse unchanged helper files; compare an existing file before replacing it, and finish any pending cleanup or recovery before changing the code that owns its state.

python -m pip install 'neo4j-agent-memory[crewai,openai]==0.7.0'
export NEO4J_URI='neo4j+s://<instance-id>.databases.neo4j.io'
export NEO4J_USERNAME='neo4j'
export NEO4J_PASSWORD='replace-with-your-Aura-password'
export NEO4J_DATABASE='neo4j'
export OPENAI_API_KEY='replace-with-your-provider-key'
export CREWAI_MODEL='openai/replace-with-an-available-model-id'

The OpenAI-backed recipes require access to text-embedding-3-small; it produces 1536-dimensional vectors in this configuration. Configure the selected chat model separately where required. Credentials are read from the environment, never printed by the programs.

2. Read the complete recipe

Create the subdirectory below, then save both complete files under the displayed names. Run commands from agent-memory-tutorials/; Python resolves common from the script’s integrations/ directory.

mkdir -p integrations

The script imports this shared helper from the same directory. It supplies explicit database settings, closes clients through the calling context manager, and fails when expected records are absent. Task traces record the observable outcome of the call; they do not expose hidden model reasoning.

Complete integrations/common.py
Save as integrations/common.py
"""Shared Aura connection through Bolt and readback for integration recipes."""

import os


def settings(embedding=None):
    from neo4j_agent_memory import BoltSettings
    from neo4j_agent_memory.llm import from_provider

    return BoltSettings(
        neo4j={
            "uri": os.environ["NEO4J_URI"],
            "username": os.environ["NEO4J_USERNAME"],
            "password": os.environ["NEO4J_PASSWORD"],
            "database": os.getenv("NEO4J_DATABASE", "neo4j"),
        },
        embedding=embedding or from_provider("openai/text-embedding-3-small", kind="embedding"),
        extraction={"extractor_type": "none"},
    )


async def verify_messages(client, session_id, expected):
    conversation = await client.short_term.get_conversation(session_id)
    contents = [message.content for message in conversation.messages]
    if not all(text in contents for text in expected):
        raise RuntimeError(f"Message readback failed for {session_id}")
    print(f"Verified stored messages; session={session_id}")


async def recorded_turn(client, session_id, prompt, respond):
    """Record a task outcome; this does not claim to capture hidden model reasoning."""
    trace = await client.reasoning.start_trace(session_id=session_id, task=prompt)
    try:
        await client.short_term.add_message(session_id, "user", prompt, extract_entities=False)
        context = await client.get_context(prompt, session_id=session_id)
        reply = str(await respond(context))
        if not reply.strip():
            raise RuntimeError("The framework returned no text")
        await client.short_term.add_message(session_id, "assistant", reply, extract_entities=False)
        await verify_messages(client, session_id, [prompt, reply])
    except Exception as exc:
        await client.reasoning.complete_trace(
            trace.id, success=False, outcome=f"Run failed: {type(exc).__name__}"
        )
        raise
    await client.reasoning.complete_trace(
        trace.id, success=True, outcome="Reply stored and read back"
    )
    stored = await client.reasoning.get_trace_with_steps(trace.id)
    if stored is None or stored.success is not True:
        raise RuntimeError(f"Trace readback failed for {trace.id}")
    print(f"Verified task trace: {trace.id}")
    return reply
Complete integrations/crewai_recipe.py
Save as integrations/crewai_recipe.py
"""Keep async memory I/O outside CrewAI's synchronous kickoff."""

import asyncio
import os
from uuid import uuid4

from common import recorded_turn, settings


def run_crew(context):
    from crewai import LLM, Agent, Crew, Task

    planner = Agent(
        role="Project planner",
        goal="Produce a concise project checklist",
        backstory="You organize fictional project exercises.",
        llm=LLM(model=os.environ["CREWAI_MODEL"]),
        allow_delegation=False,
    )
    task = Task(
        description=(
            "Suggest a simple project planning checklist. Retrieved background:\n" + context
        ),
        expected_output="A short checklist",
        agent=planner,
    )
    return str(Crew(agents=[planner], tasks=[task], memory=False).kickoff())


async def exercise(client, kickoff=run_crew):
    async def respond(context):
        # The worker runs framework code only. The connected async client stays on its owning loop.
        return await asyncio.to_thread(kickoff, context)

    return await recorded_turn(
        client,
        f"docs-crew-{uuid4().hex[:8]}",
        "Suggest a simple project planning checklist.",
        respond,
    )


async def main():
    from neo4j_agent_memory import MemoryClient

    async with MemoryClient(settings()) as client:
        print(await exercise(client))


if __name__ == "__main__":
    asyncio.run(main())

3. Run and verify persistence

python integrations/crewai_recipe.py

Expected: a stored user/assistant exchange, a successful trace readback, and a checklist. The recipe does not enable CrewAI’s separate built-in memory store. If kickoff() fails, the helper records success=False and propagates the exception; it does not fabricate an assistant reply.

Troubleshooting and cleanup

Stop when the program raises an exception; a printed model response alone does not verify persistence. Check the configured database, provider access and vector dimensions before retrying. The program prints the session or trace IDs needed for inspection and closes the client. Retained example records remain in the dedicated database; follow the Aura tutorial’s cleanup procedure for that dedicated instance only when you no longer need them. Reusing a session groups records; it does not authorize a user to read them.

Adapt the integration

Add more CrewAI agents and tasks inside run_crew() when the workflow needs them. Pass retrieved context in task descriptions and persist each completed task at the orchestration boundary if individual task provenance matters. A research/analysis/writing pipeline is an application design, not an automatic library feature.

Do not call synchronous Neo4jCrewMemory.remember() or recall() directly from the same running event loop as its connected async client. The current bridge submits work to that loop and then waits synchronously. The complete recipe avoids this bridge: the worker receives text only, and all memory I/O is awaited outside kickoff(). A worker must not create a new event loop for each operation on an already-connected client.

The shipped CrewAI adapter’s recall currently performs unscoped message search. This recipe selects Bolt and does not claim hosted compatibility or per-user isolation for that adapter. crew_id, user_id metadata and shared database access are distinct from authorization.

llm_provider_from_crewai(llm) accepts a configured CrewAI LLM for separately configured extraction. See Configure providers. For the rationale and consistency boundary of shared memory, see Multi-agent sharing. Await persistence and, when enabled, extraction readiness before a dependent task retrieves the result.

See Backend capabilities and scoping before adding hosted or multi-user behavior. Neo4j Agent Memory is a Neo4j Labs project with community support.