"""Memory tool adapter for multi-provider support. This module provides a unified adapter for memory tools across different LLM providers. It handles provider detection, tool injection, and tool call execution with appropriate format conversions for each provider. Supported providers: - Anthropic: Native memory_20250818 tool and custom tools - OpenAI: Function calling format - Gemini: Function calling format - Generic: Fallback for unknown providers Usage: config = MemoryToolAdapterConfig(enabled=True) adapter = MemoryToolAdapter(config) # Detect provider from request provider = adapter.detect_provider(request_headers, model_name) # Inject tools tools, beta_headers = adapter.inject_tools(existing_tools, provider) # Handle tool calls in response if adapter.has_memory_tool_calls(response, provider): results = await adapter.handle_tool_calls(response, user_id, provider) """ from __future__ import annotations import json import logging from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal if TYPE_CHECKING: from headroom.memory.backends.local import LocalBackend logger = logging.getLogger(__name__) # ============================================================================= # Provider Types # ============================================================================= Provider = Literal["anthropic", "openai", "gemini", "generic"] # ============================================================================= # Tool Names # ============================================================================= # Custom memory tool names (Headroom's tools) MEMORY_TOOL_NAMES = {"memory_save", "memory_search", "memory_update", "memory_delete"} # Anthropic's native memory tool NATIVE_MEMORY_TOOL_NAME = "memory" NATIVE_MEMORY_TOOL_TYPE = "memory_20250818" # Beta header for Anthropic's native memory tool ANTHROPIC_BETA_HEADER = "context-management-2025-06-27" # ============================================================================= # Tool Schemas - Anthropic Native Tool # ============================================================================= ANTHROPIC_NATIVE_TOOL: dict[str, Any] = { "type": NATIVE_MEMORY_TOOL_TYPE, "name": NATIVE_MEMORY_TOOL_NAME, } # ============================================================================= # Tool Schemas - Anthropic Custom Tools # ============================================================================= ANTHROPIC_CUSTOM_TOOLS: list[dict[str, Any]] = [ { "name": "memory_save", "description": """Save important information to long-term memory for future reference. Use this tool when you encounter information that should be remembered across conversations: - User preferences (e.g., "prefers Python over JavaScript") - Personal facts (e.g., "works at Acme Corp", "has a dog named Max") - Project context (e.g., "working on a CLI tool", "using React 18") - Decisions made (e.g., "chose PostgreSQL for the database") - Important relationships (e.g., "Alice is Bob's manager") DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""", "input_schema": { "type": "object", "properties": { "content": { "type": "string", "description": "The information to remember. Be specific and self-contained.", }, "importance": { "type": "number", "minimum": 0.0, "maximum": 1.0, "description": "Importance score from 0.0 (low) to 1.0 (critical).", }, "facts": { "type": "array", "items": {"type": "string"}, "description": "Pre-extracted discrete facts for efficient storage.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Entity names referenced in this memory.", }, "extracted_entities": { "type": "array", "items": { "type": "object", "properties": { "entity": {"type": "string"}, "entity_type": {"type": "string"}, }, "required": ["entity", "entity_type"], }, "description": "Pre-extracted entities with types.", }, "extracted_relationships": { "type": "array", "items": { "type": "object", "properties": { "source": {"type": "string"}, "relationship": {"type": "string"}, "destination": {"type": "string"}, }, "required": ["source", "relationship", "destination"], }, "description": "Pre-extracted relationships for graph storage.", }, }, "required": ["content", "importance"], }, }, { "name": "memory_search", "description": """Search stored memories to recall relevant information. Use this tool to retrieve previously saved information before responding to questions about: - User preferences or past decisions - Personal or professional context - Previously discussed topics or projects - Relationships between people, systems, or concepts Search BEFORE saving to avoid duplicates.""", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "Natural language search query.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Filter to memories mentioning these entities.", }, "include_related": { "type": "boolean", "description": "Also retrieve connected memories.", }, "top_k": { "type": "integer", "minimum": 1, "maximum": 50, "description": "Maximum number of memories to retrieve (default 10).", }, }, "required": ["query"], }, }, { "name": "memory_update", "description": """Update an existing memory with corrected or evolved information. Use when: - User provides a correction to stored information - Information has changed over time - Adding detail or clarification to an existing memory""", "input_schema": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to update.", }, "new_content": { "type": "string", "description": "The updated content.", }, "reason": { "type": "string", "description": "Explanation for the update.", }, }, "required": ["memory_id", "new_content"], }, }, { "name": "memory_delete", "description": """Delete a memory that is no longer relevant or was stored in error. Use when: - User explicitly asks to forget something - Information is outdated and no longer applicable - A memory was saved in error""", "input_schema": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to delete.", }, "reason": { "type": "string", "description": "Explanation for the deletion.", }, }, "required": ["memory_id"], }, }, ] # ============================================================================= # Tool Schemas - OpenAI Function Calling Format # ============================================================================= OPENAI_TOOLS: list[dict[str, Any]] = [ { "type": "function", "function": { "name": "memory_save", "description": """Save important information to long-term memory for future reference. Use this tool when you encounter information that should be remembered across conversations: - User preferences, personal facts, project context, decisions, relationships DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""", "parameters": { "type": "object", "properties": { "content": { "type": "string", "description": "The information to remember. Be specific and self-contained.", }, "importance": { "type": "number", "minimum": 0.0, "maximum": 1.0, "description": "Importance score from 0.0 (low) to 1.0 (critical).", }, "facts": { "type": "array", "items": {"type": "string"}, "description": "Pre-extracted discrete facts.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Entity names referenced in this memory.", }, "extracted_entities": { "type": "array", "items": { "type": "object", "properties": { "entity": {"type": "string"}, "entity_type": {"type": "string"}, }, "required": ["entity", "entity_type"], }, "description": "Pre-extracted entities with types.", }, "extracted_relationships": { "type": "array", "items": { "type": "object", "properties": { "source": {"type": "string"}, "relationship": {"type": "string"}, "destination": {"type": "string"}, }, "required": ["source", "relationship", "destination"], }, "description": "Pre-extracted relationships.", }, }, "required": ["content", "importance"], }, }, }, { "type": "function", "function": { "name": "memory_search", "description": "Search stored memories to recall relevant information.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Natural language search query.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Filter to memories mentioning these entities.", }, "include_related": { "type": "boolean", "description": "Also retrieve connected memories.", }, "top_k": { "type": "integer", "minimum": 1, "maximum": 50, "description": "Maximum number of memories to retrieve.", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "memory_update", "description": "Update an existing memory with corrected or evolved information.", "parameters": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to update.", }, "new_content": { "type": "string", "description": "The updated content.", }, "reason": { "type": "string", "description": "Explanation for the update.", }, }, "required": ["memory_id", "new_content"], }, }, }, { "type": "function", "function": { "name": "memory_delete", "description": "Delete a memory that is no longer relevant.", "parameters": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to delete.", }, "reason": { "type": "string", "description": "Explanation for the deletion.", }, }, "required": ["memory_id"], }, }, }, ] # ============================================================================= # Tool Schemas - Gemini Function Calling Format # ============================================================================= # Gemini uses a similar format to OpenAI but with slight differences GEMINI_TOOLS: list[dict[str, Any]] = [ { "name": "memory_save", "description": """Save important information to long-term memory for future reference. Use this tool when you encounter information that should be remembered across conversations: - User preferences, personal facts, project context, decisions, relationships DO NOT save: transient info, sensitive data (passwords, keys), redundant info.""", "parameters": { "type": "object", "properties": { "content": { "type": "string", "description": "The information to remember. Be specific and self-contained.", }, "importance": { "type": "number", "description": "Importance score from 0.0 (low) to 1.0 (critical).", }, "facts": { "type": "array", "items": {"type": "string"}, "description": "Pre-extracted discrete facts.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Entity names referenced in this memory.", }, }, "required": ["content", "importance"], }, }, { "name": "memory_search", "description": "Search stored memories to recall relevant information.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Natural language search query.", }, "entities": { "type": "array", "items": {"type": "string"}, "description": "Filter to memories mentioning these entities.", }, "include_related": { "type": "boolean", "description": "Also retrieve connected memories.", }, "top_k": { "type": "integer", "description": "Maximum number of memories to retrieve.", }, }, "required": ["query"], }, }, { "name": "memory_update", "description": "Update an existing memory with corrected or evolved information.", "parameters": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to update.", }, "new_content": { "type": "string", "description": "The updated content.", }, "reason": { "type": "string", "description": "Explanation for the update.", }, }, "required": ["memory_id", "new_content"], }, }, { "name": "memory_delete", "description": "Delete a memory that is no longer relevant.", "parameters": { "type": "object", "properties": { "memory_id": { "type": "string", "description": "The unique ID of the memory to delete.", }, "reason": { "type": "string", "description": "Explanation for the deletion.", }, }, "required": ["memory_id"], }, }, ] # ============================================================================= # Configuration # ============================================================================= @dataclass class MemoryToolAdapterConfig: """Configuration for the memory tool adapter. Attributes: enabled: Whether memory features are enabled. use_native_tool: Use Anthropic's native memory_20250818 tool (Anthropic only). inject_tools: Whether to inject memory tools into requests. inject_context: Whether to inject memory context into requests. db_path: Path to the local memory database. top_k: Number of memories to retrieve in searches. min_similarity: Minimum similarity score for memory retrieval. """ enabled: bool = False use_native_tool: bool = True # Default to native for Anthropic (subscription-safe) inject_tools: bool = True inject_context: bool = True db_path: str = "headroom_memory.db" top_k: int = 10 min_similarity: float = 0.3 # ============================================================================= # Memory Tool Adapter # ============================================================================= class MemoryToolAdapter: """Adapter for memory tools across different LLM providers. This adapter provides a unified interface for: 1. Detecting the LLM provider from requests 2. Injecting memory tools in provider-specific formats 3. Providing required beta headers 4. Detecting memory tool calls in responses 5. Handling tool calls with the semantic backend Example: adapter = MemoryToolAdapter(config) provider = adapter.detect_provider(headers, model) tools, headers = adapter.inject_tools(existing_tools, provider) # Later, when processing response if adapter.has_memory_tool_calls(response, provider): results = await adapter.handle_tool_calls(response, user_id, provider) """ def __init__(self, config: MemoryToolAdapterConfig) -> None: """Initialize the adapter. Args: config: Configuration for the adapter. """ self.config = config self._backend: LocalBackend | Any = None self._initialized = False async def _ensure_initialized(self) -> None: """Lazy initialization of the semantic backend. Imports and initializes the LocalBackend from memory_handler to provide semantic search and storage capabilities. """ if self._initialized: return if not self.config.enabled: return from headroom.memory.backends.local import LocalBackend, LocalBackendConfig backend_config = LocalBackendConfig(db_path=self.config.db_path) self._backend = LocalBackend(backend_config) await self._backend._ensure_initialized() self._initialized = True logger.info(f"MemoryToolAdapter: Initialized backend at {self.config.db_path}") def detect_provider( self, request_headers: dict[str, str] | None = None, model_name: str | None = None, ) -> Provider: """Detect the LLM provider from request headers and model name. Detection priority: 1. Explicit headers (x-api-key for Anthropic, authorization for OpenAI) 2. Model name patterns (claude-*, gpt-*, gemini-*) 3. Fallback to generic Args: request_headers: HTTP headers from the request (optional). model_name: Name of the model being used (optional). Returns: The detected provider. """ headers = request_headers or {} model = (model_name or "").lower() # Check headers for provider hints if "x-api-key" in headers or "anthropic-version" in headers: return "anthropic" if headers.get("authorization", "").startswith("Bearer sk-"): # OpenAI uses sk-* API keys return "openai" # Check model name patterns if model.startswith("claude"): return "anthropic" if model.startswith("gpt") or model.startswith("o1") or model.startswith("o3"): return "openai" if model.startswith("gemini") or "gemma" in model: return "gemini" # Fallback to generic return "generic" def inject_tools( self, tools: list[dict[str, Any]] | None, provider: Provider, ) -> tuple[list[dict[str, Any]], dict[str, str]]: """Inject memory tools into the tools list for the given provider. Args: tools: Existing tools list (may be None). provider: The LLM provider to format tools for. Returns: Tuple of (updated_tools, beta_headers). beta_headers contains any required headers (e.g., anthropic-beta). """ if not self.config.inject_tools: return tools or [], {} tools = list(tools) if tools else [] beta_headers: dict[str, str] = {} # Get existing tool names existing_names = self._get_existing_tool_names(tools) # Handle Anthropic native tool if provider == "anthropic" and self.config.use_native_tool: if NATIVE_MEMORY_TOOL_NAME not in existing_names: tools.append(ANTHROPIC_NATIVE_TOOL.copy()) beta_headers["anthropic-beta"] = ANTHROPIC_BETA_HEADER logger.info("MemoryToolAdapter: Injected native memory tool for Anthropic") return tools, beta_headers # Handle custom tools by provider if provider == "anthropic": tools, was_injected = self._inject_anthropic_tools(tools, existing_names) elif provider == "openai": tools, was_injected = self._inject_openai_tools(tools, existing_names) elif provider == "gemini": tools, was_injected = self._inject_gemini_tools(tools, existing_names) else: # Generic fallback uses OpenAI format tools, was_injected = self._inject_openai_tools(tools, existing_names) if was_injected: logger.info(f"MemoryToolAdapter: Injected custom tools for {provider}") return tools, beta_headers def _get_existing_tool_names(self, tools: list[dict[str, Any]]) -> set[str]: """Extract tool names from existing tools list.""" names: set[str] = set() for tool in tools: # Anthropic format if "name" in tool: names.add(tool["name"]) # OpenAI format if "function" in tool and "name" in tool["function"]: names.add(tool["function"]["name"]) return names def _inject_anthropic_tools( self, tools: list[dict[str, Any]], existing_names: set[str], ) -> tuple[list[dict[str, Any]], bool]: """Inject Anthropic-formatted custom memory tools.""" was_injected = False for memory_tool in ANTHROPIC_CUSTOM_TOOLS: if memory_tool["name"] not in existing_names: tools.append(memory_tool.copy()) was_injected = True return tools, was_injected def _inject_openai_tools( self, tools: list[dict[str, Any]], existing_names: set[str], ) -> tuple[list[dict[str, Any]], bool]: """Inject OpenAI-formatted memory tools.""" was_injected = False for memory_tool in OPENAI_TOOLS: tool_name = memory_tool["function"]["name"] if tool_name not in existing_names: tools.append(memory_tool.copy()) was_injected = True return tools, was_injected def _inject_gemini_tools( self, tools: list[dict[str, Any]], existing_names: set[str], ) -> tuple[list[dict[str, Any]], bool]: """Inject Gemini-formatted memory tools.""" was_injected = False for memory_tool in GEMINI_TOOLS: if memory_tool["name"] not in existing_names: tools.append(memory_tool.copy()) was_injected = True return tools, was_injected def get_beta_headers(self, provider: Provider) -> dict[str, str]: """Get any required beta headers for the provider. Args: provider: The LLM provider. Returns: Dict of header name -> value for any required beta headers. """ if provider == "anthropic" and self.config.use_native_tool: return {"anthropic-beta": ANTHROPIC_BETA_HEADER} return {} def has_memory_tool_calls( self, response: dict[str, Any], provider: Provider, ) -> bool: """Check if the response contains memory tool calls. Args: response: The API response from the LLM. provider: The LLM provider. Returns: True if response contains memory tool calls. """ tool_calls = self._extract_tool_calls(response, provider) for tc in tool_calls: name = self._get_tool_name(tc, provider) if name in MEMORY_TOOL_NAMES or name == NATIVE_MEMORY_TOOL_NAME: return True return False def _extract_tool_calls( self, response: dict[str, Any], provider: Provider, ) -> list[dict[str, Any]]: """Extract tool calls from response based on provider format.""" if provider == "anthropic": content = response.get("content", []) if isinstance(content, list): return [block for block in content if block.get("type") == "tool_use"] return [] elif provider == "openai": choices = response.get("choices", []) if choices: message = choices[0].get("message", {}) return list(message.get("tool_calls", []) or []) return [] elif provider == "gemini": # Gemini format: candidates[0].content.parts[*].functionCall candidates = response.get("candidates", []) if candidates: content = candidates[0].get("content", {}) parts = content.get("parts", []) return [p for p in parts if "functionCall" in p] return [] # Generic fallback - try both formats tool_calls = [] # Try Anthropic format content = response.get("content", []) if isinstance(content, list): tool_calls.extend([block for block in content if block.get("type") == "tool_use"]) # Try OpenAI format choices = response.get("choices", []) if choices: message = choices[0].get("message", {}) tool_calls.extend(list(message.get("tool_calls", []) or [])) return tool_calls def _get_tool_name(self, tool_call: dict[str, Any], provider: Provider) -> str: """Get the tool name from a tool call.""" if provider == "anthropic": return str(tool_call.get("name", "")) elif provider == "openai": return str(tool_call.get("function", {}).get("name", "")) elif provider == "gemini": func_call = tool_call.get("functionCall", {}) return str(func_call.get("name", "")) else: # Generic - try both return str(tool_call.get("name", "") or tool_call.get("function", {}).get("name", "")) def _get_tool_id(self, tool_call: dict[str, Any], provider: Provider) -> str: """Get the tool call ID.""" if provider == "anthropic": return str(tool_call.get("id", "")) elif provider == "openai": return str(tool_call.get("id", "")) elif provider == "gemini": # Gemini doesn't use IDs in the same way return str(tool_call.get("functionCall", {}).get("name", "")) else: return str(tool_call.get("id", "")) def _get_tool_input( self, tool_call: dict[str, Any], provider: Provider, ) -> dict[str, Any]: """Get the tool input/arguments from a tool call.""" if provider == "anthropic": result = tool_call.get("input", {}) return dict(result) if isinstance(result, dict) else {} elif provider == "openai": args_str = tool_call.get("function", {}).get("arguments", "{}") try: parsed = json.loads(args_str) return dict(parsed) if isinstance(parsed, dict) else {} except json.JSONDecodeError: return {} elif provider == "gemini": result = tool_call.get("functionCall", {}).get("args", {}) return dict(result) if isinstance(result, dict) else {} else: # Generic - try both if "input" in tool_call: result = tool_call["input"] return dict(result) if isinstance(result, dict) else {} args_str = tool_call.get("function", {}).get("arguments", "{}") try: parsed = json.loads(args_str) return dict(parsed) if isinstance(parsed, dict) else {} except json.JSONDecodeError: return {} async def handle_tool_calls( self, response: dict[str, Any], user_id: str, provider: Provider, ) -> list[dict[str, Any]]: """Handle memory tool calls and return results in provider format. Args: response: The API response containing tool calls. user_id: User identifier for memory operations. provider: The LLM provider. Returns: List of tool results in provider-appropriate format. """ await self._ensure_initialized() tool_calls = self._extract_tool_calls(response, provider) results: list[dict[str, Any]] = [] for tc in tool_calls: tool_name = self._get_tool_name(tc, provider) tool_id = self._get_tool_id(tc, provider) input_data = self._get_tool_input(tc, provider) # Skip non-memory tools if tool_name not in MEMORY_TOOL_NAMES and tool_name != NATIVE_MEMORY_TOOL_NAME: continue # Execute the tool if tool_name == NATIVE_MEMORY_TOOL_NAME: result_content = await self._execute_native_tool(input_data, user_id) else: result_content = await self._execute_custom_tool(tool_name, input_data, user_id) # Format result for provider result = self._format_tool_result(tool_id, result_content, provider) results.append(result) logger.info(f"MemoryToolAdapter: Executed {tool_name} for user {user_id}") return results def _format_tool_result( self, tool_id: str, content: str, provider: Provider, ) -> dict[str, Any]: """Format a tool result for the given provider.""" if provider == "anthropic": return { "type": "tool_result", "tool_use_id": tool_id, "content": content, } elif provider == "openai": return { "role": "tool", "tool_call_id": tool_id, "content": content, } elif provider == "gemini": return { "functionResponse": { "name": tool_id, "response": {"result": content}, } } else: # Generic uses OpenAI format return { "role": "tool", "tool_call_id": tool_id, "content": content, } async def _execute_native_tool( self, input_data: dict[str, Any], user_id: str, ) -> str: """Execute Anthropic's native memory tool. This translates native memory commands to our semantic backend: - view: semantic search or list memories - create: save to vector store - str_replace: update memory - delete: remove from vector store """ if not self._backend: return "Error: Memory backend not initialized" command = input_data.get("command", "") try: if command == "view": return await self._native_view(input_data, user_id) elif command == "create": return await self._native_create(input_data, user_id) elif command == "str_replace": return await self._native_update(input_data, user_id) elif command == "delete": return await self._native_delete(input_data, user_id) else: return f"Error: Unknown command '{command}'" except Exception as e: logger.error(f"MemoryToolAdapter: Native tool error: {e}") return f"Error: {e}" async def _native_view(self, input_data: dict[str, Any], user_id: str) -> str: """Handle VIEW command - semantic search or list memories.""" path = input_data.get("path", "/memories") # Normalize path if path.startswith("/memories"): subpath = path[len("/memories") :].lstrip("/") else: subpath = path.lstrip("/") # Search pattern: /memories/search/ if subpath.startswith("search/"): query = subpath[len("search/") :] if not query: return "Error: Please provide a search query" return await self._semantic_search(query, user_id) # Recent: /memories/recent if subpath == "recent": return await self._semantic_search("recent memories", user_id, top_k=10) # Root: /memories if not subpath: return await self._get_memory_overview(user_id) # Treat path as search topic return await self._semantic_search( subpath.replace("/", " ").replace("_", " "), user_id, ) async def _native_create(self, input_data: dict[str, Any], user_id: str) -> str: """Handle CREATE command - save to vector store.""" path = input_data.get("path", "") file_text = input_data.get("file_text", "") if not file_text: return "Error: file_text is required" topic = path.replace("/memories/", "").replace("/", "_").replace(".txt", "") memory = await self._backend.save_memory( content=file_text, user_id=user_id, importance=0.5, metadata={"virtual_path": path, "topic": topic}, ) logger.info(f"MemoryToolAdapter: Created memory {memory.id} for {user_id}") return f"File created successfully at: {path}" async def _native_update(self, input_data: dict[str, Any], user_id: str) -> str: """Handle STR_REPLACE command - update memory content.""" old_str = input_data.get("old_str", "") new_str = input_data.get("new_str", "") if not old_str: return "Error: old_str is required" # Search for memory containing old_str results = await self._backend.search_memories( query=old_str, user_id=user_id, top_k=5, ) # Find exact match matching_memory = None for r in results: if old_str in r.memory.content: matching_memory = r.memory break if not matching_memory: return "No replacement performed, old_str not found in memories" # Perform replacement new_content = matching_memory.content.replace(old_str, new_str, 1) if hasattr(self._backend, "update_memory"): await self._backend.update_memory( memory_id=matching_memory.id, new_content=new_content, user_id=user_id, ) else: await self._backend.delete_memory(matching_memory.id) await self._backend.save_memory( content=new_content, user_id=user_id, importance=0.5, ) return "The memory has been edited." async def _native_delete(self, input_data: dict[str, Any], user_id: str) -> str: """Handle DELETE command - remove from vector store.""" path = input_data.get("path", "") topic = path.replace("/memories/", "").replace("/", " ").replace("_", " ") results = await self._backend.search_memories( query=topic, user_id=user_id, top_k=10, ) if not results: return f"Error: The path {path} does not exist" deleted_count = 0 for r in results: metadata = getattr(r.memory, "metadata", {}) or {} if metadata.get("virtual_path") == path or r.score > 0.8: await self._backend.delete_memory(r.memory.id) deleted_count += 1 if deleted_count == 0: return f"Error: The path {path} does not exist" return f"Successfully deleted {path}" async def _semantic_search( self, query: str, user_id: str, top_k: int = 5, ) -> str: """Perform semantic search and format results.""" results = await self._backend.search_memories( query=query, user_id=user_id, top_k=top_k, include_related=True, ) if not results: return f"No memories found matching '{query}'" lines = [f"Found {len(results)} memories matching '{query}':\n"] for i, r in enumerate(results, 1): score_pct = int(r.score * 100) content_preview = r.memory.content[:200] if len(r.memory.content) > 200: content_preview += "..." lines.append(f"{i}. [{score_pct}% match] {content_preview}") return "\n".join(lines) async def _get_memory_overview(self, user_id: str) -> str: """Get memory overview with search instructions.""" results = await self._backend.search_memories( query="*", user_id=user_id, top_k=100, ) count = len(results) if results else 0 return f"""Memory System ({count} memories stored) To SEARCH: view /memories/search/ To see RECENT: view /memories/recent To SAVE: create /memories/.txt "content" """ async def _execute_custom_tool( self, tool_name: str, input_data: dict[str, Any], user_id: str, ) -> str: """Execute a custom memory tool.""" if not self._backend: return json.dumps({"error": "Memory backend not initialized"}) try: if tool_name == "memory_save": return await self._execute_save(input_data, user_id) elif tool_name == "memory_search": return await self._execute_search(input_data, user_id) elif tool_name == "memory_update": return await self._execute_update(input_data, user_id) elif tool_name == "memory_delete": return await self._execute_delete(input_data, user_id) else: return json.dumps({"error": f"Unknown tool: {tool_name}"}) except Exception as e: logger.error(f"MemoryToolAdapter: Tool {tool_name} failed: {e}") return json.dumps({"status": "error", "error": str(e)}) async def _execute_save(self, input_data: dict[str, Any], user_id: str) -> str: """Execute memory_save tool.""" content = input_data.get("content", "") if not content: return json.dumps({"status": "error", "error": "content is required"}) importance = input_data.get("importance", 0.5) facts = input_data.get("facts") entities = input_data.get("entities") extracted_entities = input_data.get("extracted_entities") extracted_relationships = input_data.get("extracted_relationships") memory = await self._backend.save_memory( content=content, user_id=user_id, importance=importance, facts=facts, entities=entities, extracted_entities=extracted_entities, relationships=extracted_relationships, extracted_relationships=extracted_relationships, ) return json.dumps( { "status": "saved", "memory_id": memory.id, "content": memory.content[:100] + "..." if len(memory.content) > 100 else memory.content, } ) async def _execute_search(self, input_data: dict[str, Any], user_id: str) -> str: """Execute memory_search tool.""" query = input_data.get("query", "") if not query: return json.dumps({"status": "error", "error": "query is required"}) top_k = input_data.get("top_k", self.config.top_k) include_related = input_data.get("include_related", True) entities_filter = input_data.get("entities") results = await self._backend.search_memories( query=query, user_id=user_id, top_k=top_k, include_related=include_related, entities=entities_filter, ) return json.dumps( { "status": "found", "count": len(results), "memories": [ { "id": r.memory.id, "content": r.memory.content, "score": round(r.score, 3), "entities": ( r.related_entities[:5] if hasattr(r, "related_entities") and r.related_entities else [] ), } for r in results ], } ) async def _execute_update(self, input_data: dict[str, Any], user_id: str) -> str: """Execute memory_update tool.""" memory_id = input_data.get("memory_id", "") new_content = input_data.get("new_content", "") if not memory_id: return json.dumps({"status": "error", "error": "memory_id is required"}) if not new_content: return json.dumps({"status": "error", "error": "new_content is required"}) reason = input_data.get("reason") if hasattr(self._backend, "update_memory"): memory = await self._backend.update_memory( memory_id=memory_id, new_content=new_content, reason=reason, user_id=user_id, ) return json.dumps({"status": "updated", "memory_id": memory.id}) else: # Fallback: delete old, save new await self._backend.delete_memory(memory_id) memory = await self._backend.save_memory( content=new_content, user_id=user_id, importance=0.5, ) return json.dumps( { "status": "updated", "memory_id": memory.id, "note": "Replaced via delete+save", } ) async def _execute_delete(self, input_data: dict[str, Any], user_id: str) -> str: """Execute memory_delete tool.""" memory_id = input_data.get("memory_id", "") if not memory_id: return json.dumps({"status": "error", "error": "memory_id is required"}) deleted = await self._backend.delete_memory(memory_id) return json.dumps( { "status": "deleted" if deleted else "not_found", "memory_id": memory_id, } ) async def close(self) -> None: """Close the backend connection.""" if self._backend and hasattr(self._backend, "close"): await self._backend.close() self._backend = None self._initialized = False logger.info("MemoryToolAdapter: Closed")