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 /instance/generate_schema — 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 xmemory_instance, 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. Convert the Pydantic schema to an xmemory schema
api = xmemory_instance(token=AUTH_TOKEN)  # also reads XMEM_AUTH_TOKEN from env if not passed
schema_response = api.generate_schema(
    schema_description=f"for following json_schema: {Contact.model_json_schema()}"
)

# 3. Create the instance
create_response = api.create_instance(
    schema_text=yaml.dump(schema_response.generated_schema, allow_unicode=True),
    schema_type=SchemaType.YML,
)
instance_id = create_response.instance_id
print(f"Created instance: {instance_id}")
# → store this in INSTANCE_ID and reuse it on subsequent runs

The instance_id is a string like "contacts-a3f8". Keep it — you’ll pass it with every write and read call.

---

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 XmemoryAPI, xmemory_instance

AUTH_TOKEN = os.environ["XMEM_AUTH_TOKEN"]
INSTANCE_ID = "contacts-a3f8"  # from Part 1


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

    api: XmemoryAPI


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."""
    result = ctx.deps.api.write(text)
    return f"Stored {len(result.cleaned_objects['objects'])} object(s)."


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


# Run it
deps = Deps(api=xmemory_instance(instance_id=INSTANCE_ID, token=AUTH_TOKEN))  # also reads XMEM_AUTH_TOKEN from env if not passed

# 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 endpoint supports three modes via the mode field:

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

Get your API key from the xmemory dashboard and pass it as a bearer token. The token encodes which instance the session is bound to, so you don’t pass instance_id explicitly in tool calls.

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

MCP_TOKEN = os.environ["XMEM_AUTH_TOKEN"]

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)

Tool	Arguments	Description
write	text: str	Extract entities from text and persist them
read	query: str	Natural-language query → answer
get_instance_id	—	Returns the bound instance ID
get_instance_schema	—	Returns the YAML schema as JSON