Pydantic AI

This guide shows how to use xmemory as a persistent memory layer for a pydantic-ai agent. Two approaches are covered:

• MCP — point pydantic-ai at the xmemory MCP server; the agent gets write and read tools automatically.
• HTTP API — call the REST API directly from custom pydantic-ai tools; useful when you want full control.

---

Prerequisites

pip install "pydantic-ai>=0.0.14" xmemory-ai pyyaml

You need an xmemory API key. Please register your interest at https://xmemory.ai and we will reach out to give access.

---

Part 1 — Create an instance from a Pydantic schema

xmemory stores data in typed instances. Each instance has a schema that describes the objects and relations you want to track. Pass your Pydantic model’s JSON schema to the schema generation endpoint — xmemory converts it into its own typed schema and returns it ready for instance creation.

This is a one-time setup script. Save the returned instance ID (e.g. in an environment variable) and reuse it in your agent.

import os
import yaml
from pydantic import BaseModel
from xmemory import XmemoryClient, SchemaType

AUTH_TOKEN = os.environ["XMEM_AUTH_TOKEN"]


# 1. Define your domain model with Pydantic
class Contact(BaseModel):
    name: str
    email: str | None = None
    company: str | None = None
    notes: str | None = None


# 2. Connect and pick a cluster
client = XmemoryClient(token=AUTH_TOKEN)
clusters = client.admin.list_clusters()
cluster_id = clusters[0].id

# 3. Convert the Pydantic schema to an xmemory schema
schema_response = client.admin.generate_schema(
    cluster_id,
    schema_description=f"for following json_schema: {Contact.model_json_schema()}",
)

# 4. Create the instance
inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="contacts",
    schema_text=yaml.dump(schema_response.data_schema, allow_unicode=True),
    schema_type=SchemaType.YML,
)
print(f"Created instance: {inst.id}")
# → store this in INSTANCE_ID and reuse it on subsequent runs

The instance ID is a UUID string. Keep it — you’ll use it to get an instance handle on subsequent runs.

---

Part 2 — Agent with custom xmemory tools (HTTP API)

Wrap write and read as pydantic-ai tools so the agent can store and recall information during a conversation.

import os
from pydantic import BaseModel, ConfigDict
from pydantic_ai import Agent, RunContext
from xmemory import XmemoryClient

AUTH_TOKEN = os.environ["XMEM_AUTH_TOKEN"]
INSTANCE_ID = os.environ["XMEM_INSTANCE_ID"]  # from Part 1


class Deps(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    client: XmemoryClient


agent = Agent(
    "anthropic:claude-opus-4-6",
    deps_type=Deps,
    system_prompt=(
        "You are a helpful assistant with access to a persistent memory store. "
        "Use `remember` to save new information and `recall` to look things up."
    ),
)


@agent.tool
def remember(ctx: RunContext[Deps], text: str) -> str:
    """Store information in long-term memory."""
    inst = ctx.deps.client.instance(INSTANCE_ID)
    result = inst.write(text)
    return f"Stored (write_id={result.write_id})."


@agent.tool
def recall(ctx: RunContext[Deps], query: str) -> str:
    """Retrieve information from long-term memory."""
    inst = ctx.deps.client.instance(INSTANCE_ID)
    result = inst.read(query)
    return result.reader_result.get("answer", str(result.reader_result))


# Run it
deps = Deps(client=XmemoryClient(token=AUTH_TOKEN))

# Store something
result = agent.run_sync(
    "Remember that Alice Johnson works at Acme Corp, her email is alice@acme.com.",
    deps=deps,
)
print(result.output)

# Recall it later
result = agent.run_sync(
    "What do you know about Alice?",
    deps=deps,
)
print(result.output)

Read modes

The read method supports three modes via read_mode:

read_mode	reader_result shape	When to use
"single-answer"	{"answer": "..."}	Natural-language question → plain text answer
"xresponse"	{"objects": [...], "relations": [...]}	Get structured objects back
"raw-tables"	{"tables": [...]}	Raw SQL result sets

---

Part 3 — MCP approach (fewer lines of code)

pydantic-ai supports MCP servers natively. The xmemory MCP server exposes write and read (and more) as ready-made tools — no boilerplate needed.

Getting an MCP token

The xmemory MCP server uses OAuth 2.0 with PKCE — your API key is not a valid MCP bearer token. It must first be exchanged for an opaque MCP access token through the OAuth2 login flow at https://oauth2.xmemory.ai. MCP-aware clients perform this handshake automatically and cache the resulting token; the snippet below assumes you have already obtained one (e.g. via your client’s MCP setup wizard) and stored it in XMEM_MCP_TOKEN. The token encodes which instance the session is bound to, so you don’t pass instance_id explicitly in tool calls. See the MCP guide for the full flow.

import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

MCP_TOKEN = os.environ["XMEM_MCP_TOKEN"]  # opaque MCP access token, not the raw API key

mcp_server = MCPServerStreamableHTTP(
    url="https://mcp.xmemory.ai/",
    headers={"Authorization": f"Bearer {MCP_TOKEN}"},
)

agent = Agent(
    "anthropic:claude-opus-4-6",
    mcp_servers=[mcp_server],
    system_prompt=(
        "You have access to a persistent memory store via the xmemory tools. "
        "Use `write` to remember things and `read` to look them up."
    ),
)


async def main():
    async with agent.run_mcp_servers():
        result = await agent.run(
            "Remember that Bob Smith is a senior engineer at Globex. "
            "Then tell me what you know about Bob."
        )
        print(result.output)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Available MCP tools (instance connection type)

The instance connection exposes 11 tools — 6 bound (get_instance_id, get_instance_schema, write, write_async, write_status, read) and 5 explicit-instance (extract, write_to, write_to_async, write_to_status, read_from). See the MCP — Tools reference for full parameter and return-shape details.

Tool	Arguments	Description
write	text: str	Extract entities from text and persist them (sync)
write_async	text: str	Same as write but returns immediately with a write_id
write_status	write_id: str	Poll the status of an async write
read	query: str	Natural-language query → answer
get_instance_id	—	Returns the bound instance ID
get_instance_schema	—	Returns the schema as JSON