PydanticAI Integration

PydanticAI uses typed Pydantic models for prompts, results, and tools. Wire CortexDB recall + capture as agent tools.

Install

pip install cortexdbai[pydanticai]

LLM provider note. This example uses OpenAI for the agent's chat model, but CortexDB itself is LLM-agnostic. The only model CortexDB invokes internally is the one used by POST /v1/answer and POST /v1/understanding/synthesize (Claude Opus 4.6 by default, configurable). Your agent's chat model is independent — swap OpenAIChat / gpt-4o for Anthropic, Gemini, Mistral, Groq, or any local model. CortexDB does not care.

Agent tools + dependencies

import os
from cortexdb import Cortex
from cortexdb.integrations.pydanticai import CortexDBDeps, register_cortexdb_tools
from pydantic_ai import Agent

client = Cortex(
    api_url="https://api-v1.cortexdb.ai",
    actor="user:alice",
    bearer=os.environ["CORTEX_TOKEN"],
)

agent = Agent("openai:gpt-4o", deps_type=CortexDBDeps)
register_cortexdb_tools(agent)  # adds cortexdb_search / store / forget tools

deps = CortexDBDeps(client=client, scope="org:acme/user:alice")
result = agent.run_sync("What do you remember about the project?", deps=deps)

The tool functions cortexdb_search / cortexdb_store / cortexdb_forget can also be registered individually via agent.tool(...).

Prefer manual control?

import os
from datetime import datetime, timezone
from uuid import uuid4
from pydantic_ai import Agent, RunContext
from cortexdb.v1 import V1Client

client = V1Client(api_url="https://api-v1.cortexdb.ai", actor="user:alice",
                  bearer=os.environ["CORTEX_TOKEN"])
SCOPE = "org:acme/user:alice"

agent = Agent("openai:gpt-4o", system_prompt="Use recall_long_term before answering.")

@agent.tool
async def recall_long_term(ctx: RunContext, query: str) -> str:
    pack = client.recall(scope=SCOPE, view="holistic", query=query,
                         include=["events", "beliefs", "facts", "episodes"],
                         budgets={"max_tokens": 3000})
    return pack.get("context_block", "") or "(no relevant context)"

@agent.tool
async def capture(ctx: RunContext, text: str, role: str = "assistant") -> bool:
    client.experience(scope=SCOPE, text=text, role=role,
                      observed_at=datetime.now(timezone.utc).isoformat(),
                      idempotency_key=f"pai-{role}-{uuid4()}")
    return True


result = await agent.run("What did we decide about the renewal?")

See also

• Python SDK