Memories

POST /v1/memories  →  HTTP 200
{
  "results": [
    { "id": "mem_abc123", "event": "ADD", "memory": "User prefers dark mode" }
  ],
  "collection_id": "col_default",
  "processing_status": "async",
  "created_count": 1,
  "updated_count": 0,
  "skipped_count": 0
}

import requests

BASE = "https://api.hebbrix.com/v1"
H = {"Authorization": "Bearer <your-api-key>"}

# 1) Dry-run first — preview the blast radius, deletes NOTHING
preview = requests.delete(
    f"{BASE}/memories/all",
    headers=H,
    params={"dry_run": "true"},
).json()
# {"deleted_count": 0, "dry_run": True, "would_delete": 1287}
print(f"Would delete {preview['would_delete']} memories")

# 2) Confirmed delete — header must equal "all-memories" if present
result = requests.delete(
    f"{BASE}/memories/all",
    headers={**H, "X-Hebbrix-Confirm-Destructive": "all-memories"},
).json()
print(f"Deleted {result['deleted_count']} memories")

# Same safety pattern for collections (confirm value "all-collections")
requests.delete(
    f"{BASE}/collections/all",
    headers={**H, "X-Hebbrix-Confirm-Destructive": "all-collections"},
)

import asyncio
from hebbrix import MemoryClient

async def main():
    async with MemoryClient(api_key="mem_sk_...") as client:
        collection = await client.collections.create(name="my-agent")

        # Simple memory (SDK calls the smart-extract POST /v1/memories endpoint)
        memory = await client.memories.create(
            collection_id=collection["id"],
            content="User's favorite color is blue",
        )

        # Memory with importance + metadata (tags live inside metadata)
        await client.memories.create(
            collection_id=collection["id"],
            content="User completed Python certification on 2024-01-15",
            importance=0.8,
            metadata={
                "tags": ["education", "python"],
                "certification_provider": "Coursera",
                "completion_date": "2024-01-15",
            },
        )

asyncio.run(main())

import os
import requests

BASE = "https://api.hebbrix.com/v1"
H = {"Authorization": f"Bearer {os.environ['HEBBRIX_API_KEY']}"}

# POST /v1/memories/process — hand it a chat transcript; the backend
# extracts atomic facts, detects conflicts, and emits ADD/UPDATE/NOOP
# events for each resulting memory.
r = requests.post(
    f"{BASE}/memories/process",
    headers=H,
    json={
        "messages": [
            {"role": "user", "content": "Hi! My name is Alex and I work at Acme Corp."},
            {"role": "assistant", "content": "Nice to meet you, Alex!"},
        ]
    },
)
result = r.json()

print(f"Facts extracted: {result['facts_extracted']}")
print(f"Created: {result['memories_created']}")
print(f"Updated: {result['memories_updated']}")

# Later, user corrects their name — the old memory is updated
r = requests.post(
    f"{BASE}/memories/process",
    headers=H,
    json={
        "messages": [
            {"role": "user", "content": "Actually, my name is Jordan, not Alex."},
            {"role": "assistant", "content": "No problem, Jordan!"},
        ]
    },
)
result = r.json()
# result["events"][0]["event"] == "UPDATE"
# result["events"][0]["previous_content"] == "Alex is the user's name"

import os
import requests

BASE = "https://api.hebbrix.com/v1"
H = {"Authorization": f"Bearer {os.environ['HEBBRIX_API_KEY']}"}

# POST /v1/memories/batch — create multiple memories in one request (max 100).
# Note: collection_id is top-level, NOT per-memory.
memories = [
    {"content": "User prefers email over phone calls"},
    {"content": "User works in software development"},
    {"content": "User is based in EST timezone"},
]

r = requests.post(
    f"{BASE}/memories/batch",
    headers=H,
    json={
        "memories": memories,
        "collection_id": "col_default",
    },
)
result = r.json()
print(f"Created {result['created']} memories")

import asyncio
import os
from hebbrix import MemoryClient

async def main():
    async with MemoryClient(api_key=os.environ["HEBBRIX_API_KEY"]) as client:
        # --- Explicit page-at-a-time access ---
        page = await client.memories.list_page(limit=50)
        # page = {"items": [...], "next_cursor": str | None,
        #         "has_more": bool, "total_count": int}
        print(f"Showing {len(page['items'])} of {page['total_count']} memories")

        # Fetch the next page if there is one
        if page["has_more"]:
            next_page = await client.memories.list_page(
                cursor=page["next_cursor"], limit=50
            )
            print(f"Next page has {len(next_page['items'])} memories")

        # Filter by collection
        page = await client.memories.list_page(
            collection_id="col_customer_support", limit=20
        )

        # --- Simple "give me everything" iteration ---
        # iter_all follows next_cursor until has_more == False
        async for memory in client.memories.iter_all(collection_id="col_customer_support"):
            print(memory["id"], memory["content"][:80])

asyncio.run(main())

import os
import requests

BASE = "https://api.hebbrix.com/v1"
H = {"Authorization": f"Bearer {os.environ['HEBBRIX_API_KEY']}"}

# List all memories (cursor-paginated response)
r = requests.get(f"{BASE}/memories", headers=H, params={"limit": 50})
page = r.json()  # {"items": [...], "next_cursor": ..., "has_more": ..., "total_count": ...}

# Paginate through all
cursor = None
while True:
    params = {"limit": 100}
    if cursor:
        params["cursor"] = cursor
    page = requests.get(f"{BASE}/memories", headers=H, params=params).json()
    for memory in page["items"]:
        process(memory)
    if not page.get("has_more") or not page.get("next_cursor"):
        break
    cursor = page["next_cursor"]