add list_by_type tool

src/basic_memory/api/routers/discovery_router.py

@@ -1,10 +1,12 @@
 """Router for knowledge discovery and analytics operations."""
 
+from typing import Optional
+
 from fastapi import APIRouter
 from loguru import logger
 
 from basic_memory.deps import EntityServiceDep, ObservationServiceDep
-from basic_memory.schemas import EntityTypeList, ObservationCategoryList
+from basic_memory.schemas import EntityTypeList, ObservationCategoryList, TypedEntityList, EntityResponse
 
 router = APIRouter(prefix="/discovery", tags=["discovery"])
 
@@ -23,3 +25,26 @@ async def get_observation_categories(observation_service: ObservationServiceDep)
     logger.debug("Getting all observation categories")
     categories = await observation_service.observation_categories()
     return ObservationCategoryList(categories=categories)
+
+
+@router.get("/entities/{entity_type}", response_model=TypedEntityList)
+async def list_entities_by_type(
+    entity_service: EntityServiceDep,
+    entity_type: str,
+    include_related: bool = False,
+    sort_by: Optional[str] = "updated_at",
+) -> TypedEntityList:
+    """List all entities of a specific type."""
+    logger.debug(f"Listing entities of type: {entity_type}")
+    entities = await entity_service.list_entities(
+        entity_type=entity_type,
+        sort_by=sort_by,
+        include_related=include_related
+    )
+    return TypedEntityList(
+        entity_type=entity_type,
+        entities=[EntityResponse.model_validate(e) for e in entities],
+        total=len(entities),
+        sort_by=sort_by,
+        include_related=include_related
+    )

src/basic_memory/mcp/tools/discovery.py

@@ -1,10 +1,10 @@
 """Tools for discovering and analyzing knowledge graph structure."""
 
-from typing import List
+from typing import List, Optional
 
 from loguru import logger
 
-from basic_memory.schemas import EntityTypeList, ObservationCategoryList
+from basic_memory.schemas import EntityTypeList, ObservationCategoryList, TypedEntityList
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 
@@ -55,3 +55,33 @@ async def get_observation_categories() -> List[str]:
     url = "/discovery/observation-categories"
     response = await client.get(url)
     return ObservationCategoryList.model_validate(response.json())
+
+
+@mcp.tool()
+async def list_by_type(
+        entity_type: str,
+        include_related: bool = False,
+        sort_by: Optional[str] = "updated_at"
+) -> TypedEntityList:
+    """List all entities of a specific type.
+
+    Example:
+        # Get all features
+        features = await list_by_type("feature")
+
+        # Get components with relations
+        components = await list_by_type(
+            "component",
+            include_related=True
+        )
+    """
+    logger.debug(f"Listing entities of type: {entity_type}")
+    params = {
+        "include_related": "true" if include_related else "false"
+    }
+    if sort_by:
+        params["sort_by"] = sort_by
+
+    url = f"/discovery/entities/{entity_type}"
+    response = await client.get(url, params=params)
+    return TypedEntityList.model_validate(response.json())

src/basic_memory/repository/entity_repository.py

@@ -2,7 +2,7 @@
 
 from typing import List, Optional, Sequence
 
-from sqlalchemy import select, or_
+from sqlalchemy import select, or_, asc, desc
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from sqlalchemy.orm import selectinload
 from sqlalchemy.orm.interfaces import LoaderOption
@@ -27,15 +27,22 @@ class EntityRepository(Repository[Entity]):
         self,
         entity_type: Optional[str] = None,
         doc_id: Optional[int] = None,
+        sort_by: Optional[str] = "updated_at",
     ) -> Sequence[Entity]:
-        """List all entities, optionally filtered by type."""
+        """List all entities, optionally filtered by type and sorted."""
         query = self.select().options(*self.get_load_options())
 
+        # Apply filters
         if entity_type:
             query = query.where(Entity.entity_type == entity_type)
         if doc_id:
             query = query.where(Entity.doc_id == doc_id)
 
+        # Apply sorting
+        if sort_by:
+            sort_field = getattr(Entity, sort_by, Entity.updated_at)
+            query = query.order_by(asc(sort_field))
+
         result = await self.execute_query(query)
         return list(result.scalars().all())
 

src/basic_memory/schemas/__init__.py

@@ -44,7 +44,7 @@ from basic_memory.schemas.response import (
 # Discovery and analytics models
 from basic_memory.schemas.discovery import (
     EntityTypeList,
-    ObservationCategoryList,
+    ObservationCategoryList, TypedEntityList,
 )
 
 # For convenient imports, export all models
@@ -76,4 +76,5 @@ __all__ = [
     # Discovery and Analytics
     "EntityTypeList",
     "ObservationCategoryList",
+    "TypedEntityList"
 ]

src/basic_memory/schemas/discovery.py

@@ -1,7 +1,9 @@
 """Schemas for knowledge discovery and analytics endpoints."""
 
-from typing import List
-from pydantic import BaseModel
+from typing import List, Optional
+from pydantic import BaseModel, Field
+
+from basic_memory.schemas.response import EntityResponse
 
 
 class EntityTypeList(BaseModel):
@@ -12,3 +14,12 @@ class EntityTypeList(BaseModel):
 class ObservationCategoryList(BaseModel):
     """List of unique observation categories in the system."""
     categories: List[str]
+
+
+class TypedEntityList(BaseModel):
+    """List of entities of a specific type."""
+    entity_type: str = Field(..., description="Type of entities in the list")
+    entities: List[EntityResponse]
+    total: int = Field(..., description="Total number of entities")
+    sort_by: Optional[str] = Field(None, description="Field used for sorting")
+    include_related: bool = Field(False, description="Whether related entities are included")

src/basic_memory/services/entity_service.py

@@ -1,6 +1,6 @@
 """Service for managing entities in the database."""
 
-from typing import Dict, Any, Sequence, List
+from typing import Dict, Any, Sequence, List, Optional
 
 from loguru import logger
 
@@ -72,6 +72,19 @@ class EntityService(BaseService[EntityRepository]):
         """Get list of all distinct entity types in the system."""
         logger.debug("Getting all distinct entity types")
         return await self.repository.get_entity_types()
+        
+    async def list_entities(
+        self, 
+        entity_type: Optional[str] = None,
+        sort_by: Optional[str] = "updated_at",
+        include_related: bool = False,
+    ) -> Sequence[EntityModel]:
+        """List entities with optional filtering and sorting."""
+        logger.debug(f"Listing entities: type={entity_type} sort={sort_by}")
+        return await self.repository.list_entities(
+            entity_type=entity_type,
+            sort_by=sort_by
+        )
 
     async def delete_entity(self, path_id: str) -> bool:
         """Delete entity from database."""
@@ -88,4 +101,4 @@ class EntityService(BaseService[EntityRepository]):
         """Delete entities and their files."""
         logger.debug(f"Deleting entities: {path_ids}")
         deleted_count = await self.repository.delete_by_path_ids(path_ids)
-        return deleted_count > 0
+        return deleted_count > 0