Google ADK

xmemory Integration Guide for Google ADK

This guide explains how to integrate xmemory-ai into agents built with the Google Agent Development Kit (ADK).

xmemory gives your agent persistent, structured memory: write free-form text, have it extracted into a typed schema, and query it back in natural language at any time.

---

Installation

pip install google-adk xmemory-ai pyyaml

---

Authentication & Configuration

The client reads credentials from environment variables by default.

Variable	Description
XMEM_API_KEY	API key for the xmemory API
XMEM_API_URL	API base URL (defaults to https://api.xmemory.ai)

Set them before running your agent:

export XMEM_API_KEY="your-api-key"

Or pass them explicitly when constructing the client (see below).

---

Core Concepts

Concept	Description
Cluster	A group of instances. List clusters with client.admin.list_clusters().
Instance	A named memory store with a typed schema inside a cluster. Identified by a UUID string.
Schema	A YAML or JSON definition of the data structure xmemory extracts and stores. Generate schemas using client.admin.generate_schema().
Write	Send free-form text; xmemory extracts structured objects and merges them into the instance.
Read	Query the instance in natural language; xmemory answers from stored structured data.

---

Quick Start

import yaml
from xmemory import XmemoryClient, SchemaType

client = XmemoryClient(api_key="your-api-key")

# Step 1: pick a cluster
clusters = client.admin.list_clusters()
cluster_id = clusters[0].id

# Step 2: generate a schema from a plain-language description
schema_resp = client.admin.generate_schema(
    cluster_id,
    "Track contacts with their name, email address, and notes.",
)

# Step 3: create a memory instance using the generated schema
inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="contacts",
    schema_text=yaml.dump(schema_resp.data_schema),
    schema_type=SchemaType.YML,
)
# inst is a bound handle — save inst.id for subsequent runs

# Step 4: write information into memory
inst.write("Alice Johnson joined the team today. Her email is alice@example.com.")

# Step 5: read it back
result = inst.read("What is Alice's email address?")
print(result.reader_result)

---

Schemas

Use client.admin.generate_schema() to create or update schemas. The generation endpoint understands xmemory’s schema format and naming conventions, and produces schemas that extract data correctly.

# Generate a schema from a plain-language description
resp = client.admin.generate_schema(
    cluster_id,
    "Track user preferences, past conversations, and open tasks with priorities and due dates.",
)
schema_yml = resp.data_schema  # pass this to create_instance

Object names in schemas use CamelCase (e.g. UserPreferences, OpenTask, ConversationEntry). The generation endpoint handles this automatically.

Updating a schema

When your memory needs change, describe the updated structure and regenerate:

resp = client.admin.generate_schema(
    cluster_id,
    "Track user preferences, past conversations, and open tasks with priorities, due dates, and assignees.",
    current_yml_schema=current_schema_yml,  # pass the existing schema so changes are incremental
)
client.admin.update_instance_schema(instance_id, yaml.dump(resp.data_schema), SchemaType.YML)

Passing current_yml_schema preserves existing objects and fields — only the described changes are applied.

---

Defining Tools for Google ADK

The recommended pattern is to wrap xmemory operations as ADK FunctionTool instances and register them on your agent. The agent decides when to call them.

import os
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from xmemory import XmemoryClient, ReadMode, XmemoryAPIError

INSTANCE_ID = os.environ["XMEM_INSTANCE_ID"]

client = XmemoryClient(api_key=os.environ["XMEM_API_KEY"])
inst = client.instance(INSTANCE_ID)


def remember(text: str) -> str:
    """
    Store information in long-term memory.
    Call this whenever the user shares facts, preferences, or context worth remembering.

    Args:
        text: Free-form text describing what to remember.

    Returns:
        Confirmation string.
    """
    try:
        resp = inst.write(text)
        return f"Stored in memory (write_id={resp.write_id})."
    except XmemoryAPIError as e:
        return f"Memory write failed: {e}"


def recall(query: str) -> str:
    """
    Retrieve information from long-term memory.
    Call this when you need to recall facts, preferences, or context stored in previous sessions.

    Args:
        query: Natural language question about stored information.

    Returns:
        Answer derived from stored memory.
    """
    try:
        resp = inst.read(query, read_mode=ReadMode.SINGLE_ANSWER)
        return str(resp.reader_result) if resp.reader_result is not None else "Nothing found."
    except XmemoryAPIError as e:
        return f"Memory read failed: {e}"


agent = Agent(
    name="my_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant with long-term memory. Use your memory tools to remember and recall information.",
    tools=[
        FunctionTool(remember),
        FunctionTool(recall),
    ],
)

---

Async Writes

For latency-sensitive agents, use write_async to enqueue a write and continue without blocking, then poll for completion with write_status.

from xmemory import WriteQueueStatus, XmemoryAPIError


def remember_async(text: str) -> str:
    """
    Enqueue a memory write without blocking. Returns a write_id to check later.

    Args:
        text: Free-form text to store.

    Returns:
        The write_id assigned to this operation.
    """
    try:
        resp = inst.write_async(text)
        return f"Write enqueued. write_id={resp.write_id}"
    except XmemoryAPIError as e:
        return f"Memory write failed: {e}"


