MCP Memory Service

Instructions

Implementation Reference

• src/mcp_memory_service/mcp_server.py:358-379 (handler)

  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(

• src/mcp_memory_service/services/memory_service.py:379-446 (helper)

  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)}"
          }

• src/mcp_memory_service/mcp_server.py:359-383 (schema)

  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,