LangChain

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

• MCP — connect LangChain to the xmemory MCP server via langchain-mcp-adapters; the agent gets write and read tools automatically.
• HTTP API — call the REST API directly from custom LangChain tools; useful when you want full control.

API key: To use xmemory APIs or integrations, you need an API key. Please register your interest at https://xmemory.ai and we will reach out to give access. Copy and securely store the key. Never share your API key publicly.

---

Prerequisites

pip install langchain langchain-anthropic xmemory-ai pyyaml

For the MCP approach (Part 3), also install:

pip install langchain-mcp-adapters

---

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 LangChain tools so the agent can store and recall information during a conversation.

import asyncio
import os
from langchain.agents import create_agent
from langchain_core.tools import tool
from xmemory import XmemoryClient

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

client = XmemoryClient(token=AUTH_TOKEN)
inst = client.instance(INSTANCE_ID)


@tool
def remember(text: str) -> str:
    """Store information in long-term memory."""
    result = inst.write(text)
    return f"Stored (write_id={result.write_id})."


@tool
def recall(query: str) -> str:
    """Retrieve information from long-term memory."""
    result = inst.read(query)
    return result.reader_result.get("answer", str(result.reader_result))


agent = create_agent(
    "anthropic:claude-opus-4-6",
    tools=[remember, recall],
    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."
    ),
)


async def main():
    # Store something
    result = await agent.ainvoke(
        {"messages": "Remember that Alice Johnson works at Acme Corp, her email is alice@acme.com."}
    )
    print(result["messages"][-1].content)

    # Recall it later
    result = await agent.ainvoke(
        {"messages": "What do you know about Alice?"}
    )
    print(result["messages"][-1].content)


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

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)

LangChain supports MCP servers via the langchain-mcp-adapters package. 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 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 asyncio
import os
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

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

client = MultiServerMCPClient(
    {
        "xmemory": {
            "transport": "http",
            "url": "https://mcp.xmemory.ai/",
            "headers": {
                "Authorization": f"Bearer {MCP_TOKEN}",
            },
        },
    }
)


async def main():
    tools = await client.get_tools()
    agent = create_agent(
        "anthropic:claude-opus-4-6",
        tools=tools,
        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."
        ),
    )

    result = await agent.ainvoke(
        {"messages": "Remember that Bob Smith is a senior engineer at Globex. "
         "Then tell me what you know about Bob."}
    )
    print(result["messages"][-1].content)


if __name__ == "__main__":
    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