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 /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 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 xmemory_instance

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

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


@tool
def remember(text: str) -> str:
    """Store information in long-term memory."""
    result = api.write(text)
    return f"Stored {len(result.cleaned_objects['objects'])} object(s)."


@tool
def recall(query: str) -> str:
    """Retrieve information from long-term memory."""
    result = api.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 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)

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

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 asyncio
import os
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

MCP_TOKEN = os.environ["XMEM_AUTH_TOKEN"]

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)

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