axom_mcp_memory
Store, retrieve, search, and manage persistent memories in the Axom database for AI agents, supporting long-term context and complex tool chaining.
Instructions
Store, retrieve, search, and manage persistent memories in the Axom database.
Memory Types:
long_term: Reusable patterns, architectural decisions, gotchas
short_term: Task-specific context, debug notes, current task state
reflex: Learned heuristics ("Always check X before Y" patterns)
dreams: Experimental ideas, creative explorations
Naming Convention: [type][descriptor][YYYYMMDD] Example: bugfix_auth_timeout_20260203
Content Format (recommended): TASK|APPROACH|OUTCOME|GOTCHAS|RELATED
Actions:
write: Store a new memory
read: Retrieve a specific memory by name
list: List memories with optional filters
search: Full-text search across memories
delete: Remove a memory by name
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| action | Yes | Memory operation to perform | |
| name | No | Memory identifier (required for read/write/delete) | |
| content | No | Memory content (required for write) | |
| memory_type | No | Type of memory storage | |
| importance | No | Importance level | |
| tags | No | Tags for categorization | |
| query | No | Search query (required for search) | |
| limit | No | Maximum results to return | |
| expires_in_days | No | Override default expiration in days (default per type: short_term=30d, long_term=365d, reflex=90d, dreams=180d) |
Implementation Reference
- src/axom_mcp/handlers/memory.py:23-51 (handler)Main handler function for axom_mcp_memory tool. Validates input using MemoryInput schema and routes to appropriate action handlers (write, read, list, search, delete, associate).async def handle_memory(arguments: Dict[str, Any]) -> str: """Handle axom_mcp_memory tool calls. Args: arguments: Tool arguments containing action and parameters Returns: JSON string with operation result """ # Validate input input_data = MemoryInput(**arguments) action = input_data.action db = await get_db_manager() if action == "write": return await _handle_write(input_data, db) elif action == "read": return await _handle_read(input_data, db) elif action == "list": return await _handle_list(input_data, db) elif action == "search": return await _handle_search(input_data, db) elif action == "delete": return await _handle_delete(input_data, db) elif action == "associate": return await _handle_associate(input_data, db) else: return json.dumps({"error": f"Unknown action: {action}"})
- src/axom_mcp/handlers/memory.py:54-88 (handler)Helper function that handles the 'write' action - creates a new memory in the database with name, content, type, importance, tags, and expiration.async def _handle_write(input_data: MemoryInput, db) -> str: """Write a new memory.""" if not input_data.name: return json.dumps({"error": "name is required for write action"}) if not input_data.content: return json.dumps({"error": "content is required for write action"}) memory_type = input_data.memory_type or MemoryType.LONG_TERM importance = input_data.importance or ImportanceLevel.NORMAL try: memory_id = await db.create_memory( name=input_data.name, content=input_data.content, memory_type=memory_type.value if isinstance(memory_type, MemoryType) else memory_type, importance=importance.value if isinstance(importance, ImportanceLevel) else importance, tags=input_data.tags, source_agent=input_data.source_agent, expires_in_days=input_data.expires_in_days, ) return json.dumps( { "success": True, "id": str(memory_id), "name": input_data.name, "message": f"Memory '{input_data.name}' stored successfully", } ) except Exception as e: logger.error(f"Failed to write memory: {e}") return json.dumps({"error": str(e)})
- src/axom_mcp/handlers/memory.py:91-144 (handler)Helper function that handles the 'read' action - retrieves a memory by name including its associated memories with 1-level extension.async def _handle_read(input_data: MemoryInput, db) -> str: """Read a memory by name including associated memories.""" if not input_data.name: return json.dumps({"error": "name is required for read action"}) try: memory = await db.get_memory_by_name(input_data.name) if memory is None: return json.dumps({"error": f"Memory not found: {input_data.name}"}) # Get associated memories with 1-level extension associated_memories = await db.get_associated_memories(str(memory["id"])) # Format associated memories formatted_associations = [] for assoc in associated_memories: formatted_associations.append( { "id": str(assoc["id"]), "name": assoc["name"], "memory_type": assoc["memory_type"], "importance": assoc["importance"], "tags": assoc.get("tags", []), "created_at": assoc["created_at"].isoformat() if assoc.get("created_at") else None, } ) return json.dumps( { "success": True, "memory": { "id": str(memory["id"]), "name": memory["name"], "content": memory["content"], "memory_type": memory["memory_type"], "importance": memory["importance"], "tags": memory.get("tags", []), "source_agent": memory.get("source_agent"), "parent_memory_id": memory.get("parent_memory_id"), "created_at": memory["created_at"].isoformat() if memory.get("created_at") else None, "updated_at": memory["updated_at"].isoformat() if memory.get("updated_at") else None, "associated_memories": formatted_associations, }, } ) except Exception as e: logger.error(f"Failed to read memory: {e}") return json.dumps({"error": str(e)})
- src/axom_mcp/handlers/memory.py:147-194 (handler)Helper function that handles the 'list' action - lists memories with optional filters for type, importance, and limit.async def _handle_list(input_data: MemoryInput, db) -> str: """List memories with optional filters.""" limit = input_data.limit or 50 try: memory_type = None if input_data.memory_type: memory_type = ( input_data.memory_type.value if isinstance(input_data.memory_type, MemoryType) else input_data.memory_type ) importance = None if input_data.importance: importance = ( input_data.importance.value if isinstance(input_data.importance, ImportanceLevel) else input_data.importance ) memories = await db.list_memories( memory_type=memory_type, importance=importance, limit=limit, ) return json.dumps( { "success": True, "count": len(memories), "memories": [ { "name": m["name"], "memory_type": m["memory_type"], "importance": m["importance"], "tags": m.get("tags", []), "created_at": m["created_at"].isoformat() if m.get("created_at") else None, } for m in memories ], } ) except Exception as e: logger.error(f"Failed to list memories: {e}") return json.dumps({"error": str(e)})
- src/axom_mcp/handlers/memory.py:197-250 (handler)Helper function that handles the 'search' action - performs full-text search across memories with query, type, importance, and tag filters.async def _handle_search(input_data: MemoryInput, db) -> str: """Search memories by query.""" if not input_data.query: return json.dumps({"error": "query is required for search action"}) limit = input_data.limit or 10 try: memory_type = None if input_data.memory_type: memory_type = ( input_data.memory_type.value if isinstance(input_data.memory_type, MemoryType) else input_data.memory_type ) importance = None if input_data.importance: importance = ( input_data.importance.value if isinstance(input_data.importance, ImportanceLevel) else input_data.importance ) memories = await db.search_memories( query=input_data.query, memory_type=memory_type, importance=importance, tags=input_data.tags, limit=limit, ) return json.dumps( { "success": True, "query": input_data.query, "count": len(memories), "results": [ { "name": m["name"], "content": m["content"][:500] + "..." if len(m["content"]) > 500 else m["content"], "memory_type": m["memory_type"], "importance": m["importance"], "relevance": m.get("rank", 0), } for m in memories ], } ) except Exception as e: logger.error(f"Failed to search memories: {e}") return json.dumps({"error": str(e)})