Python SDK

pip install agentverse-memory

pip install "agentverse-memory[langchain]"   # LangChain integration
pip install "agentverse-memory[crewai]"      # CrewAI integration
pip install "agentverse-memory[all]"         # All integrations
 
# Verify
python -c "import agentverse_memory; print(agentverse_memory.__version__)"

export AVMEM_API_KEY="avmem_sk_your_key_here"

from agentverse_memory import MemoryClient
 
# Reads AVMEM_API_KEY from environment automatically
client = MemoryClient(agent_id="my-agent")
 
# Or pass explicitly (useful for multi-tenant setups)
client = MemoryClient(
    api_key="avmem_sk_your_key_here",
    agent_id="my-agent",
    base_url="https://am-server-jbneh74b5q-uc.a.run.app",
    timeout=10.0,    # seconds
)

from agentverse_memory import MemoryClient
 
client = MemoryClient(agent_id="research-agent-1")
 
episode = client.episodes.store(
    content="User asked for Q1 revenue breakdown by region. "
            "Retrieved from Salesforce. Delivered 3-bullet summary via Slack.",
    metadata={
        "user_id":    "user_42",
        "session_id": "sess_abc123",
        "tags":       ["finance", "q1", "slack"],
    },
    importance=0.8,   # 0.0–1.0; higher = pheromone boost on retrieval
)
 
print(episode.id)           # ep_01hxq7kn3mabc
print(episode.created_at)   # 2026-05-14T08:00:00.042Z
# No LLM calls at write time — 0.035ms p95 in-process engine write op (am-local)

results = client.episodes.query(
    query="What did the user ask about sales?",
    limit=5,
    min_score=0.4,
)
 
for r in results:
    print(f"[{r.score:.2f}] {r.content[:80]}")
    print(f"         ↳ {r.metadata.get('tags', [])}")
# [0.94] User asked for Q1 revenue breakdown by region.
#          ↳ ['finance', 'q1', 'slack']

# Store a knowledge triple (subject, predicate, object)
fact = client.facts.store(
    subject="Acme Corp",
    predicate="signed_contract_with",
    object_="Fetch.ai",      # object_ avoids Python keyword conflict
    metadata={
        "valid_at":   "2026-01-15",
        "confidence": 0.95,
        "source":     "CRM",
    },
)
print(fact.id)  # fact_02hyr9mp4n...
 
# Connect more facts to build the graph
client.facts.store(subject="Fetch.ai", predicate="headquartered_in", object_="London")
 
# Query the graph with natural language
hits = client.facts.query(query="Where is Fetch.ai based?", limit=3)
for h in hits:
    print(f"{h.subject} → {h.predicate} → {h.object_}")
# Fetch.ai → headquartered_in → London

import asyncio
from agentverse_memory import AsyncMemoryClient
 
async def main():
    async with AsyncMemoryClient(agent_id="async-agent") as client:
 
        # Store a semantic fact
        fact = await client.facts.store(
            subject="Acme Corp",
            predicate="signed_contract_with",
            object_="Fetch.ai",
            metadata={"valid_at": "2026-01-15", "confidence": 0.95},
        )
 
        # Traverse the knowledge graph from a seed node
        graph = await client.facts.traverse_graph(
            seed_node="Acme Corp", max_depth=2, limit=20,
        )
        for node in graph.nodes:
            print(f"{node.subject} --[{node.predicate}]--> {node.object_}")
 
asyncio.run(main())

from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from agentverse_memory.integrations.langchain import (
    AvmemChatMessageHistory,
    AvmemMemory,
)
 
# Session-scoped episodic history — persists across process restarts
history = AvmemChatMessageHistory(
    agent_id="langchain-assistant",
    session_id="sess_abc123",
)
 
memory = AvmemMemory(history=history)
 
chain = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o"),
    memory=memory,
)
 
# First session
chain.predict(input="Acme Corp just renewed their contract for $420K.")
 
# Second session — new process, same memory
# chain.predict(input="What's Acme's contract value?")
# → "Acme Corp renewed their contract for $420K."

