ConvoMem

Python SDK

Install, initialize, and use the ConvoMem Python SDK.

The ConvoMem Python SDK supports Python 3.11+ with both sync and async clients.

Install

pip install convomem
 
# or with uv
uv add convomem

Initialize

import os
from convomem import ConvoMem
 
client = ConvoMem(api_key=os.environ["CONVOMEM_API_KEY"])

Optional configuration:

client = ConvoMem(
    api_key=os.environ["CONVOMEM_API_KEY"],
    timeout=10.0,    # seconds per attempt (default 10)
    max_retries=3,   # retries on 429 / 5xx / timeout (default 3)
)

Retries wait a random delay of up to retry_delay × attempt (retry_delay defaults to 1 second). They do not read the Retry-After header.

Identity

Every method accepts a CustomerIdentity that tells ConvoMem which customer to read or write. If customer_id is set, the SDK routes to the direct /customers/:id/… endpoint. Otherwise the server resolves the customer from email, phone, or external_id.

from convomem import CustomerIdentity
 
# By known ID — most direct
identity = CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12")
 
# By email — server resolves the customer
identity = CustomerIdentity(email="[email protected]")
 
# By phone (E.164 format recommended)
identity = CustomerIdentity(phone="+14155550100")
 
# By external system ID
identity = CustomerIdentity(external_id="crm-456")

Capture

Send conversation turns for background memory extraction. Acceptance is not completion: poll the returned capture ID before checking extracted memories.

client.capture(
    [
        {"role": "user",      "content": "I need help with my order"},
        {"role": "assistant", "content": "Of course, let me look that up."},
    ],
    CustomerIdentity(email="[email protected]"),
    channel="CHAT",
)
result = client.capture(
    [{"role": "user", "content": "I prefer email contact"}],
    CustomerIdentity(email="[email protected]"),
)
status = client.get_capture_status(result.capture_id)
print(status.status)  # queued, processing, completed, or failed

Use bounded polling with a delay and timeout. A failed capture needs investigation; it is not evidence that the customer has no memories.

To close the conversation, send the final turns with end_conversation=True (or escalate=True) rather than calling end_conversation() right after a capture:

client.capture(
    [{"role": "assistant", "content": "All sorted — anything else?"}],
    CustomerIdentity(phone="+14155550101"),
    channel="VOICE",
    end_conversation=True,
    outcome="Refund issued",
)

Cold open

At the start of a conversation, before the customer has said why they are calling, get the profile and a call-opening brief that leads with any unresolved issue:

brief = client.cold_open(CustomerIdentity(phone="+14155550100"), auto_create=True)
 
print(brief.context)
if brief.open_issue:
    print(f"Still open: {brief.open_issue.summary}")

Look up context

Recall semantically relevant memories for a topic before generating a reply.

ctx = client.lookup(
    "order and shipping preferences",
    CustomerIdentity(email="[email protected]"),
)
 
print(ctx.context)          # prompt-ready context string
for mem in ctx.memories:
    print(f"  - {mem.content}")

Agent loop pattern

import os
from convomem import ConvoMem, CustomerIdentity
 
client = ConvoMem(api_key=os.environ["CONVOMEM_API_KEY"])
 
def handle_message(user_msg: str, email: str):
    identity = CustomerIdentity(email=email)
 
    # 1. Recall context
    ctx = client.lookup("customer support", identity)
 
    # 2. Build prompt + call your LLM
    reply = call_llm(ctx.context, user_msg)
 
    # 3. Capture the turns
    result = client.capture(
        [
            {"role": "user",      "content": user_msg},
            {"role": "assistant", "content": reply},
        ],
        identity,
        channel="CHAT",
    )
 
    return reply, result.conversation_id

Memories

from convomem import CustomerIdentity
 
identity = CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12")
 
# List all memories
result = client.list_memories(identity, page=1, limit=20)
for mem in result.memories:
    print(f"- {mem.content}")
 
# Add manually
memory = client.add_memory(
    "Prefers email contact",
    identity,
    category="preference",
)
 
# Update (requires customer_id)
client.update_memory(
    "c2a7e4b1-9f3d-4a6e-8b5c-1d2e3f4a5b6c",
    identity,
    content="Prefers email and text contact",
)
 
# Delete (requires customer_id)
client.delete_memory("c2a7e4b1-9f3d-4a6e-8b5c-1d2e3f4a5b6c", identity)

Customers

from convomem import CustomerIdentity
 
# Create
customer = client.create_customer(
    name="Alice",
    email="[email protected]",
)
 
# Get by any identity
customer = client.get_customer(CustomerIdentity(email="[email protected]"))
print(f"Customer ID: {customer.id}")
 
# Update — by ID (direct) or by identity (server resolves)
updated = client.update_customer(
    CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12"),
    name="Alice Smith",
)
 
# List
result = client.list_customers(page=1, limit=20)
print(f"{len(result.customers)} of {result.total}")
 
# Delete
client.delete_customer(CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12"))

Conversations

from convomem import CustomerIdentity
 
identity = CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12")
 
# Start
conversation = client.start_conversation(identity, "CHAT")
print(f"Conversation: {conversation.id}")
 
# List
result = client.list_conversations(identity, page=1, limit=20)
 
# End — path-based when both IDs are known
client.end_conversation(
    "87c53d10-85f8-4447-920f-b1e847c23a1a",
    identity,
    outcome="Resolved billing issue",
)
 
# End — flat route: closes the customer's active conversation
client.end_conversation(
    None,
    CustomerIdentity(email="[email protected]"),
    outcome="Resolved",
)
 
# Escalate
client.escalate_conversation(
    "87c53d10-85f8-4447-920f-b1e847c23a1a",
    identity,
    reason="Complex billing dispute",
)

Handoff

Generate a briefing for the human agent taking over the conversation.

handoff = client.get_handoff(CustomerIdentity(email="[email protected]"))
 
print(f"Narrative: {handoff.narrative}")
for mem in handoff.key_memories:
    print(f"  - {mem.content}")

Route a live transfer

Resolve which human destination a transfer should go to. Destinations are configured in the dashboard.

route = client.route_handoff(CustomerIdentity(phone="+14155550100"))
 
if route.destination:
    print(route.destination.value, route.source)

Embed tokens

result = client.create_embed_token(
    CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12"),
    ttl_seconds=3600,
)
 
print(f"Token: {result.token}")
print(f"Expires in: {result.expires_in}s")

Error handling

from convomem import ConvoMemApiError
 
try:
    customer = client.get_customer(CustomerIdentity(customer_id="f3b39d1b-2522-4a00-98df-8cdde783ba12"))
except ConvoMemApiError as e:
    print(f"Status: {e.status}")
    print(f"Body: {e.body}")

Async client

import asyncio
import os
from convomem import AsyncConvoMem, CustomerIdentity
 
async def main():
    async with AsyncConvoMem(api_key=os.environ["CONVOMEM_API_KEY"]) as client:
        ctx = await client.lookup(
            "billing question",
            CustomerIdentity(email="[email protected]"),
        )
        print(ctx.context)
 
        await client.capture(
            messages=[{"role": "user", "content": "What's my invoice?"}],
            identity=CustomerIdentity(email="[email protected]"),
            channel="CHAT",
        )
 
asyncio.run(main())

Other SDKs