Python SDK
The agentmemory client for Python 3.9+ — install, configure, and remember, recall, report and forget using only the standard library.
agentmemory is a synchronous client for the /v1 API built on urllib — standard library only, Python 3.9 or later. Current version: 0.4.0.
Install
The package is not published to a public index. Install the wheel your Agent Memory deployment serves, or install from a checkout of the source repository.
Every Agent Memory instance serves its SDK packages under /sdk/:
pip install "$AGENT_MEMORY_BASE_URL/sdk/agentmemory-0.4.0-py3-none-any.whl"Open $AGENT_MEMORY_BASE_URL/sdk/ in a browser to see the available versions.
0.4.0 is a breaking release
search() now returns the whole context (with receiptId) instead of a list, and delete() returns {"status": ...}. The 0.3.0 wheel stays available under /sdk/ for pinned installs.
Configure
import os
from agentmemory import AgentMemoryClient
memory = AgentMemoryClient(
os.environ["AGENT_MEMORY_API_KEY"], # api_key: amk_…
os.environ["AGENT_MEMORY_BASE_URL"], # base_url, without /v1
os.environ["AGENT_MEMORY_WORKSPACE_ID"], # workspace_id
)AgentMemoryClient(api_key, base_url, workspace_id, principal=None). principal is an optional dict of defaults, {"agent_id": ..., "user_id": ...}.
Leave agent_id out: the server acts as the agent the key is bound to. Set it only when a key is bound to several agents. user_id is the end user a memory is about — never the agent — and requires a key with act for users enabled. See Identity and visibility.
Remember — add()
out = memory.add(
"Alex is vegetarian",
type="preference",
user_id="alex",
run_id="onboarding-42",
visibility="user",
source_kind="user",
tags={"facet": "diet"},
idempotency_key="onboarding-42:diet",
)
print(out["id"], out["merged"], out["operationId"])Writes go to POST /v1/memory/append with layer: "Workspace", so the memory is durable and searchable at once.
| Keyword | Default | Notes |
|---|---|---|
type | "decision" | decision, correction, violation, approved_pattern, artifact_ref, preference, raw. |
run_id | "default" | Use one stable run ID per task. |
user_id, agent_id | principal | |
tags | — | dict[str, str]. |
visibility | server default workspace | private, agent, user, workspace, tenant. |
source_kind | server default agent | agent, user, document or import. |
source_ref, observed_at, valid_from, valid_to | — | Provenance and valid time (ISO 8601 strings). |
idempotency_key | — | Sent as Idempotency-Key. |
Returns {"id", "merged", "contentReplaced", "operationId"}.
Recall — search()
context = memory.search("what can Alex eat?", user_id="alex", run_id="onboarding-42", token_budget=2000)
for entry in context["workspaceEntries"]:
print(entry["content"])
receipt_id = context["receiptId"]Keywords: run_id (default "default"), user_id, agent_id, top_n, mode ("assisted" or "deterministic"), rerank, include_superseded, as_of, token_budget, explain.
Returns the full context as a dict with the API's camelCase keys.
Report — outcome() and feedback()
These take a dict with the API's camelCase field names. Pass the recall's receiptId and the same runId:
memory.outcome({
"runId": "onboarding-42",
"receiptId": receipt_id,
"buildSuccess": True,
"testPassRate": 1.0,
})
memory.feedback({
"runId": "onboarding-42",
"receiptId": receipt_id,
"usedEntryIds": [context["workspaceEntries"][0]["id"]],
})Both return None. See Learning loop.
Read, explain and track
entry = memory.get(out["id"]) # dict | None
how = memory.explain(out["id"]) # dict | None
receipt = memory.operation(out["operationId"]) # {"status": "pending" | "indexed" | "failed", ...} | Noneget(), explain() and erasure() accept agent_id= and user_id= and return None when the memory does not exist or is not visible.
Update — update()
result = memory.update(out["id"], "Alex is vegan", if_match=entry["version"])
# {"status": "updated", "version": n} | {"status": "notFound"} | {"status": "forbidden"}
# | {"status": "versionMismatch", "currentVersion": n}Share — share()
memory.share(out["id"], "workspace") # "shared" | "notFound" | "forbidden" | "invalid"Forget — delete() and erasure()
result = memory.delete(out["id"])
if result["status"] == "pending":
print(memory.erasure(out["id"])) # {"status": "pending" | "complete" | ..., ...}
elif result["status"] == "blocked":
print(result["dependents"])
memory.delete(out["id"], cascade=True) # flag dependents for review and proceed
elif result["status"] == "forbidden":
print(result["reason"]) # "policy" | "tenant_scope"Statuses: deleted, pending, blocked, not_found, forbidden. Deleting needs a key with the delete capability, and never cascades unless you pass cascade=True. See Updating and forgetting.
Errors
Unexpected HTTP statuses raise AgentMemoryError (a RuntimeError) with .status and .body (the raw response text):
from agentmemory import AgentMemoryError
try:
memory.add("…")
except AgentMemoryError as err:
print(err.status, err.body)Not in the client
The Python client covers the memory lifecycle. For consolidation, blocks, user profiles, episodes, procedures, files and categories, call the REST API directly — or rely on the background worker to consolidate idle runs.
Method reference
| Method | Endpoint |
|---|---|
add(content, **kw) | POST /v1/memory/append |
search(query, **kw) | POST /v1/memory/retrieve |
get(memory_id, **kw) | GET /v1/memory/{id} |
update(memory_id, content, **kw) | PUT /v1/memory/{id} |
delete(memory_id, **kw) | DELETE /v1/memory/{id} |
erasure(memory_id, **kw) | GET /v1/memory/{id}/erasure |
operation(operation_id) | GET /v1/memory/operations/{id} |
explain(memory_id, **kw) | GET /v1/memory/{id}/explain |
share(memory_id, visibility, **kw) | POST /v1/memory/{id}/share |
outcome(report) | POST /v1/outcomes |
feedback(report) | POST /v1/retrieval-feedback |