SDK Reference

All methods are available on a GraphMemClient instance. Authenticate once — all calls use the same client.

setup
from graphmem import GraphMemClient

gm = GraphMemClient(
    "https://api.graphmem.site",
    api_key="gm_sk_...",
)

conversations

A conversation is the top-level container. All memory nodes and topics belong to one.

gm.conversations.create(metadata: dict | None = None) → str

Create a new conversation. Returns a conversation_id — pass this to all memory and topic calls.

Parameters

Name	Type	Default	Description
metadata	dict \| None	None	Arbitrary key-value pairs stored with the conversation (e.g. user_id, session).

→ str

UUID string — the conversation identifier. Store this to add memory nodes later.

example
conv_id = gm.conversations.create(metadata={"user_id": "u_123"})
# "3f7a1b2c-9e4d-..."

HTTP: POST /conversations

gm.conversations.list() → list[dict]

List all conversations for your API key.

→ list[dict]

conversation_idstrUUID of the conversation.

node_countintNumber of complete memory nodes.

last_topic_recomputestr | NoneISO timestamp of last topic clustering run.

metadatadictMetadata passed at creation time.

example
convs = gm.conversations.list()
for c in convs:
    print(c["conversation_id"], c["node_count"])

HTTP: GET /conversations

gm.conversations.stats(conversation_id: str) → dict

Graph statistics for a conversation — node count, edge count, density, topic count, average degree.

→ dict

node_countintTotal complete memory nodes.

edge_countintNeptune graph edges (cosine ≥ threshold).

topic_countintNumber of detected topics.

avg_degreefloatAverage edges per node.

graph_densityfloatRatio of actual to possible edges.

example
s = gm.conversations.stats(conv_id)
print(f"{s['node_count']} nodes · {s['topic_count']} topics")

HTTP: GET /conversations/{conversation_id}/stats

gm.conversations.export(conversation_id: str) → list[dict]

Export all memory nodes in a conversation as a flat list of dicts. Useful for migration or offline analysis.

HTTP: GET /conversations/{conversation_id}/export

gm.conversations.delete(conversation_id: str) → None

Delete a conversation and all its nodes, edges, and topics. Irreversible.

HTTP: DELETE /conversations/{conversation_id}

memory

Memory nodes are prompt + response pairs. A node becomes complete once both parts are stored, at which point it is embedded and wired into the similarity graph asynchronously.

gm.memory.add_prompt(
    conversation_id: str,
    content: str,
    *,
    timestamp: str | None = None,
    metadata: dict | None = None,
) → str

Store a user prompt. Returns a memory_id — pass it to add_response() to complete the node. The node is pending and not searchable until a response is attached.

Parameters

Name	Type	Default	Description
conversation_id	str	required	The conversation this prompt belongs to.
content	str	required	The user's message text.
timestamp	str \| None	None	ISO 8601 timestamp. Defaults to server time if omitted.
metadata	dict \| None	None	Arbitrary key-value pairs attached to this node.

→ str

UUID string — the memory_id of the new pending node.

example
memory_id = gm.memory.add_prompt(
    conv_id,
    "How does async/await work in Python?",
)
# "9e4c3d2a-..."

HTTP: POST /memory/prompt

gm.memory.add_response(
    memory_id: str,
    conversation_id: str,
    content: str,
    *,
    timestamp: str | None = None,
) → dict

Attach the assistant's response to a pending node. This completes the node and immediately queues it for embedding + graph insertion (SQS → Lambda → Bedrock + Neptune). The node becomes searchable within seconds.

Parameters

Name	Type	Default	Description
memory_id	str	required	The memory_id returned by `add_prompt()`.
conversation_id	str	required	Must match the conversation used in `add_prompt()`.
content	str	required	The assistant's response text.
timestamp	str \| None	None	ISO 8601 timestamp. Defaults to server time.

example — typical agent loop
user_msg = "How does async/await work in Python?"

memory_id = gm.memory.add_prompt(conv_id, user_msg)
reply     = llm.complete(user_msg)
gm.memory.add_response(memory_id, conv_id, reply)
# node is now complete and queued for embedding

HTTP: POST /memory/{memory_id}/response

gm.memory.add_batch(items: list[dict]) → list[str]

Add up to 100 complete prompt + response pairs in a single call. All nodes are immediately queued for embedding. Use this to pre-load existing conversation history.

Item schema

Field	Type	Default	Description
conversation_id	str	required	Target conversation.
prompt	str	required	User message.
response	str	required	Assistant response.
timestamp_prompt	str \| None	None	ISO 8601 timestamp for the prompt.
timestamp_response	str \| None	None	ISO 8601 timestamp for the response.
metadata	dict \| None	None	Arbitrary key-value pairs.

example — load history
ids = gm.memory.add_batch([
    {
        "conversation_id": conv_id,
        "prompt":          "What is a closure?",
        "response":        "A closure captures variables from its enclosing scope...",
    },
    # ...up to 100 items
])
# ["uuid1", "uuid2", ...]

HTTP: POST /memory/batch

