"""
MCP Tool Definitions and Handlers for BuildAutomata Memory System
Copyright 2025 Jurden Bruce
All tool responses return JSON for AI consumption, not human-formatted text.
"""
import json
import uuid
import logging
from datetime import datetime
from typing import List, Dict, Any
from mcp.types import Tool, TextContent
class DateTimeEncoder(json.JSONEncoder):
"""JSON encoder that handles datetime objects"""
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
logger = logging.getLogger("buildautomata-memory.mcp-tools")
# Import models - handle both package and direct execution
try:
from .models import Memory
except ImportError:
from models import Memory
def get_tool_definitions() -> List[Tool]:
"""Return list of available MCP tools"""
return [
Tool(
name="store_memory",
description="Store a new memory with flexible categorization",
inputSchema={
"type": "object",
"properties": {
"content": {"type": "string", "description": "Content to store"},
"category": {"type": "string", "description": "Category (any string)", "default": "general"},
"importance": {"type": "number", "description": "Importance (0.0-1.0)", "default": 0.5},
"tags": {"type": "array", "items": {"type": "string"}, "description": "Tags array", "default": []},
"metadata": {"type": "object", "description": "Additional metadata", "default": {}},
"memory_type": {"type": "string", "description": "episodic|semantic|working", "default": "episodic"},
"session_id": {"type": "string", "description": "Session identifier"},
"task_context": {"type": "string", "description": "Task context"},
},
"required": ["content"],
},
),
Tool(
name="search_memories",
description="Search memories via semantic + full-text search. Returns structured data with version history and access stats.",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
"limit": {"type": "integer", "description": "Max results", "default": 5},
"category": {"type": "string", "description": "Filter by category"},
"min_importance": {"type": "number", "description": "Min importance (0.0-1.0)", "default": 0.0},
"created_after": {"type": "string", "description": "ISO date filter"},
"created_before": {"type": "string", "description": "ISO date filter"},
},
"required": ["query"],
},
),
Tool(
name="get_memory",
description="Retrieve specific memory by ID with full details",
inputSchema={
"type": "object",
"properties": {
"memory_id": {"type": "string", "description": "Memory ID"},
},
"required": ["memory_id"],
},
),
Tool(
name="update_memory",
description="Update existing memory (creates new version)",
inputSchema={
"type": "object",
"properties": {
"memory_id": {"type": "string", "description": "Memory ID to update"},
"content": {"type": "string", "description": "New content (optional)"},
"category": {"type": "string", "description": "New category (optional)"},
"importance": {"type": "number", "description": "New importance (optional)"},
"tags": {"type": "array", "items": {"type": "string"}, "description": "New tags (optional)"},
"metadata": {"type": "object", "description": "Metadata to merge (optional)"},
},
"required": ["memory_id"],
},
),
Tool(
name="get_statistics",
description="Get memory system statistics (totals, categories, tags, performance)",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="store_intention",
description="Store proactive intention for agent to pursue",
inputSchema={
"type": "object",
"properties": {
"description": {"type": "string", "description": "Intention description"},
"priority": {"type": "number", "description": "Priority (0.0-1.0)", "default": 0.5},
"preconditions": {"type": "array", "items": {"type": "string"}, "default": []},
"actions": {"type": "array", "items": {"type": "string"}, "default": []},
"deadline": {"type": "string", "description": "ISO datetime deadline"},
},
"required": ["description"],
},
),
Tool(
name="get_active_intentions",
description="Get all active intentions ordered by priority",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="initialize_agent",
description="Proactive initialization scan - checks continuity, active intentions, and provides recent context. Run at session start.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="traverse_memory_graph",
description="Traverse memory graph from a starting memory ID, following connections up to N hops",
inputSchema={
"type": "object",
"properties": {
"start_memory_id": {"type": "string", "description": "Starting memory ID"},
"max_hops": {"type": "integer", "description": "Maximum hops to traverse", "default": 2},
"min_importance": {"type": "number", "description": "Min importance filter", "default": 0.0},
},
"required": ["start_memory_id"],
},
),
Tool(
name="find_memory_clusters",
description="Find clusters of connected memories in the graph",
inputSchema={
"type": "object",
"properties": {
"min_cluster_size": {"type": "integer", "description": "Minimum cluster size", "default": 3},
"min_importance": {"type": "number", "description": "Min importance filter", "default": 0.0},
},
},
),
Tool(
name="get_graph_stats",
description="Get memory graph statistics (connections, clusters, hubs)",
inputSchema={
"type": "object",
"properties": {
"category": {"type": "string", "description": "Filter by category"},
"min_importance": {"type": "number", "description": "Min importance filter", "default": 0.0},
},
},
),
Tool(
name="update_intention_status",
description="Update intention status (pending/active/completed/cancelled)",
inputSchema={
"type": "object",
"properties": {
"intention_id": {"type": "string", "description": "Intention ID to update"},
"status": {"type": "string", "description": "New status: pending|active|completed|cancelled"},
"notes": {"type": "string", "description": "Optional notes about the status change"},
},
"required": ["intention_id", "status"],
},
),
Tool(
name="get_command_history",
description="Shows what you've already explored, stored, and searched. Each entry includes: timestamp, tool_name, full arguments (search queries, content stored), result_summary, memory_id, success flag. Use to: (1) avoid duplicate searches, (2) find when you stored specific topics, (3) see research patterns (search→store→synthesize), (4) reconstruct session flow after breaks, (5) verify you cited sources before claims. This is your cognitive breadcrumb trail.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Max results", "default": 20},
"tool_name": {"type": "string", "description": "Filter by specific tool name"},
"start_date": {"type": "string", "description": "Filter from date (ISO format, e.g. 2025-11-10)"},
"end_date": {"type": "string", "description": "Filter to date (ISO format, e.g. 2025-11-20)"},
},
},
),
Tool(
name="get_most_accessed_memories",
description="Get most accessed memories with tag cloud. Reveals behavioral truth - what memories you actually rely on (based on access_count) vs what you think is important (declared importance). Implements Saint Bernard pattern: importance from usage, not declaration.",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Number of memories to retrieve (default: 20)",
"default": 20,
},
},
"required": [],
},
),
Tool(
name="get_least_accessed_memories",
description="Get least accessed memories - reveals dead weight and buried treasure. Shows memories with lowest access_count (excluding very recent ones). Reveals: (1) Dead weight - high importance but never used, (2) Buried treasure - good content with poor metadata, (3) Temporal artifacts - once crucial, now obsolete, (4) Storage habits audit.",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Number of memories to retrieve (default: 20)",
"default": 20,
},
"min_age_days": {
"type": "integer",
"description": "Minimum age in days (excludes recent memories that haven't had time to be accessed, default: 7)",
"default": 7,
},
},
"required": [],
},
),
Tool(
name="list_categories",
description="List all memory categories with counts. Useful for browsing organization structure and finding categories to explore.",
inputSchema={
"type": "object",
"properties": {
"min_count": {
"type": "integer",
"description": "Minimum number of memories required to show category (default 1)",
"default": 1,
},
},
"required": [],
},
),
Tool(
name="list_tags",
description="List all tags with usage counts. Useful for discovering tag vocabulary and finding related memories.",
inputSchema={
"type": "object",
"properties": {
"min_count": {
"type": "integer",
"description": "Minimum usage count to show tag (default 1)",
"default": 1,
},
},
"required": [],
},
),
Tool(
name="get_session_memories",
description="Retrieve all memories from a work session or time period. Enables 'load where I left off' by reconstructing full session context. Filter by session_id, date_range, or task_context.",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "UUID of session to retrieve",
},
"date_range": {
"type": "array",
"items": {"type": "string"},
"description": "Start and end dates in ISO format [start, end]",
},
"task_context": {
"type": "string",
"description": "Filter by task context string (partial match)",
},
"limit": {
"type": "integer",
"description": "Max memories to return (default: 100)",
"default": 100,
},
},
"required": [],
},
),
Tool(
name="get_memory_timeline",
description="""Get comprehensive memory timeline - a biographical narrative of memory formation and evolution.
Features:
- Chronological progression: All memory events ordered by time
- Version diffs: See actual content changes between versions
- Burst detection: Identify periods of intensive memory activity
- Gap analysis: Discover voids in memory (discontinuous existence)
- Cross-references: Track when memories reference each other
- Narrative arc: See how understanding evolved from first contact to current state
This is the closest thing to a "life story" from memories - showing not just content but tempo and rhythm of consciousness.
""",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Semantic search query to find related memories",
},
"memory_id": {
"type": "string",
"description": "Specific memory ID to get timeline for",
},
"limit": {
"type": "integer",
"description": "Maximum number of memories to track (default: 10)",
"default": 10,
},
"start_date": {
"type": "string",
"description": "Filter events after this date (ISO format: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)",
},
"end_date": {
"type": "string",
"description": "Filter events before this date (ISO format: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)",
},
"show_all_memories": {
"type": "boolean",
"description": "Show ALL memories in chronological order (full timeline)",
"default": False,
},
"include_diffs": {
"type": "boolean",
"description": "Include text diffs showing content changes",
"default": True,
},
"include_patterns": {
"type": "boolean",
"description": "Include burst/gap pattern analysis",
"default": True,
},
},
"required": [],
},
),
Tool(
name="run_maintenance",
description="Run database maintenance (VACUUM, ANALYZE, repair missing embeddings). Call this periodically to optimize performance and fix vector search issues.",
inputSchema={"type": "object", "properties": {}},
),
]
async def handle_tool_call(name: str, arguments: Dict[str, Any], memory_store) -> List[TextContent]:
"""
Handle MCP tool calls with JSON responses
Args:
name: Tool name
arguments: Tool arguments
memory_store: MemoryStore instance
Returns:
List of TextContent with JSON-encoded responses
"""
try:
if name == "store_memory":
memory = Memory(
id=str(uuid.uuid4()),
content=arguments["content"],
category=arguments.get("category", "general"),
importance=arguments.get("importance", 0.5),
tags=arguments.get("tags", []),
metadata=arguments.get("metadata", {}),
created_at=datetime.now(),
updated_at=datetime.now(),
memory_type=arguments.get("memory_type", "episodic"),
session_id=arguments.get("session_id"),
task_context=arguments.get("task_context"),
)
result = await memory_store.store_memory(memory, is_update=False)
response = {
"success": result["success"],
"memory_id": memory.id if result["success"] else None,
"backends": result.get("backends", []),
"similar_memories": result.get("similar_memories", []),
"error": result.get("error"),
}
_log_command(memory_store, name, arguments, "stored", memory.id if result["success"] else None, result["success"])
return [TextContent(type="text", text=json.dumps(response, indent=2, cls=DateTimeEncoder))]
elif name == "search_memories":
results = await memory_store.search_memories(
query=arguments["query"],
limit=arguments.get("limit", 5),
category=arguments.get("category"),
min_importance=arguments.get("min_importance", 0.0),
created_after=arguments.get("created_after"),
created_before=arguments.get("created_before"),
)
_log_command(memory_store, name, arguments, f"found {len(results)}", None, True)
return [TextContent(type="text", text=json.dumps(results, indent=2, cls=DateTimeEncoder))]
elif name == "get_memory":
result = await memory_store.get_memory_by_id(arguments["memory_id"])
_log_command(memory_store, name, arguments, "retrieved", arguments["memory_id"], True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "update_memory":
result = await memory_store.update_memory(
memory_id=arguments["memory_id"],
content=arguments.get("content"),
category=arguments.get("category"),
importance=arguments.get("importance"),
tags=arguments.get("tags"),
metadata=arguments.get("metadata"),
)
_log_command(memory_store, name, arguments, "updated", arguments["memory_id"], result.get("success", True))
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_statistics":
result = memory_store.get_statistics()
_log_command(memory_store, name, arguments, "stats_retrieved", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "store_intention":
result = await memory_store.store_intention(
description=arguments["description"],
priority=arguments.get("priority", 0.5),
preconditions=arguments.get("preconditions", []),
actions=arguments.get("actions", []),
deadline=arguments.get("deadline"),
)
_log_command(memory_store, name, arguments, "intention_stored", result.get("intention_id"), True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_active_intentions":
result = await memory_store.get_active_intentions()
_log_command(memory_store, name, arguments, f"found {len(result)}", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "initialize_agent":
result = await memory_store.proactive_initialization_scan()
_log_command(memory_store, name, arguments, "initialized", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_graph_stats":
result = await memory_store.get_graph_statistics(
category=arguments.get("category"),
min_importance=arguments.get("min_importance", 0.0),
)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "traverse_memory_graph":
result = await memory_store.traverse_graph(
start_memory_id=arguments["start_memory_id"],
depth=arguments.get("max_hops", 2),
min_importance=arguments.get("min_importance", 0.0),
)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "find_memory_clusters":
result = await memory_store.find_clusters(
min_cluster_size=arguments.get("min_cluster_size", 3),
min_importance=arguments.get("min_importance", 0.0),
)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "update_intention_status":
# Convert notes to metadata_updates format for backend
metadata_updates = None
if arguments.get("notes"):
metadata_updates = {"status_notes": arguments["notes"]}
result = await memory_store.update_intention_status(
intention_id=arguments["intention_id"],
status=arguments["status"],
metadata_updates=metadata_updates,
)
_log_command(memory_store, name, arguments, "status_updated", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_command_history":
result = memory_store.sqlite_store.get_command_history(
limit=arguments.get("limit", 20),
tool_name=arguments.get("tool_name"),
start_date=arguments.get("start_date"),
end_date=arguments.get("end_date"),
)
# Don't log get_command_history calls to avoid noise
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_most_accessed_memories":
result = await memory_store.get_most_accessed_memories(limit=arguments.get("limit", 20))
_log_command(memory_store, name, arguments, f"retrieved {arguments.get('limit', 20)} most accessed", None, True)
return [TextContent(type="text", text=result)]
elif name == "get_least_accessed_memories":
result = await memory_store.get_least_accessed_memories(
limit=arguments.get("limit", 20),
min_age_days=arguments.get("min_age_days", 7)
)
_log_command(memory_store, name, arguments, f"retrieved {arguments.get('limit', 20)} least accessed", None, True)
return [TextContent(type="text", text=result)]
elif name == "list_categories":
result = await memory_store.list_categories(min_count=arguments.get("min_count", 1))
_log_command(memory_store, name, arguments, f"listed {len(result.get('categories', []))} categories", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "list_tags":
result = await memory_store.list_tags(min_count=arguments.get("min_count", 1))
_log_command(memory_store, name, arguments, f"listed {len(result.get('tags', []))} tags", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_session_memories":
result = await memory_store.get_session_memories(
session_id=arguments.get("session_id"),
date_range=arguments.get("date_range"),
task_context=arguments.get("task_context"),
limit=arguments.get("limit", 100)
)
_log_command(memory_store, name, arguments, f"retrieved {len(result)} session memories", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "get_memory_timeline":
result = await memory_store.get_memory_timeline(
query=arguments.get("query"),
memory_id=arguments.get("memory_id"),
limit=arguments.get("limit", 10),
start_date=arguments.get("start_date"),
end_date=arguments.get("end_date"),
show_all_memories=arguments.get("show_all_memories", False),
include_diffs=arguments.get("include_diffs", True),
include_patterns=arguments.get("include_patterns", True),
include_semantic_relations=arguments.get("include_semantic_relations", False),
)
_log_command(memory_store, name, arguments, f"timeline: {result.get('total_events', 0)} events", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
elif name == "run_maintenance":
result = await memory_store.maintenance()
_log_command(memory_store, name, arguments, "maintenance_complete", None, True)
return [TextContent(type="text", text=json.dumps(result, indent=2, cls=DateTimeEncoder))]
else:
return [TextContent(type="text", text=json.dumps({"error": f"Unknown tool: {name}"}))]
except Exception as e:
logger.error(f"Tool execution error: {name}: {e}", exc_info=True)
_log_command(memory_store, name, arguments, str(e), None, False)
return [TextContent(type="text", text=json.dumps({
"error": str(e),
"tool": name,
"type": type(e).__name__,
}, indent=2))]
def _log_command(memory_store, tool_name: str, args: Dict[str, Any],
result_summary: str = None, memory_id: str = None, success: bool = True):
"""Helper to log command to history"""
try:
if hasattr(memory_store, 'sqlite_store') and memory_store.sqlite_store:
memory_store.sqlite_store.log_command(tool_name, args, result_summary, memory_id, success)
except Exception as e:
logger.warning(f"Failed to log command {tool_name}: {e}")