CrewAI integration

The zep-crewai package gives CrewAI agents persistent memory backed by Zep’s temporal knowledge graph. You persist conversation turns and business data with Zep storage adapters, and give your agents Zep tools so they can retrieve relevant context when they need it. This lets agents carry context across executions, share a common knowledge base, and ground their decisions in what was learned before.

Core benefits

• Persistent memory — Conversations and knowledge persist across sessions and crew runs.
• On-demand retrieval — Agents search Zep through tools and pull in context exactly when a task calls for it.
• Dual storage — User graphs for individual memory and shared Context Graphs for organizational data.
• Tool integration — Search and add-data tools let agents read from and write to Zep during execution.

How it works

Memory in this integration is explicit and tool-driven. There are two distinct steps, and you control both.

Persisting context. Create a ZepUserStorage or ZepGraphStorage adapter and call storage.save(value, metadata={"type": ...}). The adapter routes each item by its type:

Retrieving context. Attach create_search_tool (and optionally create_add_data_tool) to an Agent(tools=[...]). The agent searches Zep when it decides the task needs it. You can also call ZepUserStorage.get_context() directly to fetch the context block that Zep assembles for a thread.

Failure isolation. save() on every storage adapter logs a Zep failure and returns normally instead of raising, so a Zep outage never crashes a crew run. When you want misconfiguration to fail loudly, provision with ensure_user and ensure_thread before the crew runs — see provisioning users and threads.

There is no automatic retrieval or storage, and no external_memory= Crew wiring. CrewAI 1.x removed the ExternalMemory(storage=...) wrapper and the storage interface it depended on, so context is never injected behind the scenes. You decide what to save with save(...), and the agent decides what to search through its tools. The package is also sync-only: its adapters are built on the synchronous Zep client.

Identifiers

Zep v4 addresses every user, thread, and graph by a server-generated UUID. A resource that v4 creates has no client-chosen identifier: a create call does not accept a user_id, a thread_id, or a graph_id. For a resource that v3 created, a user_id or a thread_id is a read-only name, not an address.

The pattern throughout this integration is:

1. Create the resource one time, during onboarding.
2. Read the uuid_ field of the response.
3. Store the UUID in your own database.
4. Pass the UUID to the storage adapters (user_uuid, thread_uuid, graph_uuid) and to the tools (graph_uuid).

The integration does not call lookup at run time. If you carry a v3 identifier, resolve it one time with user.lookup or thread.lookup, store the returned UUID, and use the UUID from then on. Lookup is only for migration. See Migrating from v3.

Zep is also asynchronous: an episode you add is processed in the background, and a fact is not instantly retrievable. Give an agent time between save() and the search that reads it, or poll.

Installation

pip install zep-crewai

Set your API key in the environment:

export ZEP_API_KEY="your-zep-api-key"

Provisioning users and threads

ensure_user and ensure_thread are helpers for onboarding. Zep v4 accepts no client-chosen identifier on create, so a create cannot match an existing resource, and each successful call creates a new resource. The helpers never call lookup. Genuine failures (auth, network, 5xx) raise. Call them one time, during onboarding, store the returned UUIDs in your own database, and read the UUIDs from there on later runs.

The optional on_created hook fires once for each user the call creates — use it for one-time per-user setup such as ontology configuration. The hook receives the Zep client and the created User, which carries uuid_ and graph_uuid.

from zep_crewai import ensure_user, ensure_thread
def setup_new_user(client, user):
    client.graph.set_ontology(user.graph_uuid, entity_types=[...])
user, created = ensure_user(
    zep_client, first_name="Alice", on_created=setup_new_user
)
thread, _ = ensure_thread(zep_client, user_uuid=user.uuid_)
# Persist these with your own records; everything else in the package takes UUIDs.
user_uuid = user.uuid_
user_graph_uuid = user.graph_uuid
thread_uuid = thread.uuid_

ZepUserStorage and ZepStorage take UUIDs only and never create a user or a thread on the turn path. ZepGraphStorage is scoped to a Context Graph rather than a Zep user; passing it on_created raises TypeError.

Storage types

User storage

Use ZepUserStorage for an individual user’s conversation history and personal context. The thread_uuid ties message storage to a conversation thread. CrewAI has no automatic persistence loop; sharing one thread across multiple agents is safe when your code or tools write each turn once.