gm.memory.recall(
    conversation_id: str,
    query: str,
    *,
    k: int = 5,
    threshold: float = 0.4,
    strategy: str = "semantic",
    recency_alpha: float = 0.7,
    recency_half_life: float | None = None,
    always_include_recent: int = 0,
    max_tokens: int | None = None,
) → dict

The core retrieval method. Embeds the query, runs a multi-set graph traversal, applies a token budget, and returns ordered context ready to inject into your LLM prompt. Internally returns the union of Set B (top-K cosine), Set A′ (threshold neighbours), Set C (Neptune 1-hop graph expansion), and Set R (recency guard).

Parameters

Name	Type	Default	Description
conversation_id	str	required	Which conversation to search.
query	str	required	The current user message or search query.
k	int	5	Top-K nodes for Set B (cosine search).
threshold	float	0.4	Cosine similarity floor for Set A′ neighbours.
strategy	"semantic" \| "semantic_recency"	"semantic"	`semantic_recency` blends cosine score with a recency decay function.
recency_alpha	float	0.7	Weight of cosine vs recency when using `semantic_recency`. 1.0 = pure cosine, 0.0 = pure recency.
recency_half_life	float \| None	None	Seconds for the recency score to halve. Defaults to server-side value if None.
always_include_recent	int	0	Always include last N nodes regardless of relevance (Set R). Exempt from token budget cuts.
max_tokens	int \| None	None	Token budget for returned context. Nodes cut by relevance score first; Set R is never cut.

→ dict

messageslist[dict]Ordered memory nodes. Each node has prompt, response, score, set_membership, token_count.

total_tokensintTotal tokens in the returned context.

strategy_usedstrThe strategy that was applied.

node_countintNumber of nodes returned.

example — inject context into LLM
result = gm.memory.recall(
    conv_id,
    user_message,
    k=5,
    always_include_recent=2,
    max_tokens=2000,
)

context = "\n\n".join(
    f"User: {m['prompt']}\nAssistant: {m['response']}"
    for m in result["messages"]
)
# inject `context` into your LLM system prompt
print(f"Retrieved {result['total_tokens']} tokens from {result['node_count']} nodes")

HTTP: POST /memory/recall

gm.memory.get(memory_id: str) → dict

Fetch a single memory node by ID. Embedding vector is not included in the response.

→ dict

memory_idstrUUID of the node.

promptstrThe user message.

responsestr | NoneThe assistant response. None if the node is still pending.

statusstr"pending" | "complete" | "embedded".

timestamp_promptstrISO 8601 timestamp of the prompt.

metadatadictMetadata attached at creation.

HTTP: GET /memory/{memory_id}

gm.memory.neighbours(
    memory_id: str,
    *,
    threshold: float = 0.7,
    limit: int | None = None,
    order: str = "relevance",
) → list[dict]

Return graph neighbours of a node — other nodes connected by a cosine-similarity edge at or above threshold. Useful for exploring related memories or debugging graph structure.

Parameters

Name	Type	Default	Description
threshold	float	0.7	Minimum cosine similarity edge weight to include.
limit	int \| None	None	Maximum neighbours to return. None = all above threshold.
order	"relevance" \| "timestamp"	"relevance"	Sort by edge weight (relevance) or insertion time.

HTTP: GET /memory/{memory_id}/neighbours

gm.memory.delete(memory_id: str) → None

Delete a memory node and its edges. If the node belonged to a topic, that topic is marked dirty and re-evaluated on the next recompute.

HTTP: DELETE /memory/{memory_id}

topics

Topics are message clusters auto-detected via Louvain community detection every 20 nodes. Each is labeled and summarized by an LLM. Use them for fine-grained context control and maximum token efficiency.

gm.topics.list(
    conversation_id: str,
    *,
    status: str | None = None,
) → list[dict]

List all topics for a conversation. This is the entry point — get topic IDs here, then pass them to other methods.

Parameters

Name	Type	Default	Description
status	"active" \| "dormant" \| "resolved" \| None	None	Filter by topic lifecycle status. None returns all.

→ list[dict]

topic_idstrUUID of the topic.

labelstrLLM-generated short label (e.g. "Python async patterns").

message_countintNumber of memory nodes assigned to this topic.

statusstr"active" | "dormant" | "resolved".

last_activestrISO timestamp of the last message in this topic.

example
topics = gm.topics.list(conv_id, status="active")
for t in topics:
    print(t["label"], t["message_count"])

HTTP: GET /topics?conversation_id={id}

gm.topics.current(
    conversation_id: str,
    *,
    mode: str = "centroid",
) → dict | None

Returns the topic the conversation is currently in — based on the last complete memory node. Use this for topic shift detection without sending a query.

Parameters

Name	Type	Default	Description
mode	"centroid" \| "hybrid" \| "llm"	"centroid"	`hybrid` blends centroid + summary. `llm` uses a full LLM inference — slower but most accurate.

HTTP: GET /topics/current?conversation_id={id}

gm.topics.status(topic_id: str) → str

