memory_operations.py
"""
Memory Operations with AgentOSClient
This example demonstrates how to manage user memories using
AgentOSClient.
Prerequisites:
1. Start an AgentOS server with an agent that has update_memory_on_run=True
2. Run this script: python 03_memory_operations.py
"""
import asyncio
from agno.client import AgentOSClient
# ---------------------------------------------------------------------------
# Create Example
# ---------------------------------------------------------------------------
async def main():
client = AgentOSClient(base_url="http://localhost:7777")
user_id = "example-user"
print("=" * 60)
print("Memory Operations")
print("=" * 60)
# Create a memory
print("\n1. Creating a memory...")
memory = await client.create_memory(
memory="User prefers dark mode for all applications",
user_id=user_id,
topics=["preferences", "ui"],
)
print(f" Created memory: {memory.memory_id}")
print(f" Content: {memory.memory}")
print(f" Topics: {memory.topics}")
# List memories for the user
print("\n2. Listing memories...")
memories = await client.list_memories(user_id=user_id)
print(f" Found {len(memories.data)} memories for user {user_id}")
for mem in memories.data:
print(f" - {mem.memory_id}: {mem.memory[:50]}...")
# Get a specific memory
print(f"\n3. Getting memory {memory.memory_id}...")
retrieved = await client.get_memory(memory.memory_id, user_id=user_id)
print(f" Memory: {retrieved.memory}")
# Update the memory
print("\n4. Updating memory...")
updated = await client.update_memory(
memory_id=memory.memory_id,
memory="User strongly prefers dark mode for all applications and websites",
user_id=user_id,
topics=["preferences", "ui", "accessibility"],
)
print(f" Updated memory: {updated.memory}")
print(f" Updated topics: {updated.topics}")
# Get memory topics
print("\n5. Getting all memory topics...")
topics = await client.get_memory_topics()
print(f" Topics: {topics}")
# Get user memory stats
print("\n6. Getting user memory stats...")
stats = await client.get_user_memory_stats()
print(f" Stats: {len(stats.data)} entries")
# Delete the memory
print(f"\n7. Deleting memory {memory.memory_id}...")
await client.delete_memory(memory.memory_id, user_id=user_id)
print(" Memory deleted")
# Verify deletion
print("\n8. Verifying deletion...")
memories_after = await client.list_memories(user_id=user_id)
print(f" Remaining memories: {len(memories_after.data)}")
# ---------------------------------------------------------------------------
# Run Example
# ---------------------------------------------------------------------------
if __name__ == "__main__":
asyncio.run(main())
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno fastapi
3
Run the example
Save the code above as
memory_operations.py, then run:python memory_operations.py