import os
from zep_cloud.client import Zep
from zep_crewai import ZepUserStorage, create_search_tool
from crewai import Agent
zep_client = Zep(api_key=os.getenv("ZEP_API_KEY"))
# Create the user and the thread one time; read the UUIDs from the responses.
# Your application stores these UUIDs alongside its own records.
user = zep_client.user.create(first_name="Alice", email="[email protected]")
thread = zep_client.thread.create(user_uuid=user.uuid_)
# Create user storage
user_storage = ZepUserStorage(
    client=zep_client,
    user_uuid=user.uuid_,
    thread_uuid=thread.uuid_,
    graph_uuid=user.graph_uuid,
)
# Persist a conversation turn (routes to the thread)
user_storage.save(
    "How can I help you today?",
    metadata={"type": "message", "role": "assistant", "name": "Helper"},
)
# Persist a preference as graph data
user_storage.save(
    "Alice prefers morning meetings",
    metadata={"type": "text"},
)
# Give an agent a Zep search tool so it can retrieve this context on demand
assistant = Agent(
    role="Personal Assistant",
    goal="Help Alice using what you know about her",
    backstory="You know Alice's preferences and conversation history.",
    tools=[create_search_tool(zep_client, graph_uuid=user.graph_uuid)],
)

graph_uuid is optional: when you omit it, the storage reads user.graph_uuid one time with user.get(user_uuid) and caches it. Pass it explicitly when your application already stored it — that avoids the extra read.

To fetch the auto-assembled Context Block for the thread directly, call get_context():

# Returns a context block string for your provider's data channel
context = user_storage.get_context()
print(context)

Graph storage

Use ZepGraphStorage for a shared Context Graph that multiple agents can read and write. Create the graph with graph.create and keep the uuid_ of the response.

from zep_cloud import SearchFilters
from zep_crewai import ZepGraphStorage, create_search_tool
from crewai import Agent
# Create the graph; the response carries the graph UUID.
graph = zep_client.graph.create(
    name="Company Knowledge Graph",
    description="Shared organizational knowledge and insights.",
)
# Create graph storage for shared knowledge
graph_storage = ZepGraphStorage(
    client=zep_client,
    graph_uuid=graph.uuid_,
    search_filters=SearchFilters(node_labels=["Technology", "Project"]),
)
# Persist knowledge
graph_storage.save(
    "Project Atlas uses Python and React",
    metadata={"type": "text"},
)
# Let agents search it through a tool
knowledge_agent = Agent(
    role="Knowledge Assistant",
    goal="Answer questions from the shared Context Graph",
    backstory="You maintain and search the team's shared knowledge.",
    tools=[create_search_tool(zep_client, graph_uuid=graph.uuid_)],
)

You can also search a graph directly. search returns a list whose entries include a Context Block wrapped in the storage’s context template — a <ZEP_CONTEXT>...</ZEP_CONTEXT> block by default:

results = graph_storage.search("project status", limit=5)
for item in results:
    print(item.get("context", ""))  # <ZEP_CONTEXT> ... </ZEP_CONTEXT>

Pass context_template="{context}" to the storage constructor to get the bare context string instead.

Customizing retrieved context

Both storage classes wrap the context string returned from search() in a template. context_template must contain a literal {context} placeholder and is rendered with plain string replacement (never str.format), so context containing {, }, or % is always safe. The default is the DEFAULT_CONTEXT_TEMPLATE export — the <ZEP_CONTEXT>...</ZEP_CONTEXT> block shared across Zep integrations.

ZepUserStorage also accepts a context_builder: a synchronous callable that replaces the default graph.get_context retrieval in search() with your own logic. The builder receives a frozen ContextInput (zep, user_uuid, thread_uuid, graph_uuid, user_message) and returns the context string, or None for no results. A builder exception is logged and degrades to empty results.

from zep_crewai import ZepUserStorage, ContextInput
def my_builder(ctx: ContextInput) -> str | None:
    edges = list(
        ctx.zep.graph.search_edges(ctx.graph_uuid, query=ctx.user_message, limit=10)
    )
    facts = [edge.fact for edge in edges if edge.fact]
    return "\n".join(facts) if facts else None
storage = ZepUserStorage(
    client=zep_client,
    user_uuid=user.uuid_,
    thread_uuid=thread.uuid_,
    graph_uuid=user.graph_uuid,
    context_builder=my_builder,
)

Tool integration

Tools are the supported extension point for exposing Zep to CrewAI agents. Every tool binds to one graph at creation time — the user graph of a user, or a standalone graph — then goes on an agent’s tools list.

