Zep
Persist and recall user facts across sessions with ZepTools and ZepAsyncTools injected as agent context.
Use ZepTools and ZepAsyncTools to add messages to Zep, retrieve user context, and search the user’s knowledge graph.
Prerequisites
- Get your Zep API key from https://app.getzep.com/
- Install dependencies:
uv pip install -U agno openai zep-cloud. - Set required environment variables:
export ZEP_API_KEY=<your-zep-api-key>andexport OPENAI_API_KEY=<your-openai-api-key>.
import asyncio
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.zep import ZepAsyncTools, ZepTools
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
def run_sync() -> None:
# Initialize the ZepTools
sync_zep_tools = ZepTools(
user_id="agno", session_id="agno-session", add_instructions=True
)
# Initialize the Agent
sync_agent = Agent(
model=OpenAIChat(),
tools=[sync_zep_tools],
dependencies={"memory": sync_zep_tools.get_zep_memory(memory_type="context")},
add_dependencies_to_context=True,
)
# Interact with the Agent so that it can learn about the user
sync_agent.print_response("My name is John Billings")
sync_agent.print_response("I live in NYC")
sync_agent.print_response("I'm going to a concert tomorrow")
# Allow the memories to sync with Zep database
time.sleep(10)
if sync_agent.dependencies:
# Refresh the context
sync_agent.dependencies["memory"] = sync_zep_tools.get_zep_memory(
memory_type="context"
)
# Ask the Agent about the user
sync_agent.print_response("What do you know about me?")
# ---------------------------------------------------------------------------
# Async Variant
# ---------------------------------------------------------------------------
async def run_async() -> None:
# Initialize the ZepAsyncTools
async_zep_tools = ZepAsyncTools(
user_id="agno", session_id="agno-async-session", add_instructions=True
)
# Initialize the Agent
async_agent = Agent(
model=OpenAIChat(),
tools=[async_zep_tools],
dependencies={
"memory": lambda: async_zep_tools.get_zep_memory(memory_type="context"),
},
add_dependencies_to_context=True,
)
# Interact with the Agent
await async_agent.aprint_response("My name is John Billings")
await async_agent.aprint_response("I live in NYC")
await async_agent.aprint_response("I'm going to a concert tomorrow")
# Allow the memories to sync with Zep database
time.sleep(10)
# Refresh the context
async_agent.dependencies["memory"] = await async_zep_tools.get_zep_memory(
memory_type="context"
)
# Ask the Agent about the user
await async_agent.aprint_response("What do you know about me?")
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
run_sync()
asyncio.run(run_async())The toolkit saves messages when add_zep_message is called; attaching it does not automatically persist every turn. The sync agent loads a context string at construction and explicitly refreshes it later. The async dependency lambda returns a coroutine that Agno awaits during an async run.
Replace time.sleep(10) inside run_async with await asyncio.sleep(10) so it does not block the event loop. The delay is illustrative and does not prove graph processing has finished. Both variants use user ID agno with different thread IDs, so they share a user graph. Use application user IDs and thread IDs for real users. Reusing these IDs on subsequent executions also reuses their remote memory.
Run the Example
# Clone and setup repo
git clone https://github.com/agno-agi/agno.git
cd agno
git checkout d703c34f3abf3c41275d3fb2da6e0518a8881f24
# Create and activate virtual environment
./scripts/demo_setup.sh
source .venvs/demo/bin/activate
uv pip install -U agno openai zep-cloud
python cookbook/91_tools/zep_tools.pyFor details, see Zep tools cookbook.