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. Get one from the xmemory console.

---

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. You normally describe what to remember in plain language and xmemory generates the schema for you — but since you’re already working with Pydantic, you can feed your model’s JSON schema straight to the 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

API_KEY = os.environ["XMEM_API_KEY"]


# 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(api_key=API_KEY)
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

API_KEY = os.environ["XMEM_API_KEY"]
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-sonnet-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(api_key=API_KEY))

# 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.

Authentication

This headless flow sends your account API key directly as the Bearer token on the /instance/<instance_id> shortcut path — no OAuth exchange needed. The instance is bound by the URL path, so you don’t pass instance_id explicitly in tool calls. See the MCP guide for the other connection options.

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

API_KEY = os.environ["XMEM_API_KEY"]
INSTANCE_ID = os.environ["XMEM_INSTANCE_ID"]  # from Part 1

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

agent = Agent(
    "anthropic:claude-sonnet-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 9 default tools — 6 bound (get_instance_id, get_instance_schema, write, write_async, write_status, read) and the 3-tool schema-evolution suggestion engine (review_suggestions, decide_suggestions, apply_pending_decisions) — plus an opt-in Schema management group (5 more). See the MCP — Tools reference for full parameter and return-shape details.