from zep_cloud import SearchFilters
from zep_crewai import create_search_tool, create_add_data_tool
from crewai import Agent
# Tools bound to the user graph
user_search_tool = create_search_tool(zep_client, graph_uuid=user.graph_uuid)
user_add_tool = create_add_data_tool(zep_client, graph_uuid=user.graph_uuid)
# Tools bound to a standalone graph
graph = zep_client.graph.create(name="knowledge base")
graph_search_tool = create_search_tool(zep_client, graph_uuid=graph.uuid_)
graph_add_tool = create_add_data_tool(zep_client, graph_uuid=graph.uuid_)
curator = Agent(
    role="Knowledge Curator",
    goal="Search existing knowledge and record new findings",
    backstory="You maintain the organization's knowledge base.",
    tools=[graph_search_tool, graph_add_tool],
    llm="gpt-5.6-terra",
)

create_search_tool and create_add_data_tool return ZepSearchTool and ZepAddDataTool instances; both classes are also exported if you prefer to construct them directly. A Zep failure inside either tool returns an error string to the agent — the tool never raises into the crew.

Search tool parameters

The search tool’s args_schema exposes every graph search parameter to the model by default. In Zep v4 each scope value calls a dedicated SDK method — graph.search_edges, graph.search_nodes, graph.search_episodes, graph.search_observations, or graph.search_thread_summaries — and auto calls graph.get_context so Zep assembles the mix on the server. Each scoped method returns a pager, which the tool reads up to limit.

Use pinned_params to fix a parameter to a constant and remove it from the model-facing schema, or hidden_params to remove it from the schema without pinning it (Zep’s server-side default applies). The scope, reranker, and limit keyword arguments each pin and hide their parameter, equivalent to putting them in pinned_params. search_filters and bfs_origin_node_uuids are constructor-only and never exposed to the model.

# Pin scope and limit (hidden from the model, always sent); hide reranker entirely
search_tool = create_search_tool(
    zep_client,
    graph_uuid=user.graph_uuid,
    pinned_params={"scope": "edges", "limit": 5},
    hidden_params={"reranker"},
)
# Constructor-only parameters are never exposed to the model
search_tool = create_search_tool(
    zep_client,
    graph_uuid=graph.uuid_,
    search_filters=SearchFilters(node_labels=["Project"]),
    bfs_origin_node_uuids=["node-uuid-1"],
)

The tool returns results to the agent as plain - fact lines, one per result.

Add-data tool parameters

• data — Content to store; payloads over Zep’s graph.episode.add ceiling are truncated to 9,900 characters instead of failing.
• data_type — Explicit type: text (default), json, or message.

Structured data with ontologies

Define entity types so Zep organizes graph data into typed entities, then set the ontology on the graph.

from zep_cloud.types import EntityProperty, EntityType
from zep_cloud import SearchFilters
from zep_crewai import ZepGraphStorage
project_entity = EntityType(
    name="Project",
    description="a project tracked in the knowledge graph",
    properties=[
        EntityProperty(name="status", type="text", description="project status"),
        EntityProperty(name="priority", type="text", description="priority level"),
        EntityProperty(name="team_size", type="text", description="team size"),
    ],
)
# Create the graph and set its ontology, addressing it by UUID
graph = zep_client.graph.create(name="projects")
zep_client.graph.set_ontology(graph.uuid_, entity_types=[project_entity])
# Use the graph with filtered search and a context size limit
graph_storage = ZepGraphStorage(
    client=zep_client,
    graph_uuid=graph.uuid_,
    search_filters=SearchFilters(node_labels=["Project"]),
    max_characters=6000,
)

Configuration options

ZepUserStorage parameters

ZepGraphStorage parameters

ZepGraphStorage has no on_created parameter — it is graph-scoped, with no Zep user to provision. Passing on_created raises TypeError.

ZepStorage parameters

ZepStorage is a standalone user-and-thread adapter that preserves the historical save(value, metadata) / search(query, limit, score_threshold) / reset() contract for existing callers. It takes client, user_uuid, and thread_uuid (all required) plus an optional graph_uuid. New code should prefer ZepUserStorage.

Size limits

• Zep rejects direct thread-message payloads over 4,096 characters; the CrewAI storage paths truncate message content to 4,000 characters before thread.add_messages, logging lengths only and never content.
• Zep rejects direct graph.episode.add payloads over 10,000 characters; CrewAI storage paths and ZepAddDataTool truncate graph payloads to 9,900 characters before calling graph.episode.add.
• The integration truncates search queries to 400 characters.

Complete example

