Knowledge Graph Memory Server

Instructions

Implementation Reference

• optimized_memory_mcp_server/storage/sqlite/operations/search_ops.py:12-44 (handler)

  The actual implementation of the search_nodes logic, which queries the database for entities matching the query and retrieves their related nodes.

  async def search_nodes(self, query: str) -> Dict[str, List[Dict[str, Any]]]:
      """Search for nodes and their relations."""
      if not query:
          raise ValueError("Search query cannot be empty")
  
      async with self.pool.get_connection() as conn:
          search_pattern = f"%{sanitize_input(query)}%"
          
          # Search entities
          cursor = await conn.execute(
              """
              SELECT * FROM entities 
              WHERE name LIKE ? 
              OR entity_type LIKE ? 
              OR observations LIKE ?
              """,
              (search_pattern, search_pattern, search_pattern)
          )
          rows = await cursor.fetchall()
          
          entities = []
          entity_names = set()
          for row in rows:
              entity = Entity(
                  name=row['name'],
                  entityType=row['entity_type'],
                  observations=row['observations'].split(',') if row['observations'] else []
              )
              entities.append(entity.to_dict())
              entity_names.add(entity.name)
  
          relations = await self._get_relations_for_entities(conn, entity_names)
          return {"entities": entities, "relations": relations}