Zep Pydantic AI Integration

A memory integration package that gives Pydantic AI agents long-term memory powered by Zep. User turns are persisted to Zep and relevant context from Zep's temporal knowledge graph is injected into the model prompt on every turn -- using Pydantic AI's native ProcessHistory capability -- plus an on-demand graph-search tool.

Installation

pip install zep-pydantic-ai

See SETUP.md for how to sign up for Zep, create an API key, configure your environment, and run the example.

Quick Start

import asyncio
from pydantic_ai import Agent
from zep_cloud.client import AsyncZep
from zep_pydantic_ai import ZepDeps, create_zep_search_tool, zep_capabilities

zep = AsyncZep(api_key="your-zep-api-key")

deps = ZepDeps(
    client=zep,
    user_id="user_123",
    thread_id="thread_abc",
    first_name="Jane",
    last_name="Smith",
)

agent = Agent(
    "openai:gpt-5-mini",
    deps_type=ZepDeps,
    capabilities=zep_capabilities(deps),
    tools=[create_zep_search_tool()],
    instructions="You are a helpful assistant with long-term memory.",
)

async def main() -> None:
    result = await agent.run("What did I tell you about my project?", deps=deps)
    print(result.output)
    # The user turn and the assistant's reply are both already persisted --
    # zep_capabilities(deps) wires in automatic assistant persistence.

asyncio.run(main())

Prefer explicit control over when the assistant's reply is persisted? Use capabilities=[ProcessHistory(zep_history_processor)] and call persist_run yourself -- see persist_run below.

How It Works

ZepDeps

A dataclass used as the agent's deps_type. It carries the Zep client and the user/thread identity (plus optional name, email, and display names). Construct one per conversation and pass it to agent.run(..., deps=deps); the history processor and the search tool both reach it via RunContext.deps. The Zep user and thread are created lazily on first use -- you do not have to pre-create them, though doing so out-of-band with ensure_user/ensure_thread is slightly faster on the first turn (see Provisioning below).

zep_history_processor

Registered via capabilities=[ProcessHistory(zep_history_processor)] (or via zep_capabilities(deps), which includes it). Pydantic AI runs a history processor immediately before every model request. On the user's turn this processor:

1. resolves the Zep client and identity from ctx.deps;
2. lazily creates the Zep user and thread;
3. persists the latest user message -- via thread.add_messages(return_context=True) by default (a single round-trip), or concurrently with a custom context_builder if one is set (see Custom context building);
4. prepends the resulting context block, wrapped in context_template, to the message history as a system message.