This example mirrors the simple_example.py from the integration repository. It persists conversation turns and business data to a user’s memory, then runs an agent that searches that memory through a Zep tool before answering.

import os
import sys
import time
from crewai import Agent, Crew, Process, Task
from zep_cloud.client import Zep
from zep_crewai import ZepUserStorage, create_search_tool
def main():
    api_key = os.environ.get("ZEP_API_KEY")
    if not api_key:
        print("Error: set your ZEP_API_KEY environment variable")
        print("Get your API key from: https://app.getzep.com")
        sys.exit(1)
    zep_client = Zep(api_key=api_key)
    # Create the user and the thread; read the UUIDs from the responses and
    # keep them with your own records.
    user = zep_client.user.create(
        first_name="John", last_name="Doe", email="[email protected]"
    )
    thread = zep_client.thread.create(user_uuid=user.uuid_)
    # Initialize the Zep storage adapter
    user_storage = ZepUserStorage(
        client=zep_client,
        user_uuid=user.uuid_,
        thread_uuid=thread.uuid_,
        graph_uuid=user.graph_uuid,
    )
    # Persist context with metadata-based routing
    # JSON data routes to the graph
    user_storage.save(
        '{"trip_type": "business", "destination": "New York", "duration": "3 days", '
        '"budget": 2000, "accommodation_preference": "mid-range hotels"}',
        metadata={"type": "json"},
    )
    # Messages route to the thread
    user_storage.save(
        "Hi, I need help planning a business trip to New York. I'll be there for 3 "
        "days and prefer mid-range hotels.",
        metadata={"type": "message", "role": "user", "name": "John Doe"},
    )
    user_storage.save(
        "I'd be happy to help you plan your New York business trip!",
        metadata={"type": "message", "role": "assistant", "name": "Travel Planning Assistant"},
    )
    # Text data routes to the graph
    user_storage.save(
        "John Doe prefers mid-range hotels with business amenities, enjoys local "
        "cuisine, and values convenient locations near business districts.",
        metadata={"type": "text"},
    )
    user_storage.save(
        "John Doe's budget constraint: around $2000 total for the trip including "
        "flights and accommodation. Looking for good value rather than luxury.",
        metadata={"type": "text"},
    )
    # Ingestion is asynchronous — allow time for indexing before the agent searches
    time.sleep(20)
    # Give the agent a Zep search tool bound to the user graph
    search_tool = create_search_tool(zep_client, graph_uuid=user.graph_uuid)
    travel_agent = Agent(
        role="Travel Planning Assistant",
        goal="Help plan business trips efficiently and within budget",
        backstory="""You are an experienced travel planner who specializes in business
        trips. You always consider the user's preferences, budget, and trip context.
        Use the Zep memory search tool to recall what you know about the user before
        answering.""",
        tools=[search_tool],
        verbose=True,
        llm="gpt-5.6-terra",
    )
    planning_task = Task(
        description="""First, search Zep memory for the user's saved preferences and
        trip context. Then provide 3 specific hotel recommendations in New York that
        would be good for a business traveler. Include hotel names and locations, price
        range per night, why each fits the user's preferences, and any business
        amenities.""",
        expected_output="A list of 3 hotel recommendations with detailed explanations",
        agent=travel_agent,
    )
    crew = Crew(
        agents=[travel_agent],
        tasks=[planning_task],
        process=Process.sequential,
        verbose=True,
    )
    result = crew.kickoff()
    print(result)
    # Optionally persist the result for future runs
    user_storage.save(str(result), metadata={"type": "message", "role": "assistant"})
if __name__ == "__main__":
    main()

Best practices

Storage selection

• Use ZepUserStorage for personal preferences, conversation history, and user-specific context.
• Use ZepGraphStorage for organizational and collaborative data in a shared Context Graph.

Memory management

• Store UUIDs — keep user.uuid_, thread.uuid_, and graph.uuid_ in your own database and address resources by UUID.
• Set up ontologies for structured graph data with EntityType and EntityProperty.
• Use search filters to target specific node types and improve relevance.
• Combine storage types when the agent needs multiple memory types.

Tool usage

• Bind tools to a specific graph at creation time — a user graph or a standalone graph.
• Pin or hide search parameters the model should not control with pinned_params and hidden_params.
• Save data with the right type (message, json, or text) so it routes correctly.
• Allow time for indexing — Zep extracts knowledge asynchronously, so facts from a turn are not instantly searchable.

Next steps

• Explore customizing graph structure for advanced knowledge organization
• Learn about searching the graph and how to tune search
• See code examples for additional patterns