Returns "active", "dormant", or "resolved". Use this to decide which topics contribute context — skip resolved ones, inject dormant ones as summaries only.

example — filter context by lifecycle
active_topics = [
    t for t in gm.topics.list(conv_id)
    if gm.topics.status(t["topic_id"]) == "active"
]

HTTP: GET /topics/{topic_id}/status

gm.topics.summary(topic_id: str) → str

Returns an LLM-generated summary paragraph for the entire topic. When a topic is relevant but not the primary focus, inject the summary instead of all raw messages — same semantic coverage at a fraction of the tokens.

example — compressed context injection
# An old topic has 20 raw messages (~3000 tokens).
# Its summary is one paragraph (~80 tokens).
summary = gm.topics.summary(old_topic_id)
system_prompt += f"\n\nEarlier context: {summary}"

HTTP: GET /topics/{topic_id}/summary

gm.topics.relevance(
    conversation_id: str,
    query: str,
    *,
    top_n: int = 3,
) → list[dict]

Score all topics against a query and return the top-N most relevant, ranked by cosine similarity to their centroids. This is Stage 1 of two-stage retrieval — O(T) comparisons instead of O(N).

Parameters

Name	Type	Default	Description
query	str	required	The prompt or search query to score topics against.
top_n	int	3	How many topics to return.

→ list[dict]

topic_idstrUUID of the topic.

labelstrTopic label.

scorefloatCosine similarity of the query to the topic centroid (0–1).

HTTP: GET /topics/relevance?conversation_id={id}&query={q}

gm.topics.messages(
    topic_id: str,
    conversation_id: str,
    *,
    query: str | None = None,
    max_tokens: int | None = None,
) → list[dict]

Retrieve all memory nodes that belong to a single topic. Optionally narrow with a similarity query and apply a token budget.

Parameters

Name	Type	Default	Description
query	str \| None	None	Further rank messages within the topic by similarity to this query.
max_tokens	int \| None	None	Token budget. Least-relevant messages cut first.

HTTP: GET /topics/{topic_id}/messages

gm.topics.retrieve(
    conversation_id: str,
    query: str,
    *,
    top_n: int = 3,
    max_tokens: int | None = None,
) → list[dict]

Full two-stage retrieval in one call. Internally runs relevance() to find top-N matching topics, then retrieves messages from each. Most token-efficient retrieval path for long conversations — compares T topic centroids, not N individual messages.

Parameters

Name	Type	Default	Description
query	str	required	The current user prompt or search query.
top_n	int	3	How many topics to pull messages from.
max_tokens	int \| None	None	Total token budget across all retrieved messages.

example — two-stage retrieval
messages = gm.topics.retrieve(
    conv_id,
    user_message,
    top_n=3,
    max_tokens=3000,
)
# Messages from the 3 most relevant topics, combined & token-budgeted

HTTP: POST /topics/retrieve

gm.topics.get(topic_id: str) → dict

Full topic object — label, description, summary, coherence score, is_mixed flag, sub-themes, status, message count.

→ dict

labelstrShort LLM-generated label.

descriptionstrLonger description of the topic.

summarystrFull LLM-generated summary paragraph.

coherencefloat0–1 score of how semantically tight the cluster is.

is_mixedboolTrue if the cluster spans multiple sub-themes.

sub_themeslist[str]Sub-themes identified by the LLM.

statusstr"active" | "dormant" | "resolved".

is_dirtyboolTrue if a member node was deleted and the topic needs recompute.

HTTP: GET /topics/{topic_id}

gm.topics.recompute(conversation_id: str) → dict

Manually trigger Louvain clustering + LLM labeling for a conversation. Topics auto-recompute every 20 nodes — use this to force early recompute after bulk deletions or large batch imports.

HTTP: POST /topics/recompute?conversation_id={id}

auth

Key management requires a Cognito JWT token, not an API key. Pass it as token= at client construction time.

gm.auth.create_key(
name: str | None = None,
permissions: dict | None = None,
) → dict

Create an API key. The full key is returned once in the response — it cannot be retrieved again. Store it immediately.

→ dict

keystrFull API key — only returned once. Store securely.

key_idstrID used to revoke this key later.

key_prefixstrFirst 12 characters — safe to display in UIs.

HTTP: POST /auth/keys · requires Bearer JWT

gm.auth.list_keys() → list[dict]

List all API keys for the authenticated user. Full key values are never returned — only prefix, status, created_at, and metadata.

HTTP: GET /auth/keys · requires Bearer JWT

gm.auth.revoke_key(key_id: str) → None

Revoke an API key immediately. Any request using it will return 401.

HTTP: DELETE /auth/keys/{key_id} · requires Bearer JWT

utils

gm.utils.count_tokens(text: str) → int

Estimate token count for any string using the same tokenizer GraphMem uses internally for max_tokens budgeting.

example
n = gm.utils.count_tokens("How does async/await work in Python?")
# 9

HTTP: POST /utils/count_tokens

gm.utils.health() → dict

Liveness check. No auth required. Returns {"status": "ok"} when the API is up.

HTTP: GET /health