AG2 integration

AG2 agents using Zep maintain context across conversations and access a temporal knowledge graph. The zep-ag2 package provides search and data tools. It also provides an automatic system-message path for trusted content.

Use the Zep search tools for context that can contain end-user or third-party data. Use the automatic loop’s system-message injection only for fully trusted, application-authored context.

Core benefits

• Automatic memory loop for trusted content: attach_to_agent persists every message and can refresh the system message with application-authored, trusted context
• Persistent memory: Conversations and extracted knowledge persist across sessions
• System message injection for trusted content: Relevant context can be added to an agent’s system message before it responds
• Knowledge graph access: Search and write to Zep’s temporal knowledge graph from AG2 agents
• Tool-based access: Register Zep search and add operations as AG2 tools the agent invokes on demand

How it works

AG2 has no native memory interface, so the integration provides three ways to give an agent memory:

• Automatic memory loop — ZepMemoryManager.attach_to_agent(agent) registers hooks on ConversableAgent that persist every message the agent receives and sends, and refresh its system message with relevant context on each turn. Use this path only for fully trusted, application-authored context.
• System message injection for trusted content — ZepMemoryManager and ZepGraphMemoryManager can add trusted context to an agent’s system message.
• Tools — factory functions return AG2-compatible tools the model can call mid-conversation to search memory or write new data. AG2 runs tools synchronously, so the tools take a synchronous Zep client and call it directly.

Use tools for end-user or third-party context. Attach the automatic loop only when all stored content is application-authored and trusted.

Installation

pip install "zep-ag2>=1.0.0"

Set up your environment variables:

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

Identifiers

Zep v4 addresses every user, thread, and graph by a server-generated UUID. A create call takes no client-chosen name. The public API of zep-ag2 takes user_uuid, thread_uuid, and graph_uuid.

Create the user and the thread one time with create_user and create_thread, read the UUIDs from the responses, and store them in your own database. The package does not resolve a name at run time.

from zep_ag2 import create_thread, create_user
user = await create_user(zep, first_name="Jane", email="jane@example.com")
thread = await create_thread(zep, user_uuid=user.uuid_)
# Store these three values in your own database.
user.uuid_, user.graph_uuid, thread.uuid_

The user has a personal graph. Pass user.graph_uuid as the graph_uuid of the managers and of the tool factories that read or write that graph.

Migrating from V3

Follow the V3 to V4 migration guide for the general procedure. The API changes that are specific to zep-ag2 1.0.0 are:

• ensure_user(client, *, user_id, ...) and ensure_thread(client, *, thread_id, user_id) are removed. create_user(client, *, first_name=None, last_name=None, email=None, on_created=None) and create_thread(client, *, user_uuid) replace them. They take no client-chosen name, and they return the created User and Thread. The on_created hook now receives the UUID of the user.
• ZepMemoryManager(client, user_id, session_id=None, ...) is now ZepMemoryManager(client, user_uuid, thread_uuid=None, *, graph_uuid=None, sync_client=None, ...). Give the stored user UUID, thread UUID, and user graph UUID here. The user_id and session_id properties are now user_uuid and thread_uuid.
• The lazy provisioning path of ZepMemoryManager is removed: ensure_user_and_thread(), and the constructor arguments first_name, last_name, email, and on_created.
• ZepGraphMemoryManager(client, graph_id) is now ZepGraphMemoryManager(client, graph_uuid, *, sync_client=None). The graph_id property is now graph_uuid.
• create_search_memory_tool(client, user_id, session_id=None, ...) is now create_search_memory_tool(client, graph_uuid, ...). Give the stored user graph UUID.
• create_add_memory_tool(client, user_id, session_id=None) is now create_add_memory_tool(client, graph_uuid, thread_uuid=None). Give the stored user graph UUID and thread UUID.
• create_search_graph_tool(client, user_id=None, graph_id=None, ...) and create_add_graph_data_tool(client, user_id=None, graph_id=None) now take one graph_uuid. Give the stored user graph UUID or the stored UUID of a named graph.
• The search_filters argument of create_search_memory_tool and create_search_graph_tool is now filters.
• register_all_tools(agent, executor, client, user_id, session_id=None, graph_id=None) is now register_all_tools(agent, executor, client, graph_uuid, thread_uuid=None, memory_graph_uuid=None). The graph tools use graph_uuid. The memory tools use memory_graph_uuid, which defaults to graph_uuid. When graph_uuid is a named graph, give the stored user graph UUID as memory_graph_uuid.
• The tool factories and register_all_tools take a synchronous zep_cloud.client.Zep client. An AsyncZep client raises TypeError when you create the tools.
• The user_id and thread_id fields of ContextInput are now user_uuid and thread_uuid.

