retrieve_memory
Locate relevant stored information by querying the MCP Memory Service. Leverages ChromaDB and sentence transformers for accurate semantic search and retrieval.
Instructions
Find relevant memories based on query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| n_results | No | ||
| query | Yes |
Implementation Reference
- The primary MCP tool handler for 'retrieve_memory'. This function is decorated with @mcp.tool(), automatically registering it as an MCP tool. It accepts a semantic query and optional n_results parameter, delegating the core logic to MemoryService.retrieve_memories().async def retrieve_memory( query: str, ctx: Context, n_results: int = 5 ) -> Dict[str, Any]: """ Retrieve memories based on semantic similarity to a query. Args: query: Search query for semantic similarity n_results: Maximum number of results to return Returns: Dictionary with retrieved memories and metadata """ # Delegate to shared MemoryService business logic memory_service = ctx.request_context.lifespan_context.memory_service return await memory_service.retrieve_memories( query=query, n_results=n_results )
- src/mcp_memory_service/mcp_server.py:358-358 (registration)The @mcp.tool() decorator registers the retrieve_memory function as an available MCP tool.async def retrieve_memory(
- Core helper method in MemoryService that implements the semantic retrieval logic. Calls the storage backend's retrieve() method, applies optional filters, formats results with relevance scores, and handles errors.async def retrieve_memories( self, query: str, n_results: int = 10, tags: Optional[List[str]] = None, memory_type: Optional[str] = None ) -> Union[RetrieveMemoriesSuccess, RetrieveMemoriesError]: """ Retrieve memories by semantic search with optional filtering. Args: query: Search query string n_results: Maximum number of results tags: Optional tag filtering memory_type: Optional memory type filtering Returns: Dictionary with search results """ try: # Retrieve memories using semantic search # Note: storage.retrieve() only supports query and n_results # We'll filter by tags/type after retrieval if needed memories = await self.storage.retrieve( query=query, n_results=n_results ) # Apply optional post-filtering filtered_memories = memories if tags or memory_type: filtered_memories = [] for memory in memories: # Filter by tags if specified if tags: memory_tags = memory.metadata.get('tags', []) if hasattr(memory, 'metadata') else [] if not any(tag in memory_tags for tag in tags): continue # Filter by memory_type if specified if memory_type: mem_type = memory.metadata.get('memory_type', '') if hasattr(memory, 'metadata') else '' if mem_type != memory_type: continue filtered_memories.append(memory) results = [] for result in filtered_memories: # Extract Memory object from MemoryQueryResult and add similarity score memory_dict = self._format_memory_response(result.memory) memory_dict['similarity_score'] = result.relevance_score results.append(memory_dict) return { "memories": results, "query": query, "count": len(results) } except Exception as e: logger.error(f"Error retrieving memories: {e}") return { "memories": [], "query": query, "error": f"Failed to retrieve memories: {str(e)}" }
- Input schema defined by function parameters: required 'query' (str), optional 'n_results' (int, default 5). Output is Dict[str, Any] containing memories list with metadata and scores.query: str, ctx: Context, n_results: int = 5 ) -> Dict[str, Any]: """ Retrieve memories based on semantic similarity to a query. Args: query: Search query for semantic similarity n_results: Maximum number of results to return Returns: Dictionary with retrieved memories and metadata """ # Delegate to shared MemoryService business logic memory_service = ctx.request_context.lifespan_context.memory_service return await memory_service.retrieve_memories( query=query, n_results=n_results ) @mcp.tool() async def search_by_tag( tags: Union[str, List[str]], ctx: Context,