def check_write_status(write_id: str) -> str:
    """
    Check the status of a previously enqueued memory write.

    Args:
        write_id: The write_id returned by remember_async.

    Returns:
        Current status: queued, processing, completed, failed, or not_found.
    """
    try:
        resp = inst.write_status(write_id)
        status = resp.write_status
        if status == WriteQueueStatus.FAILED:
            return f"Write failed: {resp.error_detail}"
        if status == WriteQueueStatus.COMPLETED:
            return f"Write completed at {resp.completed_at}."
        return f"Write status: {status}."
    except XmemoryAPIError as e:
        return f"Status check failed: {e}"

---

Managing Instances

Create a new instance

Generate the schema first, then create the instance:

import yaml
from xmemory import XmemoryClient, SchemaType

client = XmemoryClient(api_key="your-api-key")
clusters = client.admin.list_clusters()
cluster_id = clusters[0].id

schema_resp = client.admin.generate_schema(
    cluster_id,
    "Track user preferences and open tasks with priorities and due dates.",
)

inst = client.admin.create_instance(
    cluster_id=cluster_id,
    name="my-memory",
    schema_text=yaml.dump(schema_resp.data_schema),
    schema_type=SchemaType.YML,
)
print(inst.id)  # persist this value

Reuse an existing instance

client = XmemoryClient(api_key="your-api-key")
inst = client.instance("your-saved-instance-id")

Update the schema of an existing instance

Generate the updated schema incrementally, then apply it:

schema_resp = client.admin.generate_schema(
    cluster_id,
    "Track user preferences and open tasks with priorities, due dates, and assignees.",
    current_yml_schema=current_schema_yml,
)

client.admin.update_instance_schema(
    instance_id, yaml.dump(schema_resp.data_schema), SchemaType.YML
)

---

Read Modes

The read method supports three modes via ReadMode:

Mode	Description
ReadMode.SINGLE_ANSWER	Natural language answer to the query (default). Best for most agent use cases.
ReadMode.RAW_TABLES	Returns raw structured data tables without summarisation.
ReadMode.XRESPONSE	Full structured response envelope with metadata.

from xmemory import ReadMode

# Natural language answer
resp = inst.read("What tasks are overdue?", read_mode=ReadMode.SINGLE_ANSWER)

# Raw structured data
resp = inst.read("List all tasks", read_mode=ReadMode.RAW_TABLES)

---

Extraction Logic

The write and write_async methods accept an extraction_logic parameter — pass ExtractionLogic.DEEP to override the FAST default:

Value	Description
ExtractionLogic.FAST	Fastest; lower accuracy (default). Use for high-volume, low-stakes writes.
ExtractionLogic.DEEP	Most thorough extraction. Best for important or complex information.

---

Error Handling

All xmemory operations raise XmemoryAPIError on failure. The exception carries an optional .status attribute with the HTTP status code.

from xmemory import XmemoryAPIError, XmemoryHealthCheckError

try:
    client.check_health()
except XmemoryHealthCheckError as e:
    print(f"API unreachable: {e}")

try:
    resp = inst.write("...")
except XmemoryAPIError as e:
    print(f"Error {e.status}: {e}")

---

Complete Agent Example

import os
import yaml
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from xmemory import (
    XmemoryClient,
    ReadMode,
    WriteQueueStatus,
    XmemoryAPIError,
)

client = XmemoryClient(api_key=os.environ["XMEM_API_KEY"])
inst = client.instance(os.environ["XMEM_INSTANCE_ID"])


def remember(text: str) -> str:
    """Store information in long-term memory. Call whenever the user shares facts worth preserving."""
    try:
        resp = inst.write(text)
        return "Remembered."
    except XmemoryAPIError as e:
        return f"Could not store memory: {e}"


def recall(query: str) -> str:
    """Recall information from long-term memory. Call when you need context from past sessions."""
    try:
        resp = inst.read(query, read_mode=ReadMode.SINGLE_ANSWER)
        return str(resp.reader_result) if resp.reader_result is not None else "Nothing found."
    except XmemoryAPIError as e:
        return f"Could not recall memory: {e}"


def remember_in_background(text: str) -> str:
    """Enqueue a memory write without waiting. Use when you don't need to confirm storage immediately."""
    try:
        resp = inst.write_async(text)
        return f"Enqueued. write_id={resp.write_id}"
    except XmemoryAPIError as e:
        return f"Could not enqueue: {e}"


def check_memory_write(write_id: str) -> str:
    """Check if a background memory write has completed."""
    try:
        resp = inst.write_status(write_id)
        if resp.write_status == WriteQueueStatus.COMPLETED:
            return "Memory saved successfully."
        if resp.write_status == WriteQueueStatus.FAILED:
            return f"Memory write failed: {resp.error_detail}"
        return f"Status: {resp.write_status}."
    except XmemoryAPIError as e:
        return f"Status check failed: {e}"


agent = Agent(
    name="persistent_memory_agent",
    model="gemini-2.5-flash",
    instruction=(
        "You are a helpful assistant with persistent long-term memory across sessions. "
        "Use `remember` to store important facts the user shares. "
        "Use `recall` before answering questions that may depend on past context. "
        "Use `remember_in_background` for non-critical information to avoid latency."
    ),
    tools=[
        FunctionTool(remember),
        FunctionTool(recall),
        FunctionTool(remember_in_background),
        FunctionTool(check_memory_write),
    ],
)

---

Reference

The Google ADK integration uses the Python SDK, so every client, admin, and instance method — with full parameters, return types, and errors — is documented in the Python SDK reference:

• Client methods
• Admin methods
• Instance methods