Automatic memory loop for trusted deployments

attach_to_agent writes conversation content to memory and later inserts that memory into a system message. Do not use this path when a user or external source can influence stored content.

import os
from autogen import AssistantAgent, UserProxyAgent, LLMConfig
from zep_cloud.client import AsyncZep, Zep
from zep_ag2 import ZepMemoryManager
zep = AsyncZep(api_key=os.environ["ZEP_API_KEY"])
# The hooks of the automatic loop run synchronously and use the sync client.
sync_zep = Zep(api_key=os.environ["ZEP_API_KEY"])
# The UUIDs come from create_user and create_thread, and your application
# stores them. See Identifiers above.
user_uuid, graph_uuid, thread_uuid = load_zep_uuids()
llm_config = LLMConfig(
    {"model": "gpt-5.6-terra", "api_key": os.environ["OPENAI_API_KEY"]}
)
assistant = AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="You are a helpful assistant with long-term memory.",
)
user_proxy = UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config=False,
)
# Trusted-only path: attach_to_agent inserts stored content into a system message.
manager = ZepMemoryManager(
    zep,
    user_uuid=user_uuid,
    thread_uuid=thread_uuid,
    graph_uuid=graph_uuid,
    sync_client=sync_zep,
)
manager.attach_to_agent(assistant)
# Every message the assistant receives is persisted and used to refresh its
# system message; every reply it sends is persisted automatically too.
user_proxy.initiate_chat(assistant, message="My name is Alice.")

attach_to_agent(agent) registers two hooks on AG2’s ConversableAgent:

• process_last_received_message fires for every message the agent receives. It persists the message and retrieves fresh context with the synchronous sync_client of the manager, then replaces the agent’s system message with its original text plus the rendered context template. The hook returns the message content unmodified — it is a side channel, not a message transform.
• process_message_before_send fires for every message the agent sends. It persists the outgoing message as an assistant message and returns it unchanged.

Both hooks wrap their entire body in error handling, so a Zep outage never breaks the agent’s conversation loop — on failure, the incoming hook skips the system-message update and the outgoing hook skips persistence, in both cases still returning the message unchanged.

attach_to_agent is optional and additive: enrich_system_message and add_messages remain available for manual control, for example to persist only some turns or inject context at a different point than “on receive”.

Manual system-message injection for trusted deployments

Use ZepMemoryManager.enrich_system_message only when all stored content is fully trusted and application-authored:

import asyncio
import os
from autogen import AssistantAgent, UserProxyAgent, LLMConfig
from zep_cloud.client import AsyncZep, Zep
from zep_ag2 import ZepMemoryManager, create_thread, create_user, register_all_tools
async def main():
    # The async helpers use AsyncZep. The synchronous AG2 tools use Zep.
    zep = AsyncZep(api_key=os.environ["ZEP_API_KEY"])
    sync_zep = Zep(api_key=os.environ["ZEP_API_KEY"])
    # Create the user and the thread one time, and store the UUIDs.
    user = await create_user(zep, first_name="Jane", email="jane@example.com")
    thread = await create_thread(zep, user_uuid=user.uuid_)
    user_uuid, graph_uuid, thread_uuid = user.uuid_, user.graph_uuid, thread.uuid_
    llm_config = LLMConfig(
        {"model": "gpt-5.6-terra", "api_key": os.environ["OPENAI_API_KEY"]}
    )
    assistant = AssistantAgent(
        name="assistant",
        llm_config=llm_config,
        system_message="You are a helpful assistant with long-term memory.",
    )
    user_proxy = UserProxyAgent(
        name="user",
        human_input_mode="NEVER",
        code_execution_config=False,
        is_termination_msg=lambda msg: "TERMINATE" in (msg.get("content") or ""),
    )
    # Trusted-only path: never pass end-user or third-party memory here.
    memory_mgr = ZepMemoryManager(
        zep, user_uuid=user_uuid, thread_uuid=thread_uuid, graph_uuid=graph_uuid
    )
    await memory_mgr.enrich_system_message(assistant, query="conversation topic")
    # Register Zep memory tools. AG2 calls them synchronously, so they take the sync client.
    register_all_tools(assistant, user_proxy, sync_zep, graph_uuid, thread_uuid)
    user_proxy.initiate_chat(assistant, message="What do you remember about me?")
