search_by_tag
Search and retrieve memories stored in MCP Memory Service using specific tags to enhance information access and organization.
Instructions
Search memories by tags
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| tags | Yes |
Implementation Reference
- Defines the input schema for the MCP search_by_tag tool: requires 'tags' array, optional 'operation' (AND/OR, default AND).MCPTool( name="search_by_tag", description="Search memories by specific tags", inputSchema={ "type": "object", "properties": { "tags": {"type": "array", "items": {"type": "string"}, "description": "Tags to search for"}, "operation": {"type": "string", "enum": ["AND", "OR"], "description": "Tag search operation", "default": "AND"} }, "required": ["tags"] } ),
- Handler for MCP 'search_by_tag' tool call in web API: extracts tags and operation, invokes storage.search_by_tags, formats and returns results.elif tool_name == "search_by_tag": tags = arguments.get("tags") operation = arguments.get("operation", "AND") results = await storage.search_by_tags(tags=tags, operation=operation) return { "results": [ { "content": memory.content, "content_hash": memory.content_hash, "tags": memory.tags, "created_at": memory.created_at_iso } for memory in results ], "total_found": len(results) }
- src/mcp_memory_service/web/api/mcp.py:65-150 (registration)Registration of search_by_tag tool in MCP_TOOLS list, exposed via tools/list endpoint for MCP discovery.MCP_TOOLS = [ MCPTool( name="store_memory", description="Store a new memory with optional tags, metadata, and client information", inputSchema={ "type": "object", "properties": { "content": {"type": "string", "description": "The memory content to store"}, "tags": {"type": "array", "items": {"type": "string"}, "description": "Optional tags for the memory"}, "memory_type": {"type": "string", "description": "Optional memory type (e.g., 'note', 'reminder', 'fact')"}, "metadata": {"type": "object", "description": "Additional metadata for the memory"}, "client_hostname": {"type": "string", "description": "Client machine hostname for source tracking"} }, "required": ["content"] } ), MCPTool( name="retrieve_memory", description="Search and retrieve memories using semantic similarity", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query for finding relevant memories"}, "limit": {"type": "integer", "description": "Maximum number of memories to return", "default": 10}, "similarity_threshold": {"type": "number", "description": "Minimum similarity score threshold (0.0-1.0)", "default": 0.7, "minimum": 0.0, "maximum": 1.0} }, "required": ["query"] } ), MCPTool( name="recall_memory", description="Retrieve memories using natural language time expressions and optional semantic search", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Natural language query specifying the time frame or content to recall"}, "n_results": {"type": "integer", "description": "Maximum number of results to return", "default": 5} }, "required": ["query"] } ), MCPTool( name="search_by_tag", description="Search memories by specific tags", inputSchema={ "type": "object", "properties": { "tags": {"type": "array", "items": {"type": "string"}, "description": "Tags to search for"}, "operation": {"type": "string", "enum": ["AND", "OR"], "description": "Tag search operation", "default": "AND"} }, "required": ["tags"] } ), MCPTool( name="delete_memory", description="Delete a specific memory by content hash", inputSchema={ "type": "object", "properties": { "content_hash": {"type": "string", "description": "Hash of the memory to delete"} }, "required": ["content_hash"] } ), MCPTool( name="check_database_health", description="Check the health and status of the memory database", inputSchema={ "type": "object", "properties": {} } ), MCPTool( name="list_memories", description="List memories with pagination and optional filtering", inputSchema={ "type": "object", "properties": { "page": {"type": "integer", "description": "Page number (1-based)", "default": 1, "minimum": 1}, "page_size": {"type": "integer", "description": "Number of memories per page", "default": 10, "minimum": 1, "maximum": 100}, "tag": {"type": "string", "description": "Filter by specific tag"}, "memory_type": {"type": "string", "description": "Filter by memory type"} } } ), ]
- Native FastMCP tool handler for search_by_tag: delegates to MemoryService.search_by_tag with tags and match_all parameter.async def search_by_tag( tags: Union[str, List[str]], ctx: Context, match_all: bool = False ) -> Dict[str, Any]: """ Search memories by tags. Args: tags: Tag or list of tags to search for match_all: If True, memory must have ALL tags; if False, ANY tag Returns: Dictionary with matching memories """ # Delegate to shared MemoryService business logic memory_service = ctx.request_context.lifespan_context.memory_service return await memory_service.search_by_tag( tags=tags, match_all=match_all )
- Shared business logic helper for tag-based search: normalizes input tags, calls storage.search_by_tag, formats Memory objects into response.async def search_by_tag( self, tags: Union[str, List[str]], match_all: bool = False ) -> Union[SearchByTagSuccess, SearchByTagError]: """ Search memories by tags with flexible matching options. Args: tags: Tag or list of tags to search for match_all: If True, memory must have ALL tags; if False, ANY tag Returns: Dictionary with matching memories """ try: # Normalize tags to list (handles all formats including comma-separated) tags = normalize_tags(tags) # Search using database-level filtering # Note: Using search_by_tag from base class (singular) memories = await self.storage.search_by_tag(tags=tags) # Format results results = [] for memory in memories: results.append(self._format_memory_response(memory)) # Determine match type description match_type = "ALL" if match_all else "ANY" return { "memories": results, "tags": tags, "match_type": match_type, "count": len(results) } except Exception as e: logger.error(f"Error searching by tags: {e}") return { "memories": [], "tags": tags if isinstance(tags, list) else [tags], "error": f"Failed to search by tags: {str(e)}" }