feat: add project_info tool (#19)

Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/api/app.py

@@ -14,7 +14,7 @@ from loguru import logger
 
 from basic_memory import db
 from basic_memory.config import config as app_config
-from basic_memory.api.routers import knowledge, search, memory, resource
+from basic_memory.api.routers import knowledge, search, memory, resource, project_info
 
 
 @asynccontextmanager
@@ -43,6 +43,7 @@ app.include_router(knowledge.router)
 app.include_router(search.router)
 app.include_router(memory.router)
 app.include_router(resource.router)
+app.include_router(project_info.router)
 
 
 @app.exception_handler(Exception)

src/basic_memory/api/routers/__init__.py

@@ -4,5 +4,6 @@ from . import knowledge_router as knowledge
 from . import memory_router as memory
 from . import resource_router as resource
 from . import search_router as search
+from . import project_info_router as project_info
 
-__all__ = ["knowledge", "memory", "resource", "search"]
+__all__ = ["knowledge", "memory", "resource", "search", "project_info"]

src/basic_memory/api/routers/project_info_router.py

@@ -0,0 +1,275 @@
+"""Router for statistics and system information."""
+
+import json
+from datetime import datetime
+
+from fastapi import APIRouter
+from sqlalchemy import text
+
+from basic_memory.config import config, config_manager
+from basic_memory.deps import (
+    ProjectInfoRepositoryDep,
+)
+from basic_memory.repository.project_info_repository import ProjectInfoRepository
+from basic_memory.schemas import (
+    ProjectInfoResponse,
+    ProjectStatistics,
+    ActivityMetrics,
+    SystemStatus,
+)
+from basic_memory.sync.watch_service import WATCH_STATUS_JSON
+
+router = APIRouter(prefix="/stats", tags=["statistics"])
+
+
+@router.get("/project-info", response_model=ProjectInfoResponse)
+async def get_project_info(
+    repository: ProjectInfoRepositoryDep,
+) -> ProjectInfoResponse:
+    """Get comprehensive information about the current Basic Memory project."""
+    # Get statistics
+    statistics = await get_statistics(repository)
+
+    # Get activity metrics
+    activity = await get_activity_metrics(repository)
+
+    # Get system status
+    system = await get_system_status()
+
+    # Get project configuration information
+    project_name = config.project
+    project_path = str(config.home)
+    available_projects = config_manager.projects
+    default_project = config_manager.default_project
+
+    # Construct the response
+    return ProjectInfoResponse(
+        project_name=project_name,
+        project_path=project_path,
+        available_projects=available_projects,
+        default_project=default_project,
+        statistics=statistics,
+        activity=activity,
+        system=system,
+    )
+
+
+async def get_statistics(repository: ProjectInfoRepository) -> ProjectStatistics:
+    """Get statistics about the current project."""
+    # Get basic counts
+    entity_count_result = await repository.execute_query(text("SELECT COUNT(*) FROM entity"))
+    total_entities = entity_count_result.scalar() or 0
+
+    observation_count_result = await repository.execute_query(
+        text("SELECT COUNT(*) FROM observation")
+    )
+    total_observations = observation_count_result.scalar() or 0
+
+    relation_count_result = await repository.execute_query(text("SELECT COUNT(*) FROM relation"))
+    total_relations = relation_count_result.scalar() or 0
+
+    unresolved_count_result = await repository.execute_query(
+        text("SELECT COUNT(*) FROM relation WHERE to_id IS NULL")
+    )
+    total_unresolved = unresolved_count_result.scalar() or 0
+
+    # Get entity counts by type
+    entity_types_result = await repository.execute_query(
+        text("SELECT entity_type, COUNT(*) FROM entity GROUP BY entity_type")
+    )
+    entity_types = {row[0]: row[1] for row in entity_types_result.fetchall()}
+
+    # Get observation counts by category
+    category_result = await repository.execute_query(
+        text("SELECT category, COUNT(*) FROM observation GROUP BY category")
+    )
+    observation_categories = {row[0]: row[1] for row in category_result.fetchall()}
+
+    # Get relation counts by type
+    relation_types_result = await repository.execute_query(
+        text("SELECT relation_type, COUNT(*) FROM relation GROUP BY relation_type")
+    )
+    relation_types = {row[0]: row[1] for row in relation_types_result.fetchall()}
+
+    # Find most connected entities (most outgoing relations)
+    connected_result = await repository.execute_query(
+        text("""
+        SELECT e.id, e.title, e.permalink, COUNT(r.id) AS relation_count
+        FROM entity e
+        JOIN relation r ON e.id = r.from_id
+        GROUP BY e.id
+        ORDER BY relation_count DESC
+        LIMIT 10
+    """)
+    )
+    most_connected = [
+        {"id": row[0], "title": row[1], "permalink": row[2], "relation_count": row[3]}
+        for row in connected_result.fetchall()
+    ]
+
+    # Count isolated entities (no relations)
+    isolated_result = await repository.execute_query(
+        text("""
+        SELECT COUNT(e.id)
+        FROM entity e
+        LEFT JOIN relation r1 ON e.id = r1.from_id
+        LEFT JOIN relation r2 ON e.id = r2.to_id
+        WHERE r1.id IS NULL AND r2.id IS NULL
+    """)
+    )
+    isolated_count = isolated_result.scalar() or 0
+
+    return ProjectStatistics(
+        total_entities=total_entities,
+        total_observations=total_observations,
+        total_relations=total_relations,
+        total_unresolved_relations=total_unresolved,
+        entity_types=entity_types,
+        observation_categories=observation_categories,
+        relation_types=relation_types,
+        most_connected_entities=most_connected,
+        isolated_entities=isolated_count,
+    )
+
+
+async def get_activity_metrics(repository: ProjectInfoRepository) -> ActivityMetrics:
+    """Get activity metrics for the current project."""
+    # Get recently created entities
+    created_result = await repository.execute_query(
+        text("""
+        SELECT id, title, permalink, entity_type, created_at 
+        FROM entity
+        ORDER BY created_at DESC
+        LIMIT 10
+    """)
+    )
+    recently_created = [
+        {
+            "id": row[0],
+            "title": row[1],
+            "permalink": row[2],
+            "entity_type": row[3],
+            "created_at": row[4],
+        }
+        for row in created_result.fetchall()
+    ]
+
+    # Get recently updated entities
+    updated_result = await repository.execute_query(
+        text("""
+        SELECT id, title, permalink, entity_type, updated_at 
+        FROM entity
+        ORDER BY updated_at DESC
+        LIMIT 10
+    """)
+    )
+    recently_updated = [
+        {
+            "id": row[0],
+            "title": row[1],
+            "permalink": row[2],
+            "entity_type": row[3],
+            "updated_at": row[4],
+        }
+        for row in updated_result.fetchall()
+    ]
+
+    # Get monthly growth over the last 6 months
+    # Calculate the start of 6 months ago
+    now = datetime.now()
+    six_months_ago = datetime(
+        now.year - (1 if now.month <= 6 else 0), ((now.month - 6) % 12) or 12, 1
+    )
+
+    # Query for monthly entity creation
+    entity_growth_result = await repository.execute_query(
+        text(f"""
+        SELECT 
+            strftime('%Y-%m', created_at) AS month,
+            COUNT(*) AS count
+        FROM entity
+        WHERE created_at >= '{six_months_ago.isoformat()}'
+        GROUP BY month
+        ORDER BY month
+    """)
+    )
+    entity_growth = {row[0]: row[1] for row in entity_growth_result.fetchall()}
+
+    # Query for monthly observation creation
+    observation_growth_result = await repository.execute_query(
+        text(f"""
+        SELECT 
+            strftime('%Y-%m', created_at) AS month,
+            COUNT(*) AS count
+        FROM observation
+        INNER JOIN entity ON observation.entity_id = entity.id
+        WHERE entity.created_at >= '{six_months_ago.isoformat()}'
+        GROUP BY month
+        ORDER BY month
+    """)
+    )
+    observation_growth = {row[0]: row[1] for row in observation_growth_result.fetchall()}
+
+    # Query for monthly relation creation
+    relation_growth_result = await repository.execute_query(
+        text(f"""
+        SELECT 
+            strftime('%Y-%m', created_at) AS month,
+            COUNT(*) AS count
+        FROM relation
+        INNER JOIN entity ON relation.from_id = entity.id
+        WHERE entity.created_at >= '{six_months_ago.isoformat()}'
+        GROUP BY month
+        ORDER BY month
+    """)
+    )
+    relation_growth = {row[0]: row[1] for row in relation_growth_result.fetchall()}
+
+    # Combine all monthly growth data
+    monthly_growth = {}
+    for month in set(
+        list(entity_growth.keys()) + list(observation_growth.keys()) + list(relation_growth.keys())
+    ):
+        monthly_growth[month] = {
+            "entities": entity_growth.get(month, 0),
+            "observations": observation_growth.get(month, 0),
+            "relations": relation_growth.get(month, 0),
+            "total": (
+                entity_growth.get(month, 0)
+                + observation_growth.get(month, 0)
+                + relation_growth.get(month, 0)
+            ),
+        }
+
+    return ActivityMetrics(
+        recently_created=recently_created,
+        recently_updated=recently_updated,
+        monthly_growth=monthly_growth,
+    )
+
+
+async def get_system_status() -> SystemStatus:
+    """Get system status information."""
+    import basic_memory
+
+    # Get database information
+    db_path = config.database_path
+    db_size = db_path.stat().st_size if db_path.exists() else 0
+    db_size_readable = f"{db_size / (1024 * 1024):.2f} MB"
+
+    # Get watch service status if available
+    watch_status = None
+    watch_status_path = config.home / ".basic-memory" / WATCH_STATUS_JSON
+    if watch_status_path.exists():
+        try:
+            watch_status = json.loads(watch_status_path.read_text())
+        except Exception:  # pragma: no cover
+            pass
+
+    return SystemStatus(
+        version=basic_memory.__version__,
+        database_path=str(db_path),
+        database_size=db_size_readable,
+        watch_status=watch_status,
+        timestamp=datetime.now(),
+    )

src/basic_memory/cli/commands/__init__.py

@@ -1,7 +1,7 @@
 """CLI commands for basic-memory."""
 
 from . import status, sync, db, import_memory_json, mcp, import_claude_conversations
-from . import import_claude_projects, import_chatgpt, tool, project
+from . import import_claude_projects, import_chatgpt, tool, project, project_info
 
 __all__ = [
     "status",
@@ -14,4 +14,5 @@ __all__ = [
     "import_chatgpt",
     "tool",
     "project",
+    "project_info",
 ]

src/basic_memory/cli/commands/project_info.py

@@ -0,0 +1,167 @@
+"""CLI command for project info status."""
+
+import asyncio
+import json
+from datetime import datetime
+
+import typer
+from rich.console import Console
+from rich.table import Table
+from rich.panel import Panel
+from rich.tree import Tree
+
+from basic_memory.cli.app import app
+from basic_memory.mcp.tools.project_info import project_info
+
+
+info_app = typer.Typer()
+app.add_typer(info_app, name="info", help="Get information about your Basic Memory project")
+
+
+@info_app.command("stats")
+def display_project_info(
+    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
+):
+    """Display detailed information and statistics about the current project."""
+    try:
+        # Get project info
+        info = asyncio.run(project_info())
+
+        if json_output:
+            # Convert to JSON and print
+            print(json.dumps(info.model_dump(), indent=2, default=str))
+        else:
+            # Create rich display
+            console = Console()
+
+            # Project configuration section
+            console.print(
+                Panel(
+                    f"[bold]Project:[/bold] {info.project_name}\n"
+                    f"[bold]Path:[/bold] {info.project_path}\n"
+                    f"[bold]Default Project:[/bold] {info.default_project}\n",
+                    title="📊 Basic Memory Project Info",
+                    expand=False,
+                )
+            )
+
+            # Statistics section
+            stats_table = Table(title="📈 Statistics")
+            stats_table.add_column("Metric", style="cyan")
+            stats_table.add_column("Count", style="green")
+
+            stats_table.add_row("Entities", str(info.statistics.total_entities))
+            stats_table.add_row("Observations", str(info.statistics.total_observations))
+            stats_table.add_row("Relations", str(info.statistics.total_relations))
+            stats_table.add_row(
+                "Unresolved Relations", str(info.statistics.total_unresolved_relations)
+            )
+            stats_table.add_row("Isolated Entities", str(info.statistics.isolated_entities))
+
+            console.print(stats_table)
+
+            # Entity types
+            if info.statistics.entity_types:
+                entity_types_table = Table(title="📑 Entity Types")
+                entity_types_table.add_column("Type", style="blue")
+                entity_types_table.add_column("Count", style="green")
+
+                for entity_type, count in info.statistics.entity_types.items():
+                    entity_types_table.add_row(entity_type, str(count))
+
+                console.print(entity_types_table)
+
+            # Most connected entities
+            if info.statistics.most_connected_entities:
+                connected_table = Table(title="🔗 Most Connected Entities")
+                connected_table.add_column("Title", style="blue")
+                connected_table.add_column("Permalink", style="cyan")
+                connected_table.add_column("Relations", style="green")
+
+                for entity in info.statistics.most_connected_entities:
+                    connected_table.add_row(
+                        entity["title"], entity["permalink"], str(entity["relation_count"])
+                    )
+
+                console.print(connected_table)
+
+            # Recent activity
+            if info.activity.recently_updated:
+                recent_table = Table(title="🕒 Recent Activity")
+                recent_table.add_column("Title", style="blue")
+                recent_table.add_column("Type", style="cyan")
+                recent_table.add_column("Last Updated", style="green")
+
+                for entity in info.activity.recently_updated[:5]:  # Show top 5
+                    updated_at = (
+                        datetime.fromisoformat(entity["updated_at"])
+                        if isinstance(entity["updated_at"], str)
+                        else entity["updated_at"]
+                    )
+                    recent_table.add_row(
+                        entity["title"],
+                        entity["entity_type"],
+                        updated_at.strftime("%Y-%m-%d %H:%M"),
+                    )
+
+                console.print(recent_table)
+
+            # System status
+            system_tree = Tree("🖥️ System Status")
+            system_tree.add(f"Basic Memory version: [bold green]{info.system.version}[/bold green]")
+            system_tree.add(
+                f"Database: [cyan]{info.system.database_path}[/cyan] ([green]{info.system.database_size}[/green])"
+            )
+
+            # Watch status
+            if info.system.watch_status:  # pragma: no cover
+                watch_branch = system_tree.add("Watch Service")
+                running = info.system.watch_status.get("running", False)
+                status_color = "green" if running else "red"
+                watch_branch.add(
+                    f"Status: [bold {status_color}]{'Running' if running else 'Stopped'}[/bold {status_color}]"
+                )
+
+                if running:
+                    start_time = (
+                        datetime.fromisoformat(info.system.watch_status.get("start_time", ""))
+                        if isinstance(info.system.watch_status.get("start_time"), str)
+                        else info.system.watch_status.get("start_time")
+                    )
+                    watch_branch.add(
+                        f"Running since: [cyan]{start_time.strftime('%Y-%m-%d %H:%M')}[/cyan]"
+                    )
+                    watch_branch.add(
+                        f"Files synced: [green]{info.system.watch_status.get('synced_files', 0)}[/green]"
+                    )
+                    watch_branch.add(
+                        f"Errors: [{'red' if info.system.watch_status.get('error_count', 0) > 0 else 'green'}]{info.system.watch_status.get('error_count', 0)}[/{'red' if info.system.watch_status.get('error_count', 0) > 0 else 'green'}]"
+                    )
+            else:
+                system_tree.add("[yellow]Watch service not running[/yellow]")
+
+            console.print(system_tree)
+
+            # Available projects
+            projects_table = Table(title="📁 Available Projects")
+            projects_table.add_column("Name", style="blue")
+            projects_table.add_column("Path", style="cyan")
+            projects_table.add_column("Default", style="green")
+
+            for name, path in info.available_projects.items():
+                is_default = name == info.default_project
+                projects_table.add_row(name, path, "✓" if is_default else "")
+
+            console.print(projects_table)
+
+            # Timestamp
+            current_time = (
+                datetime.fromisoformat(str(info.system.timestamp))
+                if isinstance(info.system.timestamp, str)
+                else info.system.timestamp
+            )
+            console.print(f"\nTimestamp: [cyan]{current_time.strftime('%Y-%m-%d %H:%M:%S')}[/cyan]")
+
+    except Exception as e:  # pragma: no cover
+        typer.echo(f"Error getting project info: {e}", err=True)
+        raise typer.Exit(1)

src/basic_memory/deps.py

@@ -16,6 +16,7 @@ from basic_memory.markdown import EntityParser
 from basic_memory.markdown.markdown_processor import MarkdownProcessor
 from basic_memory.repository.entity_repository import EntityRepository
 from basic_memory.repository.observation_repository import ObservationRepository
+from basic_memory.repository.project_info_repository import ProjectInfoRepository
 from basic_memory.repository.relation_repository import RelationRepository
 from basic_memory.repository.search_repository import SearchRepository
 from basic_memory.services import (
@@ -107,6 +108,15 @@ async def get_search_repository(
 SearchRepositoryDep = Annotated[SearchRepository, Depends(get_search_repository)]
 
 
+def get_project_info_repository(
+    session_maker: SessionMakerDep,
+):
+    """Dependency for StatsRepository."""
+    return ProjectInfoRepository(session_maker)
+
+
+ProjectInfoRepositoryDep = Annotated[ProjectInfoRepository, Depends(get_project_info_repository)]
+
 ## services
 
 

src/basic_memory/mcp/tools/__init__.py

@@ -7,7 +7,7 @@ all tools with the MCP server.
 
 # Import tools to register them with MCP
 from basic_memory.mcp.tools.delete_note import delete_note
-from basic_memory.mcp.tools.read_file import read_file
+from basic_memory.mcp.tools.read_content import read_content
 from basic_memory.mcp.tools.build_context import build_context
 from basic_memory.mcp.tools.recent_activity import recent_activity
 from basic_memory.mcp.tools.read_note import read_note
@@ -19,7 +19,7 @@ __all__ = [
     "build_context",
     "canvas",
     "delete_note",
-    "read_file",
+    "read_content",
     "read_note",
     "recent_activity",
     "search",

src/basic_memory/mcp/tools/project_info.py

@@ -0,0 +1,53 @@
+"""Project info tool for Basic Memory MCP server."""
+
+import logfire
+from loguru import logger
+
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_get
+from basic_memory.schemas import ProjectInfoResponse
+from basic_memory.mcp.async_client import client
+
+
+@mcp.tool(
+    description="Get information and statistics about the current Basic Memory project.",
+)
+@logfire.instrument(extract_args=False)
+async def project_info() -> ProjectInfoResponse:
+    """Get comprehensive information about the current Basic Memory project.
+
+    This tool provides detailed statistics and status information about your
+    Basic Memory project, including:
+
+    - Project configuration
+    - Entity, observation, and relation counts
+    - Graph metrics (most connected entities, isolated entities)
+    - Recent activity and growth over time
+    - System status (database, watch service, version)
+
+    Use this tool to:
+    - Verify your Basic Memory installation is working correctly
+    - Get insights into your knowledge base structure
+    - Monitor growth and activity over time
+    - Identify potential issues like unresolved relations
+
+    Returns:
+        Detailed project information and statistics
+
+    Examples:
+        # Get information about the current project
+        info = await project_info()
+
+        # Check entity counts
+        print(f"Total entities: {info.statistics.total_entities}")
+
+        # Check system status
+        print(f"Basic Memory version: {info.system.version}")
+    """
+    logger.info("Getting project info")
+
+    # Call the API endpoint
+    response = await call_get(client, "/stats/project-info")
+
+    # Convert response to ProjectInfoResponse
+    return ProjectInfoResponse.model_validate(response.json())

src/basic_memory/mcp/tools/read_content.py

@@ -144,7 +144,7 @@ def optimize_image(img, content_length, max_output_bytes=350000):
 
 
 @mcp.tool(description="Read a file's raw content by path or permalink")
-async def read_file(path: str) -> dict:
+async def read_content(path: str) -> dict:
     """Read a file's raw content by path or permalink.
 
     This tool provides direct access to file content in the knowledge base,

src/basic_memory/repository/project_info_repository.py

@@ -0,0 +1,9 @@
+from basic_memory.repository.repository import Repository
+
+
+class ProjectInfoRepository(Repository):
+    """Repository for statistics queries."""
+
+    def __init__(self, session_maker):
+        # Initialize with a dummy model since we're just using the execute_query method
+        super().__init__(session_maker, None)  # type: ignore

src/basic_memory/repository/repository.py

@@ -29,10 +29,11 @@ class Repository[T: Base]:
 
     def __init__(self, session_maker: async_sessionmaker[AsyncSession], Model: Type[T]):
         self.session_maker = session_maker
-        self.Model = Model
-        self.mapper = inspect(self.Model).mapper
-        self.primary_key: Column[Any] = self.mapper.primary_key[0]
-        self.valid_columns = [column.key for column in self.mapper.columns]
+        if Model:
+            self.Model = Model
+            self.mapper = inspect(self.Model).mapper
+            self.primary_key: Column[Any] = self.mapper.primary_key[0]
+            self.valid_columns = [column.key for column in self.mapper.columns]
 
     def get_model_data(self, entity_data):
         model_data = {

src/basic_memory/schemas/__init__.py

@@ -37,6 +37,13 @@ from basic_memory.schemas.response import (
     DeleteEntitiesResponse,
 )
 
+from basic_memory.schemas.project_info import (
+    ProjectStatistics,
+    ActivityMetrics,
+    SystemStatus,
+    ProjectInfoResponse,
+)
+
 # For convenient imports, export all models
 __all__ = [
     # Base
@@ -59,4 +66,9 @@ __all__ = [
     "DeleteEntitiesResponse",
     # Delete Operations
     "DeleteEntitiesRequest",
+    # Project Info
+    "ProjectStatistics",
+    "ActivityMetrics",
+    "SystemStatus",
+    "ProjectInfoResponse",
 ]

src/basic_memory/schemas/project_info.py

@@ -0,0 +1,96 @@
+"""Schema for project info response."""
+
+from datetime import datetime
+from typing import Dict, List, Optional, Any
+
+from pydantic import Field, BaseModel
+
+
+class ProjectStatistics(BaseModel):
+    """Statistics about the current project."""
+
+    # Basic counts
+    total_entities: int = Field(description="Total number of entities in the knowledge base")
+    total_observations: int = Field(description="Total number of observations across all entities")
+    total_relations: int = Field(description="Total number of relations between entities")
+    total_unresolved_relations: int = Field(
+        description="Number of relations with unresolved targets"
+    )
+
+    # Entity counts by type
+    entity_types: Dict[str, int] = Field(
+        description="Count of entities by type (e.g., note, conversation)"
+    )
+
+    # Observation counts by category
+    observation_categories: Dict[str, int] = Field(
+        description="Count of observations by category (e.g., tech, decision)"
+    )
+
+    # Relation counts by type
+    relation_types: Dict[str, int] = Field(
+        description="Count of relations by type (e.g., implements, relates_to)"
+    )
+
+    # Graph metrics
+    most_connected_entities: List[Dict[str, Any]] = Field(
+        description="Entities with the most relations, including their titles and permalinks"
+    )
+    isolated_entities: int = Field(description="Number of entities with no relations")
+
+
+class ActivityMetrics(BaseModel):
+    """Activity metrics for the current project."""
+
+    # Recent activity
+    recently_created: List[Dict[str, Any]] = Field(
+        description="Recently created entities with timestamps"
+    )
+    recently_updated: List[Dict[str, Any]] = Field(
+        description="Recently updated entities with timestamps"
+    )
+
+    # Growth over time (last 6 months)
+    monthly_growth: Dict[str, Dict[str, int]] = Field(
+        description="Monthly growth statistics for entities, observations, and relations"
+    )
+
+
+class SystemStatus(BaseModel):
+    """System status information."""
+
+    # Version information
+    version: str = Field(description="Basic Memory version")
+
+    # Database status
+    database_path: str = Field(description="Path to the SQLite database")
+    database_size: str = Field(description="Size of the database in human-readable format")
+
+    # Watch service status
+    watch_status: Optional[Dict[str, Any]] = Field(
+        default=None, description="Watch service status information (if running)"
+    )
+
+    # System information
+    timestamp: datetime = Field(description="Timestamp when the information was collected")
+
+
+class ProjectInfoResponse(BaseModel):
+    """Response for the project_info tool."""
+
+    # Project configuration
+    project_name: str = Field(description="Name of the current project")
+    project_path: str = Field(description="Path to the current project files")
+    available_projects: Dict[str, str] = Field(
+        description="Map of configured project names to paths"
+    )
+    default_project: str = Field(description="Name of the default project")
+
+    # Statistics
+    statistics: ProjectStatistics = Field(description="Statistics about the knowledge base")
+
+    # Activity metrics
+    activity: ActivityMetrics = Field(description="Activity and growth metrics")
+
+    # System status
+    system: SystemStatus = Field(description="System and service status information")

src/basic_memory/sync/watch_service.py

@@ -15,6 +15,8 @@ from basic_memory.config import ProjectConfig
 from basic_memory.services.file_service import FileService
 from basic_memory.sync.sync_service import SyncService
 
+WATCH_STATUS_JSON = "watch-status.json"
+
 
 class WatchEvent(BaseModel):
     timestamp: datetime
@@ -74,7 +76,7 @@ class WatchService:
         self.file_service = file_service
         self.config = config
         self.state = WatchServiceState()
-        self.status_path = config.home / ".basic-memory" / "watch-status.json"
+        self.status_path = config.home / ".basic-memory" / WATCH_STATUS_JSON
         self.status_path.parent.mkdir(parents=True, exist_ok=True)
         self.console = Console()
 

tests/api/test_project_info_router.py

@@ -0,0 +1,110 @@
+"""Tests for the stats router API endpoints."""
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_get_project_info_endpoint(test_graph, client, test_config):
+    """Test the project-info endpoint returns correctly structured data."""
+    # Set up some test data in the database
+
+    # Call the endpoint
+    response = await client.get("/stats/project-info")
+
+    # Verify response
+    assert response.status_code == 200
+    data = response.json()
+
+    # Check top-level keys
+    assert "project_name" in data
+    assert "project_path" in data
+    assert "available_projects" in data
+    assert "default_project" in data
+    assert "statistics" in data
+    assert "activity" in data
+    assert "system" in data
+
+    # Check statistics
+    stats = data["statistics"]
+    assert "total_entities" in stats
+    assert stats["total_entities"] >= 0
+    assert "total_observations" in stats
+    assert stats["total_observations"] >= 0
+    assert "total_relations" in stats
+    assert stats["total_relations"] >= 0
+
+    # Check activity
+    activity = data["activity"]
+    assert "recently_created" in activity
+    assert "recently_updated" in activity
+    assert "monthly_growth" in activity
+
+    # Check system
+    system = data["system"]
+    assert "version" in system
+    assert "database_path" in system
+    assert "database_size" in system
+    assert "timestamp" in system
+
+
+@pytest.mark.asyncio
+async def test_get_project_info_content(test_graph, client, test_config):
+    """Test that project-info contains actual data from the test database."""
+    # Call the endpoint
+    response = await client.get("/stats/project-info")
+
+    # Verify response
+    assert response.status_code == 200
+    data = response.json()
+
+    # Check that test_graph content is reflected in statistics
+    stats = data["statistics"]
+
+    # Our test graph should have at least a few entities
+    assert stats["total_entities"] > 0
+
+    # It should also have some observations
+    assert stats["total_observations"] > 0
+
+    # And relations
+    assert stats["total_relations"] > 0
+
+    # Check that entity types include 'test'
+    assert "test" in stats["entity_types"] or "entity" in stats["entity_types"]
+
+
+@pytest.mark.asyncio
+async def test_get_project_info_watch_status(test_graph, client, test_config):
+    """Test that project-info correctly handles watch status."""
+    # Create a mock watch status file
+    mock_watch_status = {
+        "running": True,
+        "start_time": "2025-03-05T18:00:42.752435",
+        "pid": 7321,
+        "error_count": 0,
+        "last_error": None,
+        "last_scan": "2025-03-05T19:59:02.444416",
+        "synced_files": 6,
+        "recent_events": [],
+    }
+
+    # Mock the Path.exists and Path.read_text methods
+    with (
+        patch("pathlib.Path.exists", return_value=True),
+        patch("pathlib.Path.read_text", return_value=json.dumps(mock_watch_status)),
+    ):
+        # Call the endpoint
+        response = await client.get("/stats/project-info")
+
+        # Verify response
+        assert response.status_code == 200
+        data = response.json()
+
+        # Check that watch status is included
+        assert data["system"]["watch_status"] is not None
+        assert data["system"]["watch_status"]["running"] is True
+        assert data["system"]["watch_status"]["pid"] == 7321
+        assert data["system"]["watch_status"]["synced_files"] == 6

tests/cli/conftest.py

@@ -0,0 +1,30 @@
+from typing import AsyncGenerator
+
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from httpx import AsyncClient, ASGITransport
+
+from basic_memory.api.app import app as fastapi_app
+from basic_memory.deps import get_project_config, get_engine_factory
+
+
+@pytest_asyncio.fixture
+def app(test_config, engine_factory) -> FastAPI:
+    """Create test FastAPI application."""
+    app = fastapi_app
+    app.dependency_overrides[get_project_config] = lambda: test_config
+    app.dependency_overrides[get_engine_factory] = lambda: engine_factory
+    return app
+
+
+@pytest_asyncio.fixture
+async def client(app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
+    """Create test client that both MCP and tests will use."""
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        yield client
+
+
+@pytest.fixture
+def cli_env(test_config, client):
+    pass

tests/cli/test_cli_tools.py

@@ -10,41 +10,15 @@ from textwrap import dedent
 from typing import AsyncGenerator
 from unittest.mock import patch
 
-import pytest
 import pytest_asyncio
-from fastapi import FastAPI
-from httpx import AsyncClient, ASGITransport
 from typer.testing import CliRunner
 
 from basic_memory.cli.commands.tool import tool_app
 from basic_memory.schemas.base import Entity as EntitySchema
-from basic_memory.api.app import app as fastapi_app
-from basic_memory.deps import get_project_config, get_engine_factory
 
 runner = CliRunner()
 
 
-@pytest_asyncio.fixture
-def app(test_config, engine_factory) -> FastAPI:
-    """Create test FastAPI application."""
-    app = fastapi_app
-    app.dependency_overrides[get_project_config] = lambda: test_config
-    app.dependency_overrides[get_engine_factory] = lambda: engine_factory
-    return app
-
-
-@pytest_asyncio.fixture
-async def client(app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
-    """Create test client that both MCP and tests will use."""
-    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
-        yield client
-
-
-@pytest.fixture
-def cli_env(test_config, client):
-    pass
-
-
 @pytest_asyncio.fixture
 async def setup_test_note(entity_service, search_service) -> AsyncGenerator[dict, None]:
     """Create a test note for CLI tests."""

tests/cli/test_project_info.py

@@ -0,0 +1,38 @@
+"""Tests for the project_info CLI command."""
+
+import json
+
+from typer.testing import CliRunner
+
+from basic_memory.cli.main import app as cli_app
+
+
+def test_info_stats_command(cli_env, test_graph):
+    """Test the 'info stats' command with default output."""
+    runner = CliRunner()
+
+    # Run the command
+    result = runner.invoke(cli_app, ["info", "stats"])
+
+    # Verify exit code
+    assert result.exit_code == 0
+
+    # Check that key data is included in the output
+    assert "Basic Memory Project Info" in result.stdout
+
+
+def test_info_stats_json(cli_env, test_graph):
+    """Test the 'info stats --json' command for JSON output."""
+    runner = CliRunner()
+
+    # Run the command with --json flag
+    result = runner.invoke(cli_app, ["info", "stats", "--json"])
+
+    # Verify exit code
+    assert result.exit_code == 0
+
+    # Parse JSON output
+    output = json.loads(result.stdout)
+
+    # Verify JSON structure matches our sample data
+    assert output["project_name"] == "main"

tests/mcp/test_tool_project_info.py

@@ -0,0 +1,127 @@
+"""Tests for the project_info MCP tool."""
+
+from unittest.mock import patch, MagicMock
+
+import pytest
+from httpx import Response
+
+from basic_memory.mcp.tools.project_info import project_info
+from basic_memory.schemas import (
+    ProjectInfoResponse,
+)
+
+
+@pytest.mark.asyncio
+async def test_project_info_tool():
+    """Test that the project_info tool calls the API and returns structured data."""
+    # Create a mock response
+    mock_response = MagicMock(spec=Response)
+    mock_response.status_code = 200
+
+    # Create sample data that matches the schema
+    sample_data = {
+        "project_name": "test",
+        "project_path": "/path/to/test",
+        "available_projects": {"test": "/path/to/test", "other": "/path/to/other"},
+        "default_project": "test",
+        "statistics": {
+            "total_entities": 42,
+            "total_observations": 24,
+            "total_relations": 18,
+            "total_unresolved_relations": 3,
+            "entity_types": {"note": 30, "conversation": 12},
+            "observation_categories": {"tech": 15, "note": 9},
+            "relation_types": {"relates_to": 10, "implements": 8},
+            "most_connected_entities": [
+                {"id": 1, "title": "Test Entity", "permalink": "test/entity", "relation_count": 5}
+            ],
+            "isolated_entities": 2,
+        },
+        "activity": {
+            "recently_created": [
+                {
+                    "id": 1,
+                    "title": "Test Entity",
+                    "permalink": "test/entity",
+                    "entity_type": "note",
+                    "created_at": "2025-03-05T12:00:00",
+                }
+            ],
+            "recently_updated": [
+                {
+                    "id": 1,
+                    "title": "Test Entity",
+                    "permalink": "test/entity",
+                    "entity_type": "note",
+                    "updated_at": "2025-03-05T12:00:00",
+                }
+            ],
+            "monthly_growth": {
+                "2025-03": {"entities": 10, "observations": 5, "relations": 8, "total": 23}
+            },
+        },
+        "system": {
+            "version": "0.1.0",
+            "database_path": "/path/to/db.sqlite",
+            "database_size": "2.50 MB",
+            "watch_status": {
+                "running": True,
+                "start_time": "2025-03-05T12:00:00",
+                "pid": 1234,
+                "error_count": 0,
+                "synced_files": 42,
+            },
+            "timestamp": "2025-03-05T12:00:00",
+        },
+    }
+
+    mock_response.json.return_value = sample_data
+
+    # Mock the call_get function
+    with patch(
+        "basic_memory.mcp.tools.project_info.call_get", return_value=mock_response
+    ) as mock_call_get:
+        # Call the function
+        result = await project_info()
+
+        # Verify that call_get was called with the correct URL
+        mock_call_get.assert_called_once()
+        args, kwargs = mock_call_get.call_args
+        assert args[1] == "/stats/project-info"
+
+        # Verify the result is a ProjectInfoResponse
+        assert isinstance(result, ProjectInfoResponse)
+
+        # Verify the content
+        assert result.project_name == "test"
+        assert result.project_path == "/path/to/test"
+        assert "test" in result.available_projects
+        assert result.default_project == "test"
+
+        # Check statistics
+        assert result.statistics.total_entities == 42
+        assert result.statistics.total_observations == 24
+        assert result.statistics.total_relations == 18
+
+        # Check activity
+        assert len(result.activity.recently_created) == 1
+        assert result.activity.recently_created[0]["title"] == "Test Entity"
+
+        # Check system
+        assert result.system.version == "0.1.0"
+        assert result.system.database_size == "2.50 MB"
+        assert result.system.watch_status is not None
+        assert result.system.watch_status["running"] is True
+
+
+@pytest.mark.asyncio
+async def test_project_info_error_handling():
+    """Test that the project_info tool handles errors gracefully."""
+    # Mock call_get to raise an exception
+    with patch("basic_memory.mcp.tools.project_info.call_get", side_effect=Exception("Test error")):
+        # Verify that the exception propagates
+        with pytest.raises(Exception) as excinfo:
+            await project_info()
+
+        # Verify error message
+        assert "Test error" in str(excinfo.value)

tests/mcp/test_tool_resource.py

@@ -7,8 +7,12 @@ from PIL import Image as PILImage
 import pytest
 from mcp.server.fastmcp.exceptions import ToolError
 
-from basic_memory.mcp.tools import read_file, write_note
-from basic_memory.mcp.tools.read_file import calculate_target_params, resize_image, optimize_image
+from basic_memory.mcp.tools import read_content, write_note
+from basic_memory.mcp.tools.read_content import (
+    calculate_target_params,
+    resize_image,
+    optimize_image,
+)
 
 
 @pytest.mark.asyncio
@@ -30,7 +34,7 @@ async def test_read_file_text_file(app, synced_files):
     assert result is not None
 
     # Now read it as a resource
-    response = await read_file("test/text-resource")
+    response = await read_content("test/text-resource")
 
     assert response["type"] == "text"
     assert "This is a test text resource" in response["text"]
@@ -51,7 +55,7 @@ async def test_read_file_image_file(app, synced_files):
     image_path = synced_files["image"].name
 
     # Read it as a resource
-    response = await read_file(image_path)
+    response = await read_content(image_path)
 
     assert response["type"] == "image"
     assert response["source"]["type"] == "base64"
@@ -79,7 +83,7 @@ async def test_read_file_pdf_file(app, synced_files):
     pdf_path = synced_files["pdf"].name
 
     # Read it as a resource
-    response = await read_file(pdf_path)
+    response = await read_content(pdf_path)
 
     assert response["type"] == "document"
     assert response["source"]["type"] == "base64"
@@ -95,7 +99,7 @@ async def test_read_file_pdf_file(app, synced_files):
 async def test_read_file_not_found(app):
     """Test trying to read a non-existent"""
     with pytest.raises(ToolError, match="Resource not found"):
-        await read_file("does-not-exist")
+        await read_content("does-not-exist")
 
 
 @pytest.mark.asyncio
@@ -110,7 +114,7 @@ async def test_read_file_memory_url(app, synced_files):
 
     # Read it with a memory:// URL
     memory_url = "memory://test/memory-url-test"
-    response = await read_file(memory_url)
+    response = await read_content(memory_url)
 
     assert response["type"] == "text"
     assert "Testing memory:// URL handling for resources" in response["text"]
@@ -174,7 +178,7 @@ async def test_image_conversion(app, synced_files):
     image_path = synced_files["image"].name
 
     # Test reading the resource
-    response = await read_file(image_path)
+    response = await read_content(image_path)
 
     assert response["type"] == "image"
     assert response["source"]["media_type"] == "image/jpeg"