asyncio.run(main())

ZepMemoryManager also exposes process_user_message() to persist a user turn and retrieve context in one call, get_memory_context() to retrieve the formatted context string directly, add_messages() to persist conversation turns, and get_session_facts() to read the thread’s context block.

Provisioning

The manager does not create Zep resources. Create the user and the thread out of band with create_user and create_thread, so a failure is raised before the first turn.

Pass first_name, last_name, and email so Zep can anchor the identity node of the user in the graph, and on_created to run one-time setup, such as an ontology or custom instructions:

from zep_ag2 import create_thread, create_user
async def setup_new_user(zep, user_uuid: str) -> None:
    ...  # one-time setup: ontology, custom instructions
user = await create_user(
    zep,
    first_name="Jane",
    last_name="Smith",
    email="jane@example.com",
    on_created=setup_new_user,
)
thread = await create_thread(zep, user_uuid=user.uuid_)

Both helpers raise on a failure, and create_user propagates an exception from the on_created hook. Make the hook idempotent, so that you can run it again for a user whose setup only partially completed.

A ZepMemoryManager is scoped to one (user_uuid, thread_uuid) pair for the lifetime of the instance. Create one manager for each user and thread, and do not share an instance between users.

Tool integration

Register Zep operations as AG2 tools so the agent can search memory or write new data during a conversation. register_all_tools wires up the full set in one call, or use the individual factories for finer control:

from zep_ag2 import create_search_graph_tool, create_add_graph_data_tool
# Create tools bound to the personal graph of the user
search_tool = create_search_graph_tool(sync_zep, graph_uuid)
add_tool = create_add_graph_data_tool(sync_zep, graph_uuid)
# Register with AG2's decorator pattern
assistant.register_for_llm(description="Search knowledge graph")(search_tool)
user_proxy.register_for_execution()(search_tool)
assistant.register_for_llm(description="Add to knowledge graph")(add_tool)
user_proxy.register_for_execution()(add_tool)

Available tool factories:

• create_search_memory_tool(client, graph_uuid, *, pinned_params=None, hidden_params=None, filters=None, bfs_origin_node_uuids=None, scope=None, limit=None) — search the graph
• create_add_memory_tool(client, graph_uuid, thread_uuid=None) — write to the thread when a thread_uuid is set, and to the graph when it is not
• create_search_graph_tool(client, graph_uuid, *, pinned_params=None, hidden_params=None, filters=None, bfs_origin_node_uuids=None, scope=None, limit=None) — search the knowledge graph
• create_add_graph_data_tool(client, graph_uuid) — add data to the knowledge graph
• register_all_tools(agent, executor, client, graph_uuid, thread_uuid=None, memory_graph_uuid=None) — register all tools at once

Every tool factory and register_all_tools take a synchronous zep_cloud.client.Zep client. An AsyncZep client raises TypeError when you create the tools.

Every tool is bound to one graph_uuid. Use user.graph_uuid for the personal graph of a user, or the UUID of a shared Context Graph for shared knowledge.

Search tool parameters

The search tool factories follow a pin-or-expose pattern: every search parameter is exposed to the model by default, each with a typed schema and documented default. Letting the model choose the scope and reranker per query produces better retrieval than a single fixed configuration; pin parameters when you need deterministic behavior instead.

