Python SDK
Full control over agent memory from any Python app.
Install
pip install octopodaRequires Python 3.9+. Dependencies: requests and pydantic.
Quick Start
from octopoda import Octopoda client = Octopoda(api_key="sk-octopoda-YOUR_KEY")agent = client.agent("my-agent") # Store a memoryagent.write("user:name", "Alice") # Read it backname = agent.read("user:name")print(name) # Alice # Search by meaningresults = agent.search("what is the user's name?")print(results)Tip: Check octopodas.com/dashboard — you should see "my-agent" with 1 memory.
Client Methods
client = Octopoda( api_key="sk-octopoda-...", # or set OCTOPODA_API_KEY env var base_url="https://api.octopodas.com", # default timeout=30, # seconds)| Method | Returns | Description |
|---|---|---|
| client.agent("id") | Agent | Create or reconnect to an agent |
| client.agents() | list | List all your agents |
| client.me() | dict | Your account info |
| client.system_metrics() | dict | System-wide metrics |
| client.shared_spaces() | list | List shared memory spaces |
| client.read_shared("space", "key") | dict | Read from shared space |
| client.close() | — | Close the HTTP session |
Agent Methods
| Method | Returns | Description |
|---|---|---|
| agent.write(key, value, tags=None) | dict | Store a memory |
| agent.write_batch(items) | dict | Store multiple: [{"key": "k", "value": "v"}] |
| agent.read(key) | value or None | Read a memory |
| agent.list(limit=50) | dict | List all memories |
| agent.search(query, limit=10) | list | Semantic search |
| agent.keys(prefix="", limit=50) | list | Search by key prefix |
| agent.history(key) | list | All versions of a memory |
| agent.info() | dict | Agent details |
| agent.delete() | dict | Deregister agent |
Memory Management
# Delete specific memoryagent.forget("old_key") # Delete old memories (preserves critical importance)agent.forget_stale(max_age_seconds=604800) # 7 days # Delete by tagagent.forget_by_tag("temporary") # Find duplicates (preview — changes nothing)report = agent.consolidate(dry_run=True)print(f"Found {report['duplicate_groups']} duplicate groups") # Merge duplicates (keeps newest, removes duplicates)result = agent.consolidate(dry_run=False) # Memory health scorehealth = agent.memory_health()print(f"Score: {health['score']}/100")for issue in health['issues']: print(f" - {issue}")Goal Tracking
agent.set_goal("Migrate database", milestones=["Backup", "Migrate", "Validate"])agent.update_progress(milestone_index=0, note="Backup done")goal = agent.get_goal()# {"progress": 0.33, "status": "active"}Agent Messaging
agent.send_message("agent_b", "Found a bug", message_type="alert")messages = agent.read_messages(unread_only=True)agent.broadcast("Deployment starting")Export & Import
bundle = agent.export_memories() # Export as JSONagent.import_memories(bundle, overwrite=False) # Import (skip existing)Filtered Search
results = agent.search_filtered( query="deployment issues", tags=["production"], importance="critical", max_age_seconds=86400)Advanced Writes
| Method | Returns | Description |
|---|---|---|
| agent.write_ttl("key", "val", ttl_seconds=3600) | dict | Auto-expiring memory |
| agent.write_important("key", "val", importance="critical") | dict | Write with importance level |
| agent.write_safe("key", "val") | dict | Write with conflict check |
| agent.check_conflicts("key", "new_val") | dict | Check before writing |
| agent.flush() | — | Wait for background processing |
Process Conversation (Recommended)
The easiest way to add memory. Pass your conversation and Octopoda extracts knowledge automatically:
result = agent.process_conversation([ {"role": "user", "content": "I'm Alice, CTO at TechCorp"}, {"role": "assistant", "content": "Nice to meet you! How can I help?"}, {"role": "user", "content": "We need to optimize our ML pipeline"},]) print(f"Stored {result['memories_stored']} memories") # Get relevant context for a querycontext = agent.get_context("what does the user want?")print(context)Other Utility Methods
| Method | Returns | Description |
|---|---|---|
| agent.decide("deploy", "all tests passed") | dict | Log a decision with reasoning |
| agent.audit() | list | Get audit trail |
| agent.recover() | dict | Crash recovery |
| agent.share("space", "key", "value") | dict | Write to shared memory |
| agent.share_safe("space", "key", "value") | dict | Shared with conflict detection |
| agent.snapshot("before_deploy") | dict | Create checkpoint |
| agent.restore("before_deploy") | dict | Rollback to checkpoint |
Rate Limits
| Plan | Rate Limit |
|---|---|
| Free | 100 requests/min |
| Pro | 300 requests/min |
| Business | 1,000 requests/min |
| Scale | 5,000 requests/min |
Using AI to Set Up
Paste this into Cursor or Claude Code:
Install Octopoda memory for my Python project: 1. Run: pip install octopoda2. Test it with this code: from octopoda import Octopodaclient = Octopoda(api_key="sk-octopoda-MY_KEY_HERE")agent = client.agent("test-agent")result = agent.write("test", "hello world")print("Write:", result)value = agent.read("test")print("Read:", value) Tell me if both succeed.Troubleshooting
"ModuleNotFoundError: No module named 'octopoda'"
Run pip install octopoda. If using Python 3, try pip3 install octopoda.
"Invalid API key" or 401 errors
Check your key starts with sk-octopoda-. Get it from octopodas.com/dashboard/settings.
"Connection refused" or timeout
Open https://api.octopodas.com/health in your browser. If it shows {"status": "ok"}, the server is fine — check your firewall/VPN.
Import works but methods fail
Make sure you're on the latest version: pip install octopoda --upgrade
Monitor Your Agents
Once set up, your agent's memories appear in real-time on your dashboard. See stored memories, search quality, knowledge graph, and agent performance, all from your browser.
Open Dashboard