OpenAI Agents SDK

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

• MCP — connect the Agents SDK to the xmemory MCP server; the agent gets write and read tools automatically.
• HTTP API — call the REST API directly from custom function tools; useful when you want full control.

API key: To use xmemory APIs or integrations, you need an API key. Get one from the xmemory console. Copy and securely store the key. Never share your API key publicly.

---

Prerequisites

pip install openai-agents xmemory-ai pyyaml

Set XMEM_API_KEY (your xmemory key — create one in the console) in your environment; every example below reads it. The Agents SDK uses OpenAI models, so set OPENAI_API_KEY too.

---

Part 1 — Create an instance

xmemory stores data in typed instances. Describe what you want to remember in plain language and xmemory designs the schema for you — natural-language generation is the primary path (you can also hand it an existing schema, e.g. a Pydantic model’s JSON schema).

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

Fastest path: the CLI creates an instance in three commands (xmemcli xmd generate → validate → instance create). The Python script below is the equivalent if you’d rather stay in code.

import os
import yaml
from xmemory import XmemoryClient, SchemaType

API_KEY = os.environ["XMEM_API_KEY"]

# Connect and pick a cluster
client = XmemoryClient(api_key=API_KEY)
cluster_id = client.admin.list_clusters()[0].id

# Describe what to remember in plain language — xmemory designs the schema
schema = client.admin.generate_schema(
    cluster_id,
    "Track contacts with name, email, company, and notes.",
)

# Create the instance from that schema
inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="contacts",
    schema_text=yaml.dump(schema.data_schema, allow_unicode=True),
    schema_type=SchemaType.YML,
)
print(f"Created instance: {inst.id}")
# → store this in XMEM_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 Agents SDK function tools so the agent can store and recall information during a conversation.

import asyncio
import os
from agents import Agent, Runner, function_tool
from xmemory import XmemoryClient

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

client = XmemoryClient(api_key=API_KEY)
inst = client.instance(INSTANCE_ID)


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


@function_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 = Agent(
    name="Assistant",
    instructions=(
        "You are a helpful assistant with access to a persistent memory store. "
        "Use `remember` to save new information and `recall` to look things up."
    ),
    tools=[remember, recall],
)


async def main():
    result = await Runner.run(
        agent,
        "Remember that Alice Johnson works at Acme Corp, her email is alice@acme.com.",
    )
    print(result.final_output)


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

remember uses write_async, which returns immediately — recall it in a later Runner.run (a separate turn); avoid reading it back in the same turn as the async write.

Read modes

The read method supports three modes via read_mode:

read_mode	reader_result shape	When to use
single-answer	{"answer": …}	Plain-text answer
xresponse	{"objects": …, "relations": …}	Structured objects back
raw-tables	{"tables": …}	Raw SQL result sets

---

Part 3 — MCP approach (fewer lines of code)

The Agents SDK connects to MCP servers natively via MCPServerStreamableHttp. 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. See the MCP guide for the other connection options.

import asyncio
import os
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

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


async def main():
    async with MCPServerStreamableHttp(
        name="xmemory",
        params={
            "url": f"https://mcp.xmemory.ai/instance/{INSTANCE_ID}",
            "headers": {"Authorization": f"Bearer {API_KEY}"},
        },
        cache_tools_list=True,
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions=(
                "You have access to a persistent memory store via the xmemory tools. "
                "Use `write_async` to remember things and `read` to look them up."
            ),
            mcp_servers=[server],
        )

        result = await Runner.run(
            agent,
            "Remember that Bob Smith is a senior engineer at Globex.",
        )
        print(result.final_output)


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

Hosted alternative

If you would rather have OpenAI connect to the MCP server server-side (through the Responses API) instead of from your own process, use HostedMCPTool and pass the token in headers (the authorization field is for OAuth access tokens, not a static header):

import asyncio
import os
from agents import Agent, Runner, HostedMCPTool

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

agent = Agent(
    name="Assistant",
    instructions="Use the xmemory tools to remember and recall information.",
    tools=[
        HostedMCPTool(
            tool_config={
                "type": "mcp",
                "server_label": "xmemory",
                "server_url": f"https://mcp.xmemory.ai/instance/{INSTANCE_ID}",
                "headers": {"Authorization": f"Bearer {API_KEY}"},
                "require_approval": "never",
            }
        )
    ],
)


async def main():
    result = await Runner.run(agent, "What do you know about Alice?")
    print(result.final_output)


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