Use pinned_params to fix a parameter to a constant (hidden from the model), or hidden_params to remove it from the schema without pinning (Zep’s server-side default applies):

# Model chooses scope/reranker/limit/mmr_lambda/center_node_uuid freely (default)
tool = create_search_graph_tool(sync_zep, graph_uuid)
# Pin scope to "nodes" and limit to 5 — hidden from the model, always sent as given
tool = create_search_graph_tool(
    sync_zep, graph_uuid, pinned_params={"scope": "nodes", "limit": 5}
)
# Hide reranker entirely — omitted from the schema and the SDK call
tool = create_search_graph_tool(sync_zep, graph_uuid, hidden_params={"reranker"})

The legacy scope and limit keyword arguments pin (and hide) the corresponding parameter — equivalent to passing them via pinned_params. filters and bfs_origin_node_uuids are constructor-only and never exposed to the model. Each search scope accepts only its own filter keys, and Zep rejects a key that the scope does not support.

Each scope accepts only some rerankers:

• edges accepts rrf, mmr, node_distance, episode_mentions, and cross_encoder.
• nodes accepts rrf, mmr, node_distance, and cross_encoder.
• episodes accepts rrf and mmr.
• observations and thread_summaries accept rrf, mmr, and cross_encoder.
• auto takes no reranker. The auto scope calls graph.get_context, which also takes no limit.

The tool schema offers only valid choices. episode_mentions appears only when filters sets episode_uuids, and node_distance needs center_node_uuid. When a reranker is pinned, the schema does not offer the auto scope.

If the model sends a scope and reranker pair that Zep rejects, the tool does not call Zep. The tool returns an error message that names the pair and lists the valid rerankers for the scope, so that the model can call the tool again. A reranker with the auto scope also returns an error message. A pinned pair that is not valid, which includes a pinned reranker with the auto scope, raises ValueError when you create the tool.

Shared Context Graph memory

Use ZepGraphMemoryManager to work with a shared Context Graph that is addressed with graph_uuid and that multiple agents can read and write:

from zep_ag2 import ZepGraphMemoryManager
# Create the graph one time. Store graph.uuid_ in your application database.
graph = await zep.graph.create(name="company_knowledge")
graph_mgr = ZepGraphMemoryManager(zep, graph_uuid=graph.uuid_)
# Add data to the graph
await graph_mgr.add_data("Project Atlas uses Python and React.", data_type="text")
# Search the graph
results = await graph_mgr.search("What technologies does Project Atlas use?", limit=5, scope="edges")

ZepGraphMemoryManager.search() accepts scope values edges, nodes, and episodes and returns structured result dicts for programmatic use. This is distinct from the search tool schema above, which exposes six scopes to the model and returns formatted strings.

Use AG2 search tools or place these results through your provider’s data channel for end-user or third-party graph content. enrich_system_message is only for a fully trusted, application-authored graph.

Custom context retrieval

By default, context is retrieved via thread.get_context(...) (or, inside process_user_message, via thread.add_messages(..., return_context=True)). Pass context_builder to replace this with custom logic — for example a filtered graph search, or a different graph entirely:

from zep_ag2.memory import ContextInput
async def my_builder(ctx: ContextInput) -> str | None:
    results = await ctx.zep.graph.search_edges(
        graph_uuid,
        query=ctx.user_message,
    )
    if not results.items:
        return None
    return "\n".join(edge.fact for edge in results.items)
manager = ZepMemoryManager(
    zep,
    user_uuid=user_uuid,
    thread_uuid=thread_uuid,
    graph_uuid=graph_uuid,
    context_builder=my_builder,
)

The builder receives a single frozen ContextInput:

If the builder raises, a warning is logged and context injection is skipped for that call — the builder never raises into process_user_message, get_memory_context, or enrich_system_message. Inside process_user_message, persistence and the builder run concurrently with per-side isolation: a builder failure never blocks the message from being persisted, and a persistence failure never prevents the builder’s context from being returned.

Customizing the injected context template

Retrieved context (from the default retrieval or a context_builder) is wrapped in context_template before injection into the agent’s system message. The default DEFAULT_CONTEXT_TEMPLATE wraps the context in <ZEP_CONTEXT> tags with a short preamble. Override it with your own wording, as long as it contains a literal {context} placeholder:

manager = ZepMemoryManager(
    zep,
    user_uuid=user_uuid,
    thread_uuid=thread_uuid,
    context_template="Relevant background:\n{context}",
)

The template uses plain string replacement (template.replace("{context}", ...)), never str.format. This prevents format-string interpretation of {, }, or %. It does not prevent the model from following instructions in the retrieved content.

Query memory

You can read memory directly, outside of agent tool calls:

# Formatted context block for the user/thread (optionally biased by a query)
context = await memory_mgr.get_memory_context(query="project status", limit=5)
# Facts extracted from the current session
facts = await memory_mgr.get_session_facts()
# Structured search over a knowledge graph
results = await graph_mgr.search("Project Atlas", limit=5)

Search result structure

The tool factories return human-readable strings formatted for the model, with formatting that adapts to the search scope. ZepGraphMemoryManager.search() returns a list of structured result dicts for programmatic use; the fields depend on the scope:

Memory vs tools

The integration supports three patterns:

Use tools as the default for context that users or external sources can influence.

Configuration options

ZepMemoryManager

• ZepMemoryManager(client, user_uuid, thread_uuid=None, *, graph_uuid=None, sync_client=None, context_builder=None, context_template=DEFAULT_CONTEXT_TEMPLATE) — initialize with an AsyncZep client and the UUIDs of the user and the thread; the configuration arguments are keyword-only. The synchronous methods call sync_client, a zep_cloud.client.Zep client, and raise TypeError when it is not set.
• attach_to_agent(agent) — register the automatic inject and persist loop; needs sync_client
• get_memory_context_sync(query=None, limit=5) and enrich_system_message_sync(agent, query=None, limit=5) — synchronous forms; need sync_client
• process_user_message(user_message, *, agent=None) — persist a user turn and retrieve context in one call
• resolve_graph_uuid() — return the graph UUID, and read it one time from the user when the constructor did not receive it
• enrich_system_message(agent, query=None, limit=5) — inject memory context into an agent
• get_memory_context(query=None, limit=5) — return the formatted context string
• add_messages(messages) — store messages in the Zep thread
• get_session_facts() — read the thread’s context block

ZepGraphMemoryManager

• ZepGraphMemoryManager(client, graph_uuid, *, sync_client=None) — initialize with an AsyncZep client and the UUID of a graph
• search_sync(query, limit=5, scope="edges") and add_data_sync(data, data_type="text") — synchronous forms; need sync_client, a zep_cloud.client.Zep client
• search(query, limit=5, scope="edges") — search the graph (scope: edges, nodes, episodes)
• add_data(data, data_type="text") — add data to the graph (data_type: text, json, message)
• enrich_system_message(agent, query=None, limit=5) — inject graph context into an agent

Provisioning helpers

• create_user(client, *, first_name=None, last_name=None, email=None, on_created=None) — create a Zep user and return it; read user.uuid_ and user.graph_uuid from the response
• create_thread(client, *, user_uuid) — create a Zep thread and return it; read thread.uuid_ from the response

Size limits

Zep rejects over-long direct SDK payloads with an HTTP 400. The AG2 integration truncates before calling Zep, logging only the before and after lengths (never the content):

• Thread messages: truncated to 4,000 characters, a safety margin under Zep’s 4,096-character thread-message limit
• Graph data (add_data, create_add_graph_data_tool): truncated to 9,900 characters, a safety margin under Zep’s 10,000-character graph.episode.add limit

Best practices

• Use the correct client — pass AsyncZep to the async helpers and the managers, and a synchronous Zep client to the tools and to sync_client
• Attach one agent per Zep thread — two managers with the same thread_uuid persist every turn twice with conflicting roles
• Bind tools to one graph — the personal graph of a user for personal memory, or a shared Context Graph for shared knowledge
• Use tools for untrusted context. Attach the automatic loop only when all stored content is application-authored and trusted.
• Allow time for indexing — Zep extracts knowledge asynchronously, so data added during a turn is 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