# Google ADK

> Google ADK integration example.

Canonical page: https://xmemory.ai/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)](https://google.github.io/adk-docs/).

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

```bash
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:

```bash
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** | The XMD contract xmemory extracts and stores against. Generate it from a plain-language description with `client.admin.generate_schema()`; review the [XMD guide](https://xmemory.ai/xmd/) only when you need explicit control. |
| **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

```python
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 schemas from descriptions when that is more
convenient.** Agents can also author [XMD](https://xmemory.ai/xmd/) directly when they want exact control. The
generation endpoint simply saves that work while following XMD naming conventions and
extraction-controlling descriptions.

```python
# 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:

```python
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.

```python
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`.

```python
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:

```python
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

```python
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:

```python
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. |

```python
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.

```python
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

```python
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](https://xmemory.ai/python/#reference):

- [Client methods](https://xmemory.ai/python/#client)
- [Admin methods](https://xmemory.ai/python/#admin-methods)
- [Instance methods](https://xmemory.ai/python/#instance-methods)