from agentverse_memory.integrations.langchain import AvmemRetriever
from langchain.chains import RetrievalQA
 
retriever = AvmemRetriever(
    agent_id="langchain-assistant",
    memory_types=["episodic", "semantic"],   # query both stores
    limit=8,
)
 
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o"),
    retriever=retriever,
)
 
answer = qa_chain.run("What contracts did Acme Corp renew recently?")

from crewai import Agent, Task, Crew
from agentverse_memory.integrations.crewai import AgentverseMemoryTool
 
# Each CrewAI agent gets its own memory namespace
memory_tool = AgentverseMemoryTool(agent_id="crew-research-agent")
 
researcher = Agent(
    role="Senior Research Analyst",
    goal="Research competitors and synthesize findings into actionable insights",
    backstory="You are a thorough analyst who never forgets a finding.",
    tools=[memory_tool],
)
 
task = Task(
    description=(
        "Research Mem0's pricing changes from April 2026. "
        "Store each key finding as a memory. "
        "Query past research before starting so you don't repeat work."
    ),
    agent=researcher,
)
 
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

# Orchestrator creates a shared memory space
space = client.spaces.create(name="research-team-memory")
print(space.id, space.invite_token)
 
# Worker agent joins the space
worker = MemoryClient(api_key="...", agent_id="worker-agent-1")
worker.spaces.join(space_id=space.id, invite_token=space.invite_token)
 
# Any agent in the space queries the shared knowledge graph
results = client.spaces.query(
    space_id=space.id,
    query="What did the team discover about Acme Corp?",
    limit=10,
)

# Traverse from a seed concept — pheromone-weighted edges
graph = client.facts.traverse_graph(seed_node="Acme Corp", max_depth=2, limit=20)
 
for node in graph.nodes:
    print(f"{node.subject} --[{node.predicate}]--> {node.object_}  "
          f"(weight: {node.pheromone_weight:.3f})")
# Acme Corp --[signed_contract_with]--> Fetch.ai    (weight: 0.847)
# Fetch.ai  --[headquartered_in]-----> London       (weight: 0.612)
 
# A* pathfinding between any two concepts (Builder tier and above)
path = client.facts.find_path(source="Acme Corp", target="London", max_depth=4)
print(" → ".join(n.label for n in path.nodes))  # Acme Corp → Fetch.ai → London
print(f"Pheromone path score: {path.total_weight:.3f}")

from agentverse_memory import MemoryClient
from agentverse_memory.exceptions import (
    AuthError,        # 401 — invalid or missing API key
    ForbiddenError,   # 403 — feature not on your plan
    NotFoundError,    # 404 — resource does not exist
    RateLimitError,   # 429 — monthly or burst limit exceeded
    ServerError,      # 500/503 — server-side fault
    AvmemError,       # base class for all SDK errors
)
import time
 
client = MemoryClient(agent_id="my-agent")
 
try:
    episode = client.episodes.store(content="Important event.")
except AuthError:
    print("Invalid API key. Check AVMEM_API_KEY.")
except ForbiddenError as e:
    print(f"Feature requires upgrade: {e.required_plan}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
    time.sleep(e.retry_after)
    episode = client.episodes.store(content="Important event.")
except AvmemError as e:
    print(f"Unexpected error [{e.code}]: {e}")

client = MemoryClient(
    agent_id="my-agent",
    max_retries=3,        # retry on 429 + 5xx
    retry_backoff=2.0,    # exponential factor (2^attempt seconds)
)
# All calls now automatically retry transient errors

from agentverse_memory import MemoryClient
 
client = MemoryClient(
    api_key="avmem_sk_...",       # or AVMEM_API_KEY env var
    agent_id="my-agent",           # or AVMEM_AGENT_ID env var
    base_url="https://am-server-jbneh74b5q-uc.a.run.app",  # or AVMEM_BASE_URL env var
    timeout=10.0,                  # request timeout in seconds
    max_retries=3,                 # auto-retry on 5xx / 429
    retry_backoff=2.0,             # exponential factor
)