A subtle but important detail: ProcessHistory fires once per model request, not once per run. A single agent.run that makes a tool call invokes the processor more than once with the same user turn. The processor therefore dedupes per run: it persists and retrieves on the first model request of a run (keyed by Pydantic AI's RunContext.run_id), caches the result, and replays it on re-invocations within the same run without touching Zep again. Two separate runs that send identical text each persist -- dedupe is scoped to the run, not the text.

Custom context building

Set context_builder on ZepDeps to replace the default context retrieval with custom logic -- for example, searching a different graph, applying filters, or combining multiple sources:

from zep_pydantic_ai import ContextInput, ZepDeps

async def my_builder(ctx: ContextInput) -> str | None:
    results = await ctx.zep.graph.search(
        user_id=ctx.user_id,
        query=ctx.user_message,
        scope="edges",
    )
    if not results.edges:
        return None
    return "\n".join(edge.fact for edge in results.edges)

deps = ZepDeps(client=zep, user_id="u", thread_id="t", context_builder=my_builder)

ContextInput bundles zep (the AsyncZep client), user_id, thread_id, user_message, and run_context (the Pydantic AI RunContext for the turn).

When context_builder is set, message persistence (add_messages without return_context) and the builder run concurrently, with per-side failure isolation:

• If the builder raises, a warning is logged and context injection is skipped for that turn -- but persistence still completes.
• If persistence raises, a warning is logged and the turn is not marked as persisted (so it retries on the next model request) -- but a successful builder result is still injected.

context_template

Controls how retrieved context is wrapped before injection. Must contain a literal {context} placeholder, rendered via plain string replacement (template.replace("{context}", context), never str.format) -- so context text containing {, }, or % is always safe to inject:

deps = ZepDeps(
    client=zep,
    user_id="u",
    thread_id="t",
    context_template="Relevant memory:\n{context}",
)

Defaults to DEFAULT_CONTEXT_TEMPLATE, an explicit <ZEP_CONTEXT>...</ZEP_CONTEXT> block -- the same canonical wording used across zep-adk's Python, Go, and TypeScript implementations.

Provisioning

ensure_user and ensure_thread (in zep_pydantic_ai.provisioning) explicitly provision the Zep user and thread out-of-band, before the first turn -- useful for onboarding flows that want genuine failures (auth, network, 5xx) to raise loudly rather than degrade silently:

from zep_pydantic_ai import ensure_thread, ensure_user

async def setup_user(zep_client, user_id: str) -> None:
    ...  # e.g. configure per-user ontology

created = await ensure_user(
    zep,
    user_id="user_123",
    first_name="Jane",
    last_name="Smith",
    email="jane@example.com",
    on_created=setup_user,  # fires exactly once, only on real creation
)
await ensure_thread(zep, thread_id="thread_abc", user_id="user_123")

Both are create-then-catch-conflict: they call the Zep SDK's create method directly and treat an "already exists" conflict as success (returning False), while genuine failures propagate. If on_created raises, that exception also propagates even though the user was created -- make the hook idempotent so it can be safely re-run.

You do not have to call these explicitly: zep_history_processor calls the same logic lazily on the turn path, but wrapped so that a genuine failure there is logged and degrades to no-memory rather than breaking the run.

create_zep_search_tool

A factory that returns a model-callable pydantic_ai.Tool over graph.search. The model decides when to search the knowledge graph for specific facts, entities, or prior episodes. By default it searches the current user's graph; pass graph_id=... to target a shared standalone graph (e.g. a documentation knowledge base).

Pin-or-expose. Every search parameter (scope, reranker, limit, mmr_lambda, center_node_uuid) is exposed to the model in the tool's JSON schema by default, with documented defaults. Use pinned_params to fix a parameter to a constant value and hide it from the schema; use hidden_params to hide a parameter without pinning it, so Zep's own server-side default applies:

# Model chooses scope/reranker/limit/mmr_lambda/center_node_uuid freely.
tool = create_zep_search_tool()

# Pin scope to "nodes" and limit to 5 -- hidden from the model, always sent.
tool = create_zep_search_tool(pinned_params={"scope": "nodes", "limit": 5})

# Hide mmr_lambda from the schema; Zep applies its own default when omitted.
tool = create_zep_search_tool(hidden_params={"mmr_lambda"})

search_filters and bfs_origin_node_uuids are always constructor-only (their complex shapes are not exposed to the model).

persist_run

Call after agent.run with result.new_messages() to persist the assistant's reply to the Zep thread. Only assistant text is sent -- the user turn (already persisted by the processor) and any tool-call/tool-return scaffolding are skipped, so Zep sees one clean assistant message per turn. Not needed if you use zep_capabilities(deps) (see below).

zep_capabilities / automatic assistant persistence

zep_capabilities(deps) bundles ProcessHistory(zep_history_processor) with an after_run hook (via Pydantic AI's Hooks capability) that automatically persists the assistant's reply once the run completes -- no explicit persist_run call needed:

from zep_pydantic_ai import zep_capabilities

agent = Agent(
    "openai:gpt-5-mini",
    deps_type=ZepDeps,
    capabilities=zep_capabilities(deps),
)

result = await agent.run("Hi", deps=deps)
# Both sides of the turn are already in Zep.

Use create_zep_after_run_hook(deps) directly if you want to compose it with your own Hooks(...) instance instead of using the bundled list.

Public API

ZepDeps

ensure_user

Returns True if newly created, False if it already existed. Genuine failures and on_created errors propagate.

ensure_thread

Returns True if newly created, False if it already existed. Genuine failures propagate.

create_zep_search_tool

Model-exposed search parameters (when not pinned/hidden), with their defaults:

Returns a pydantic_ai.Tool[ZepDeps] -- pass it directly in tools=[...].

Features

• Native ProcessHistory capability -- the current Pydantic AI hook, not the deprecated history_processors= kwarg
• Single round-trip -- persist + retrieve context in one add_messages call (or concurrently, with a custom context_builder)
• Once-per-run dedupe -- correct under tool-calling runs that re-invoke the processor
• Out-of-band provisioning -- ensure_user/ensure_thread for onboarding flows that want failures to raise loudly, plus lazy fallback on the turn path
• Pin-or-expose search tool -- every search parameter model-exposed by default, or pinned/hidden per deployment
• Automatic assistant persistence -- zep_capabilities(deps) wires up Hooks(after_run=...), or persist explicitly with persist_run
• Graceful error handling -- Zep failures are logged but never crash the agent run
• Fully typed -- ships type hints; passes mypy --strict-style checks

Error Handling

Every Zep call on the turn path (history processor, search tool, and the after_run hook) is wrapped: a Zep outage, auth failure, or transient error is logged and the agent run continues without memory for that turn. When persistence fails the turn is not cached, so the next model request retries it. ensure_user/ensure_thread, called directly (out-of-band), are the exception: they raise genuine failures so misconfiguration is caught before the agent ever runs.

Configuration

export ZEP_API_KEY="your-zep-api-key"
export OPENAI_API_KEY="your-openai-api-key"   # or another provider supported by Pydantic AI

Examples

See the examples/ directory:

• basic_agent.py -- fact seeding and memory recall with the history processor + search tool.

Development

make install      # uv sync --extra dev
make format       # ruff format
make lint         # ruff check
make type-check   # mypy src/
make test         # pytest
make all          # format + lint + type-check + test
make build        # uv build

Requirements

• Python 3.11+
• pydantic-ai>=1.107,<2
• zep-cloud>=3.23.0

Support

• Zep Documentation
• Pydantic AI Documentation
• GitHub Issues

License

Apache 2.0 - see LICENSE for details.

Contributing

Contributions are welcome! Please see our Contributing Guide for details.