AMP Python Client SDK

amp-client is the official Python SDK client for the Agent Memory Protocol (AMP). It provides standard synchronous and asynchronous clients to interact with an AMP memory server, as well as native integrations with LangChain.

Installation

Not yet on PyPI, so install from a clone of the repo:

pip install -e sdk/

To include LangChain support:

pip install -e "sdk/[langchain]"

Quickstart

Store and retrieve memories with the synchronous client in under 5 lines:

from amp_client import AMPClient

client = AMPClient("http://localhost:8765", agent_id="agent_assistant")

# Only when the server runs with AMP_API_KEYS_FILE; without it the agent id is
# accepted on its own, which is the binding the spec describes.
client = AMPClient("http://localhost:8765", agent_id="agent_assistant", api_key="...")
client.remember(content="User prefers email correspondence.", owner_id="user_123")
memories = client.recall(query="communication preferences", owner_id="user_123")
print(memories[0]["content"]["text"])

Reading one cell

cell = client.get_memory("mem_01J5A3B7K9M2N4P6Q8R0S1T3V5")

Reading is what resets a cell's decay clock server-side: the response carries the bumped scoring.access_count and lifecycle.last_accessed_at.

Paging

recall and list_memories take an offset, which skips that many results the agent may read - so page 2 is page 2 of what it can see, not of what the store holds:

page = client.list_memories(owner_id="user_123", limit=20, offset=20)

A page shorter than limit is the last one. The HTTP response also carries has_more; these methods return the cells, so use client.session (or client._client on the async client) if you need it.

Full API Reference

AMPClient (Sync)

__init__(server_url: str, agent_id: str)

Initializes the client. Auto-normalizes the server_url to append /amp/v1.

• server_url: Base URL of the AMP server.
• agent_id: Identifier of the client agent (maps to X-AMP-Agent-ID header).

remember(content: str | dict, owner_id: str, type: str = "semantic", importance: float = 0.5, readable_by: list[str] | None = None) -> dict

Store a piece of information.

• content: Memory content (either a string, or a dict with text and metadata).
• owner_id: ID of the owner entity (e.g. user ID).
• type: Memory type ("semantic", "episodic", or "procedural").
• importance: Float score in range [0.0, 1.0].
• readable_by: Optional list of agent ID patterns permitted to read this cell.
• Returns: Dictionary representation of the created MemoryCell.

recall(query: str, owner_id: str, limit: int = 5, include_stale: bool = False) -> list[dict]

Perform semantic search for memories.

• query: Natural language query.
• owner_id: Owner ID of target memories.
• limit: Maximum results to return.
• include_stale: If True, includes stale cells in search.
• Returns: List of matching MemoryCell dictionaries.

forget(memory_id: str) -> bool

Archive and delete a memory.

• memory_id: The ID of the memory cell.
• Returns: True if successfully deleted.

list_memories(owner_id: str, type: str | None = None, limit: int = 20) -> list[dict]

Filter active memories by structured criteria (no semantic search).

• owner_id: Owner ID.
• type: Optional memory type filter.
• limit: Maximum results to return.
• Returns: List of active MemoryCell dictionaries.

health() -> bool

Check server health status.

• Returns: True if healthy.

Async Client (AsyncAMPClient)

The asynchronous client has the exact same API surface as AMPClient, but all network calls must be awaited. It also implements async context manager support:

import asyncio
from amp_client import AsyncAMPClient

async def main():
    async with AsyncAMPClient("http://localhost:8765", "agent_assistant") as client:
        await client.remember("User likes dark mode.", "user_123")
        results = await client.recall("UI preferences", "user_123")
        print(results)

asyncio.run(main())

LangChain Integration

AMPMemory is a LangChain BaseChatMessageHistory backed by AMP, so a chain can persist conversation turns as memory cells and read them back. The example below is exactly what the test suite runs and what a live server was verified with; note that langchain-core 1.x removed BaseMemory and ConversationChain, so this uses the current chat-history interface.

from amp_client import AMPClient
from amp_client.integrations.langchain import AMPMemory
from langchain_core.messages import HumanMessage, AIMessage

client = AMPClient("http://localhost:8765", agent_id="chatbot_agent")
memory = AMPMemory(client=client, owner_id="user_john", memory_key="history")

memory.add_message(HumanMessage(content="Hi, my name is John and I write Python."))
memory.add_message(AIMessage(content="Nice to meet you, John."))

for message in memory.messages:
    print(message.type, message.content)

Any LangChain component that accepts a chat message history works with it directly. If you want the older string form instead of message objects, use return_messages=False and read memory.load_memory_variables({})["history"].

Error Handling

All client errors resulting from non-2xx responses or connection issues raise AMPError.

from amp_client import AMPClient, AMPError

client = AMPClient("http://localhost:8765", agent_id="agent_test")
try:
    client.remember(content="", owner_id="user_123")  # Invalid schema
except AMPError as e:
    print(f"Error occurred: {e} (Status: {e.status_code})")