feat: v0.13.0 pre (#122)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude <noreply@anthropic.com>

src/basic_memory/__init__.py

@@ -1,3 +1,9 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
-__version__ = "0.12.3"
+try:
+    from importlib.metadata import version
+
+    __version__ = version("basic-memory")
+except Exception:  # pragma: no cover
+    # Fallback if package not installed (e.g., during development)
+    __version__ = "0.0.0"  # pragma: no cover

src/basic_memory/alembic/versions/cc7172b46608_update_search_index_schema.py

@@ -56,11 +56,6 @@ def upgrade() -> None:
     );
     """)
 
-    # Print instruction to manually reindex after migration
-    print("\n------------------------------------------------------------------")
-    print("IMPORTANT: After migration completes, manually run the reindex command:")
-    print("basic-memory sync")
-    print("------------------------------------------------------------------\n")
 
 
 def downgrade() -> None:

src/basic_memory/api/app.py

@@ -60,15 +60,18 @@ app = FastAPI(
 
 # Include routers
 app.include_router(knowledge.router, prefix="/{project}")
-app.include_router(management.router, prefix="/{project}")
 app.include_router(memory.router, prefix="/{project}")
 app.include_router(resource.router, prefix="/{project}")
 app.include_router(search.router, prefix="/{project}")
-app.include_router(project.router, prefix="/{project}")
+app.include_router(project.project_router, prefix="/{project}")
 app.include_router(directory_router.router, prefix="/{project}")
 app.include_router(prompt_router.router, prefix="/{project}")
 app.include_router(importer_router.router, prefix="/{project}")
 
+# Project resource router works accross projects
+app.include_router(project.project_resource_router)
+app.include_router(management.router)
+
 # Auth routes are handled by FastMCP automatically when auth is enabled
 
 

src/basic_memory/api/routers/directory_router.py

@@ -1,6 +1,8 @@
 """Router for directory tree operations."""
 
-from fastapi import APIRouter
+from typing import List, Optional
+
+from fastapi import APIRouter, Query
 
 from basic_memory.deps import DirectoryServiceDep, ProjectIdDep
 from basic_memory.schemas.directory import DirectoryNode
@@ -27,3 +29,35 @@ async def get_directory_tree(
 
     # Return the hierarchical tree
     return tree
+
+
+@router.get("/list", response_model=List[DirectoryNode])
+async def list_directory(
+    directory_service: DirectoryServiceDep,
+    project_id: ProjectIdDep,
+    dir_name: str = Query("/", description="Directory path to list"),
+    depth: int = Query(1, ge=1, le=10, description="Recursion depth (1-10)"),
+    file_name_glob: Optional[str] = Query(
+        None, description="Glob pattern for filtering file names"
+    ),
+):
+    """List directory contents with filtering and depth control.
+
+    Args:
+        directory_service: Service for directory operations
+        project_id: ID of the current project
+        dir_name: Directory path to list (default: root "/")
+        depth: Recursion depth (1-10, default: 1 for immediate children only)
+        file_name_glob: Optional glob pattern for filtering file names (e.g., "*.md", "*meeting*")
+
+    Returns:
+        List of DirectoryNode objects matching the criteria
+    """
+    # Get directory listing with filtering
+    nodes = await directory_service.list_directory(
+        dir_name=dir_name,
+        depth=depth,
+        file_name_glob=file_name_glob,
+    )
+
+    return nodes

src/basic_memory/api/routers/knowledge_router.py

@@ -10,6 +10,10 @@ from basic_memory.deps import (
     get_search_service,
     SearchServiceDep,
     LinkResolverDep,
+    ProjectPathDep,
+    FileServiceDep,
+    ProjectConfigDep,
+    AppConfigDep,
 )
 from basic_memory.schemas import (
     EntityListResponse,
@@ -17,6 +21,7 @@ from basic_memory.schemas import (
     DeleteEntitiesResponse,
     DeleteEntitiesRequest,
 )
+from basic_memory.schemas.request import EditEntityRequest, MoveEntityRequest
 from basic_memory.schemas.base import Permalink, Entity
 
 router = APIRouter(prefix="/knowledge", tags=["knowledge"])
@@ -43,41 +48,31 @@ async def create_entity(
     result = EntityResponse.model_validate(entity)
 
     logger.info(
-        "API response",
-        endpoint="create_entity",
-        title=result.title,
-        permalink=result.permalink,
-        status_code=201,
+        f"API response: endpoint='create_entity' title={result.title}, permalink={result.permalink}, status_code=201"
     )
     return result
 
 
 @router.put("/entities/{permalink:path}", response_model=EntityResponse)
 async def create_or_update_entity(
+    project: ProjectPathDep,
     permalink: Permalink,
     data: Entity,
     response: Response,
     background_tasks: BackgroundTasks,
     entity_service: EntityServiceDep,
     search_service: SearchServiceDep,
+    file_service: FileServiceDep,
 ) -> EntityResponse:
     """Create or update an entity. If entity exists, it will be updated, otherwise created."""
     logger.info(
-        "API request",
-        endpoint="create_or_update_entity",
-        permalink=permalink,
-        entity_type=data.entity_type,
-        title=data.title,
+        f"API request: create_or_update_entity for {project=}, {permalink=}, {data.entity_type=}, {data.title=}"
     )
 
     # Validate permalink matches
     if data.permalink != permalink:
         logger.warning(
-            "API validation error",
-            endpoint="create_or_update_entity",
-            permalink=permalink,
-            data_permalink=data.permalink,
-            error="Permalink mismatch",
+            f"API validation error: creating/updating entity with permalink mismatch - url={permalink}, data={data.permalink}",
         )
         raise HTTPException(
             status_code=400,
@@ -93,16 +88,107 @@ async def create_or_update_entity(
     result = EntityResponse.model_validate(entity)
 
     logger.info(
-        "API response",
-        endpoint="create_or_update_entity",
-        title=result.title,
-        permalink=result.permalink,
-        created=created,
-        status_code=response.status_code,
+        f"API response: {result.title=}, {result.permalink=}, {created=}, status_code={response.status_code}"
     )
     return result
 
 
+@router.patch("/entities/{identifier:path}", response_model=EntityResponse)
+async def edit_entity(
+    identifier: str,
+    data: EditEntityRequest,
+    background_tasks: BackgroundTasks,
+    entity_service: EntityServiceDep,
+    search_service: SearchServiceDep,
+) -> EntityResponse:
+    """Edit an existing entity using various operations like append, prepend, find_replace, or replace_section.
+
+    This endpoint allows for targeted edits without requiring the full entity content.
+    """
+    logger.info(
+        f"API request: endpoint='edit_entity', identifier='{identifier}', operation='{data.operation}'"
+    )
+
+    try:
+        # Edit the entity using the service
+        entity = await entity_service.edit_entity(
+            identifier=identifier,
+            operation=data.operation,
+            content=data.content,
+            section=data.section,
+            find_text=data.find_text,
+            expected_replacements=data.expected_replacements,
+        )
+
+        # Reindex the updated entity
+        await search_service.index_entity(entity, background_tasks=background_tasks)
+
+        # Return the updated entity response
+        result = EntityResponse.model_validate(entity)
+
+        logger.info(
+            "API response",
+            endpoint="edit_entity",
+            identifier=identifier,
+            operation=data.operation,
+            permalink=result.permalink,
+            status_code=200,
+        )
+
+        return result
+
+    except Exception as e:
+        logger.error(f"Error editing entity: {e}")
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/move")
+async def move_entity(
+    data: MoveEntityRequest,
+    background_tasks: BackgroundTasks,
+    entity_service: EntityServiceDep,
+    project_config: ProjectConfigDep,
+    app_config: AppConfigDep,
+    search_service: SearchServiceDep,
+) -> EntityResponse:
+    """Move an entity to a new file location with project consistency.
+
+    This endpoint moves a note to a different path while maintaining project
+    consistency and optionally updating permalinks based on configuration.
+    """
+    logger.info(
+        f"API request: endpoint='move_entity', identifier='{data.identifier}', destination='{data.destination_path}'"
+    )
+
+    try:
+        # Move the entity using the service
+        moved_entity = await entity_service.move_entity(
+            identifier=data.identifier,
+            destination_path=data.destination_path,
+            project_config=project_config,
+            app_config=app_config,
+        )
+
+        # Get the moved entity to reindex it
+        entity = await entity_service.link_resolver.resolve_link(data.destination_path)
+        if entity:
+            await search_service.index_entity(entity, background_tasks=background_tasks)
+
+        logger.info(
+            "API response",
+            endpoint="move_entity",
+            identifier=data.identifier,
+            destination=data.destination_path,
+            status_code=200,
+        )
+        result = EntityResponse.model_validate(moved_entity)
+        return result
+
+    except Exception as e:
+        logger.error(f"Error moving entity: {e}")
+        raise HTTPException(status_code=400, detail=str(e))
+
+
 ## Read endpoints
 
 
@@ -164,8 +250,8 @@ async def delete_entity(
     # Delete the entity
     deleted = await entity_service.delete_entity(entity.permalink or entity.id)
 
-    # Remove from search index
-    background_tasks.add_task(search_service.delete_by_permalink, entity.permalink)
+    # Remove from search index (entity, observations, and relations)
+    background_tasks.add_task(search_service.handle_delete, entity)
 
     result = DeleteEntitiesResponse(deleted=deleted)
     return result
@@ -188,4 +274,4 @@ async def delete_entities(
         background_tasks.add_task(search_service.delete_by_permalink, permalink)
 
     result = DeleteEntitiesResponse(deleted=deleted)
-    return result
+    return result

src/basic_memory/api/routers/project_router.py

@@ -8,17 +8,18 @@ from basic_memory.schemas import ProjectInfoResponse
 from basic_memory.schemas.project_info import (
     ProjectList,
     ProjectItem,
-    ProjectSwitchRequest,
+    ProjectInfoRequest,
     ProjectStatusResponse,
-    ProjectWatchStatus,
 )
 
-# Define the router - we'll combine stats and project operations
-router = APIRouter(prefix="/project", tags=["project"])
+# Router for resources in a specific project
+project_router = APIRouter(prefix="/project", tags=["project"])
+
+# Router for managing project resources
+project_resource_router = APIRouter(prefix="/projects", tags=["project_management"])
 
 
-# Get project information (moved from project_info_router.py)
-@router.get("/info", response_model=ProjectInfoResponse)
+@project_router.get("/info", response_model=ProjectInfoResponse)
 async def get_project_info(
     project_service: ProjectServiceDep,
 ) -> ProjectInfoResponse:
@@ -26,8 +27,49 @@ async def get_project_info(
     return await project_service.get_project_info()
 
 
+# Update a project
+@project_router.patch("/{name}", response_model=ProjectStatusResponse)
+async def update_project(
+    project_service: ProjectServiceDep,
+    project_name: str = Path(..., description="Name of the project to update"),
+    path: Optional[str] = Body(None, description="New path for the project"),
+    is_active: Optional[bool] = Body(None, description="Status of the project (active/inactive)"),
+) -> ProjectStatusResponse:
+    """Update a project's information in configuration and database.
+
+    Args:
+        project_name: The name of the project to update
+        path: Optional new path for the project
+        is_active: Optional status update for the project
+
+    Returns:
+        Response confirming the project was updated
+    """
+    try:  # pragma: no cover
+        # Get original project info for the response
+        old_project = ProjectItem(
+            name=project_name,
+            path=project_service.projects.get(project_name, ""),
+        )
+
+        await project_service.update_project(project_name, updated_path=path, is_active=is_active)
+
+        # Get updated project info
+        updated_path = path if path else project_service.projects.get(project_name, "")
+
+        return ProjectStatusResponse(
+            message=f"Project '{project_name}' updated successfully",
+            status="success",
+            default=(project_name == project_service.default_project),
+            old_project=old_project,
+            new_project=ProjectItem(name=project_name, path=updated_path),
+        )
+    except ValueError as e:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(e))
+
+
 # List all available projects
-@router.get("/projects", response_model=ProjectList)
+@project_resource_router.get("/projects", response_model=ProjectList)
 async def list_projects(
     project_service: ProjectServiceDep,
 ) -> ProjectList:
@@ -36,32 +78,28 @@ async def list_projects(
     Returns:
         A list of all projects with metadata
     """
-    projects_dict = project_service.projects
+    projects = await project_service.list_projects()
     default_project = project_service.default_project
-    current_project = project_service.current_project
 
-    project_items = []
-    for name, path in projects_dict.items():
-        project_items.append(
-            ProjectItem(
-                name=name,
-                path=path,
-                is_default=(name == default_project),
-                is_current=(name == current_project),
-            )
+    project_items = [
+        ProjectItem(
+            name=project.name,
+            path=project.path,
+            is_default=project.is_default or False,
         )
+        for project in projects
+    ]
 
     return ProjectList(
         projects=project_items,
         default_project=default_project,
-        current_project=current_project,
     )
 
 
 # Add a new project
-@router.post("/projects", response_model=ProjectStatusResponse)
+@project_resource_router.post("/projects", response_model=ProjectStatusResponse)
 async def add_project(
-    project_data: ProjectSwitchRequest,
+    project_data: ProjectInfoRequest,
     project_service: ProjectServiceDep,
 ) -> ProjectStatusResponse:
     """Add a new project to configuration and database.
@@ -82,10 +120,8 @@ async def add_project(
             message=f"Project '{project_data.name}' added successfully",
             status="success",
             default=project_data.set_default,
-            new_project=ProjectWatchStatus(
-                name=project_data.name,
-                path=project_data.path,
-                watch_status=None,
+            new_project=ProjectItem(
+                name=project_data.name, path=project_data.path, is_default=project_data.set_default
             ),
         )
     except ValueError as e:  # pragma: no cover
@@ -93,7 +129,7 @@ async def add_project(
 
 
 # Remove a project
-@router.delete("/projects/{name}", response_model=ProjectStatusResponse)
+@project_resource_router.delete("/{name}", response_model=ProjectStatusResponse)
 async def remove_project(
     project_service: ProjectServiceDep,
     name: str = Path(..., description="Name of the project to remove"),
@@ -106,28 +142,26 @@ async def remove_project(
     Returns:
         Response confirming the project was removed
     """
-    try:  # pragma: no cover
-        # Get project info before removal for the response
-        old_project = ProjectWatchStatus(
-            name=name,
-            path=project_service.projects.get(name, ""),
-            watch_status=None,
-        )
+    try:
+        old_project = await project_service.get_project(name)
+        if not old_project:  # pragma: no cover
+            raise HTTPException(status_code=404, detail=f"Project: '{name}' does not exist")  # pragma: no cover
 
         await project_service.remove_project(name)
 
-        return ProjectStatusResponse(  # pyright: ignore [reportCallIssue]
+        return ProjectStatusResponse(
             message=f"Project '{name}' removed successfully",
             status="success",
             default=False,
-            old_project=old_project,
+            old_project=ProjectItem(name=old_project.name, path=old_project.path),
+            new_project=None,
         )
     except ValueError as e:  # pragma: no cover
         raise HTTPException(status_code=400, detail=str(e))
 
 
 # Set a project as default
-@router.put("/projects/{name}/default", response_model=ProjectStatusResponse)
+@project_resource_router.put("/{name}/default", response_model=ProjectStatusResponse)
 async def set_default_project(
     project_service: ProjectServiceDep,
     name: str = Path(..., description="Name of the project to set as default"),
@@ -140,78 +174,39 @@ async def set_default_project(
     Returns:
         Response confirming the project was set as default
     """
-    try:  # pragma: no cover
+    try:
         # Get the old default project
-        old_default = project_service.default_project
-        old_project = None
-        if old_default != name:
-            old_project = ProjectWatchStatus(
-                name=old_default,
-                path=project_service.projects.get(old_default, ""),
-                watch_status=None,
+        default_name = project_service.default_project
+        default_project = await project_service.get_project(default_name)
+        if not default_project:  # pragma: no cover
+            raise HTTPException(  # pragma: no cover
+                status_code=404, detail=f"Default Project: '{default_name}' does not exist"
             )
 
+        # get the new project
+        new_default_project = await project_service.get_project(name)
+        if not new_default_project:  # pragma: no cover
+            raise HTTPException(status_code=404, detail=f"Project: '{name}' does not exist")  # pragma: no cover
+
         await project_service.set_default_project(name)
 
         return ProjectStatusResponse(
             message=f"Project '{name}' set as default successfully",
             status="success",
             default=True,
-            old_project=old_project,
-            new_project=ProjectWatchStatus(
+            old_project=ProjectItem(name=default_name, path=default_project.path),
+            new_project=ProjectItem(
                 name=name,
-                path=project_service.projects.get(name, ""),
-                watch_status=None,
+                path=new_default_project.path,
+                is_default=True,
             ),
         )
     except ValueError as e:  # pragma: no cover
         raise HTTPException(status_code=400, detail=str(e))
 
 
-# Update a project
-@router.patch("/projects/{name}", response_model=ProjectStatusResponse)
-async def update_project(
-    project_service: ProjectServiceDep,
-    name: str = Path(..., description="Name of the project to update"),
-    path: Optional[str] = Body(None, description="New path for the project"),
-    is_active: Optional[bool] = Body(None, description="Status of the project (active/inactive)"),
-) -> ProjectStatusResponse:
-    """Update a project's information in configuration and database.
-
-    Args:
-        name: The name of the project to update
-        path: Optional new path for the project
-        is_active: Optional status update for the project
-
-    Returns:
-        Response confirming the project was updated
-    """
-    try:  # pragma: no cover
-        # Get original project info for the response
-        old_project = ProjectWatchStatus(
-            name=name,
-            path=project_service.projects.get(name, ""),
-            watch_status=None,
-        )
-
-        await project_service.update_project(name, updated_path=path, is_active=is_active)
-
-        # Get updated project info
-        updated_path = path if path else project_service.projects.get(name, "")
-
-        return ProjectStatusResponse(
-            message=f"Project '{name}' updated successfully",
-            status="success",
-            default=(name == project_service.default_project),
-            old_project=old_project,
-            new_project=ProjectWatchStatus(name=name, path=updated_path, watch_status=None),
-        )
-    except ValueError as e:  # pragma: no cover
-        raise HTTPException(status_code=400, detail=str(e))
-
-
 # Synchronize projects between config and database
-@router.post("/sync", response_model=ProjectStatusResponse)
+@project_resource_router.post("/sync", response_model=ProjectStatusResponse)
 async def synchronize_projects(
     project_service: ProjectServiceDep,
 ) -> ProjectStatusResponse:

src/basic_memory/cli/app.py

@@ -2,6 +2,9 @@ from typing import Optional
 
 import typer
 
+from basic_memory.config import get_project_config
+from basic_memory.mcp.project_session import session
+
 
 def version_callback(value: bool) -> None:
     """Show version and exit."""
@@ -39,25 +42,6 @@ def app_callback(
 ) -> None:
     """Basic Memory - Local-first personal knowledge management."""
 
-    # We use the project option to set the BASIC_MEMORY_PROJECT environment variable
-    # The config module will pick this up when loading
-    if project:  # pragma: no cover
-        import os
-        import importlib
-        from basic_memory import config as config_module
-
-        # Set the environment variable
-        os.environ["BASIC_MEMORY_PROJECT"] = project
-
-        # Reload the config module to pick up the new project
-        importlib.reload(config_module)
-
-        # Update the local reference
-        global app_config
-        from basic_memory.config import app_config as new_config
-
-        app_config = new_config
-
     # Run initialization for every command unless --version was specified
     if not version and ctx.invoked_subcommand is not None:
         from basic_memory.config import app_config
@@ -65,6 +49,21 @@ def app_callback(
 
         ensure_initialization(app_config)
 
+        # Initialize MCP session with the specified project or default
+        if project:  # pragma: no cover
+            # Use the project specified via --project flag
+            current_project_config = get_project_config(project)
+            session.set_current_project(current_project_config.name)
+
+            # Update the global config to use this project
+            from basic_memory.config import update_current_project
+
+            update_current_project(project)
+        else:
+            # Use the default project
+            current_project = app_config.default_project
+            session.set_current_project(current_project)
+
 
 # Register sub-command groups
 import_app = typer.Typer(help="Import data from various sources")

src/basic_memory/cli/commands/project.py

@@ -10,7 +10,8 @@ from rich.table import Table
 
 from basic_memory.cli.app import app
 from basic_memory.config import config
-from basic_memory.mcp.tools.project_info import project_info
+from basic_memory.mcp.project_session import session
+from basic_memory.mcp.resources.project_info import project_info
 import json
 from datetime import datetime
 
@@ -35,7 +36,7 @@ def format_path(path: str) -> str:
     """Format a path for display, using ~ for home directory."""
     home = str(Path.home())
     if path.startswith(home):
-        return path.replace(home, "~", 1)
+        return path.replace(home, "~", 1)  # pragma: no cover
     return path
 
 
@@ -58,7 +59,7 @@ def list_projects() -> None:
 
         for project in result.projects:
             is_default = "✓" if project.is_default else ""
-            is_active = "✓" if project.is_current else ""
+            is_active = "✓" if session.get_current_project() == project.name else ""
             table.add_row(project.name, format_path(project.path), is_default, is_active)
 
         console.print(table)
@@ -148,40 +149,6 @@ def set_default_project(
     console.print("[green]Project activated for current session[/green]")
 
 
-@project_app.command("current")
-def show_current_project() -> None:
-    """Show the current project."""
-    # Use API to get current project
-
-    project_url = config.project_url
-
-    try:
-        response = asyncio.run(call_get(client, f"{project_url}/project/projects"))
-        result = ProjectList.model_validate(response.json())
-
-        # Find the current project from the API response
-        current_project = result.current_project
-        default_project = result.default_project
-
-        # Find the project details in the list
-        for project in result.projects:
-            if project.name == current_project:
-                console.print(f"Current project: [cyan]{project.name}[/cyan]")
-                console.print(f"Path: [green]{format_path(project.path)}[/green]")
-                # Use app_config for database_path, not project config
-                from basic_memory.config import app_config
-
-                console.print(
-                    f"Database: [blue]{format_path(str(app_config.app_database_path))}[/blue]"
-                )
-                console.print(f"Default project: [yellow]{default_project}[/yellow]")
-                break
-    except Exception as e:
-        console.print(f"[red]Error getting current project: {str(e)}[/red]")
-        console.print("[yellow]Note: Make sure the Basic Memory server is running.[/yellow]")
-        raise typer.Exit(1)
-
-
 @project_app.command("sync")
 def synchronize_projects() -> None:
     """Synchronize projects between configuration file and database."""

src/basic_memory/config.py

@@ -9,10 +9,12 @@ from typing import Any, Dict, Literal, Optional, List
 from loguru import logger
 from pydantic import Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
+from setuptools.command.setopt import config_file
 
 import basic_memory
 from basic_memory.utils import setup_logging, generate_permalink
 
+
 DATABASE_NAME = "memory.db"
 APP_DATABASE_NAME = "memory.db"  # Using the same name but in the app directory
 DATA_DIR_NAME = ".basic-memory"
@@ -147,7 +149,11 @@ class ConfigManager:
 
     def __init__(self) -> None:
         """Initialize the configuration manager."""
-        self.config_dir = Path.home() / DATA_DIR_NAME
+        home = os.getenv("HOME", Path.home())
+        if isinstance(home, str):
+            home = Path(home)
+
+        self.config_dir = home / DATA_DIR_NAME
         self.config_file = self.config_dir / CONFIG_FILE_NAME
 
         # Ensure config directory exists
@@ -156,9 +162,6 @@ class ConfigManager:
         # Load or create configuration
         self.config = self.load_config()
 
-        # Current project context for the session
-        self.current_project_id: int
-
     def load_config(self) -> BasicMemoryConfig:
         """Load configuration from file or create default."""
         if self.config_file.exists():
@@ -177,7 +180,7 @@ class ConfigManager:
 
     def save_config(self, config: BasicMemoryConfig) -> None:
         """Save configuration to file."""
-        try:
+        try:            
             self.config_file.write_text(json.dumps(config.model_dump(), indent=2))
         except Exception as e:  # pragma: no cover
             logger.error(f"Failed to save config: {e}")
@@ -192,7 +195,7 @@ class ConfigManager:
         """Get the default project name."""
         return self.config.default_project
 
-    def add_project(self, name: str, path: str) -> None:
+    def add_project(self, name: str, path: str) -> ProjectConfig:
         """Add a new project to the configuration."""
         if name in self.config.projects:  # pragma: no cover
             raise ValueError(f"Project '{name}' already exists")
@@ -203,6 +206,7 @@ class ConfigManager:
 
         self.config.projects[name] = str(project_path)
         self.save_config(self.config)
+        return ProjectConfig(name=name, home=project_path)
 
     def remove_project(self, name: str) -> None:
         """Remove a project from the configuration."""
@@ -224,15 +228,36 @@ class ConfigManager:
         self.save_config(self.config)
 
 
-def get_project_config() -> ProjectConfig:
-    """Get the project configuration for the current session."""
+def get_project_config(project_name: Optional[str] = None) -> ProjectConfig:
+    """
+    Get the project configuration for the current session.
+    If project_name is provided, it will be used instead of the default project.
+    """
 
-    # Get project name from environment variable or use provided name or default
-    env_project_name = os.environ.get("BASIC_MEMORY_PROJECT", None)
-    actual_project_name = env_project_name or config_manager.default_project
+    actual_project_name = None
+
+    # load the config from file
+    global app_config
+    app_config = config_manager.load_config()
+
+    # Get project name from environment variable
+    os_project_name = os.environ.get("BASIC_MEMORY_PROJECT", None)
+    if os_project_name:  # pragma: no cover
+        logger.warning(
+            f"BASIC_MEMORY_PROJECT is not supported anymore. Use the --project flag or set the default project in the config instead. Setting default project to {os_project_name}"
+        )
+        actual_project_name = project_name
+    # if the project_name is passed in, use it
+    elif not project_name:
+        # use default
+        actual_project_name = app_config.default_project
+    else:  # pragma: no cover
+        actual_project_name = project_name
 
     # the config contains a dict[str,str] of project names and absolute paths
-    project_path = config_manager.projects.get(actual_project_name)
+    assert actual_project_name is not None, "actual_project_name cannot be None"
+
+    project_path = app_config.projects.get(actual_project_name)
     if not project_path:  # pragma: no cover
         raise ValueError(f"Project '{actual_project_name}' not found")
 
@@ -249,6 +274,15 @@ app_config: BasicMemoryConfig = config_manager.config
 config: ProjectConfig = get_project_config()
 
 
+def update_current_project(project_name: str) -> None:
+    """Update the global config to use a different project.
+
+    This is used by the CLI when --project flag is specified.
+    """
+    global config
+    config = get_project_config(project_name)  # pragma: no cover
+
+
 # setup logging to a single log file in user home directory
 user_home = Path.home()
 log_dir = user_home / DATA_DIR_NAME

src/basic_memory/deps.py

@@ -1,6 +1,7 @@
 """Dependency injection functions for basic-memory services."""
 
 from typing import Annotated
+from loguru import logger
 
 from fastapi import Depends, HTTPException, Path, status
 from sqlalchemy.ext.asyncio import (
@@ -8,9 +9,10 @@ from sqlalchemy.ext.asyncio import (
     AsyncEngine,
     async_sessionmaker,
 )
+import pathlib
 
 from basic_memory import db
-from basic_memory.config import ProjectConfig, config, BasicMemoryConfig
+from basic_memory.config import ProjectConfig, BasicMemoryConfig
 from basic_memory.importers import (
     ChatGPTImporter,
     ClaudeConversationsImporter,
@@ -44,8 +46,30 @@ AppConfigDep = Annotated[BasicMemoryConfig, Depends(get_app_config)]  # pragma:
 ## project
 
 
-def get_project_config() -> ProjectConfig:  # pragma: no cover
-    return config
+async def get_project_config(
+    project: "ProjectPathDep", project_repository: "ProjectRepositoryDep"
+) -> ProjectConfig:  # pragma: no cover
+    """Get the current project referenced from request state.
+
+    Args:
+        request: The current request object
+        project_repository: Repository for project operations
+
+    Returns:
+        The resolved project config
+
+    Raises:
+        HTTPException: If project is not found
+    """
+
+    project_obj = await project_repository.get_by_permalink(str(project))
+    if project_obj:
+        return ProjectConfig(name=project_obj.name, home=pathlib.Path(project_obj.path))
+
+    # Not found
+    raise HTTPException(  # pragma: no cover
+        status_code=status.HTTP_404_NOT_FOUND, detail=f"Project '{project}' not found."
+    )
 
 
 ProjectConfigDep = Annotated[ProjectConfig, Depends(get_project_config)]  # pragma: no cover
@@ -203,7 +227,12 @@ MarkdownProcessorDep = Annotated[MarkdownProcessor, Depends(get_markdown_process
 async def get_file_service(
     project_config: ProjectConfigDep, markdown_processor: MarkdownProcessorDep
 ) -> FileService:
-    return FileService(project_config.home, markdown_processor)
+    logger.debug(
+        f"Creating FileService for project: {project_config.name}, base_path: {project_config.home}"
+    )
+    file_service = FileService(project_config.home, markdown_processor)
+    logger.debug(f"Created FileService for project: {file_service} ")
+    return file_service
 
 
 FileServiceDep = Annotated[FileService, Depends(get_file_service)]

src/basic_memory/markdown/entity_parser.py

@@ -92,7 +92,6 @@ class EntityParser:
     async def parse_file(self, path: Path | str) -> EntityMarkdown:
         """Parse markdown file into EntityMarkdown."""
 
-        # TODO move to api endpoint to check if absolute path was requested
         # Check if the path is already absolute
         if (
             isinstance(path, Path)
@@ -101,12 +100,16 @@ class EntityParser:
         ):
             absolute_path = Path(path)
         else:
-            absolute_path = self.base_path / path
+            absolute_path = self.get_file_path(path)
 
         # Parse frontmatter and content using python-frontmatter
         file_content = absolute_path.read_text(encoding="utf-8")
         return await self.parse_file_content(absolute_path, file_content)
 
+    def get_file_path(self, path):
+        """Get absolute path for a file using the base path for the project."""
+        return self.base_path / path
+
     async def parse_file_content(self, absolute_path, file_content):
         post = frontmatter.loads(file_content)
         # Extract file stat info

src/basic_memory/mcp/project_session.py

@@ -0,0 +1,103 @@
+"""Project session management for Basic Memory MCP server.
+
+Provides simple in-memory project context for MCP tools, allowing users to switch
+between projects during a conversation without restarting the server.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+from loguru import logger
+
+from basic_memory.config import ProjectConfig, get_project_config
+
+
+@dataclass
+class ProjectSession:
+    """Simple in-memory project context for MCP session.
+
+    This class manages the current project context that tools use when no explicit
+    project is specified. It's initialized with the default project from config
+    and can be changed during the conversation.
+    """
+
+    current_project: Optional[str] = None
+    default_project: Optional[str] = None
+
+    def initialize(self, default_project: str) -> None:
+        """Set the default project from config on startup.
+
+        Args:
+            default_project: The project name from configuration
+        """
+        self.default_project = default_project
+        self.current_project = default_project
+        logger.info(f"Initialized project session with default project: {default_project}")
+
+    def get_current_project(self) -> str:
+        """Get the currently active project name.
+
+        Returns:
+            The current project name, falling back to default, then 'main'
+        """
+        return self.current_project or self.default_project or "main"
+
+    def set_current_project(self, project_name: str) -> None:
+        """Set the current project context.
+
+        Args:
+            project_name: The project to switch to
+        """
+        previous = self.current_project
+        self.current_project = project_name
+        logger.info(f"Switched project context: {previous} -> {project_name}")
+
+    def get_default_project(self) -> str:
+        """Get the default project name from startup.
+
+        Returns:
+            The default project name, or 'main' if not set
+        """
+        return self.default_project or "main"  # pragma: no cover
+
+    def reset_to_default(self) -> None:  # pragma: no cover
+        """Reset current project back to the default project."""
+        self.current_project = self.default_project  # pragma: no cover
+        logger.info(f"Reset project context to default: {self.default_project}")  # pragma: no cover
+
+
+# Global session instance
+session = ProjectSession()
+
+
+def get_active_project(project_override: Optional[str] = None) -> ProjectConfig:
+    """Get the active project name for a tool call.
+
+    This is the main function tools should use to determine which project
+    to operate on.
+
+    Args:
+        project_override: Optional explicit project name from tool parameter
+
+    Returns:
+        The project name to use (override takes precedence over session context)
+    """
+    if project_override:  # pragma: no cover
+        project = get_project_config(project_override)
+        session.set_current_project(project_override)
+        return project
+
+    current_project = session.get_current_project()
+    return get_project_config(current_project)
+
+
+def add_project_metadata(result: str, project_name: str) -> str:
+    """Add project context as metadata footer for LLM awareness.
+
+    Args:
+        result: The tool result string
+        project_name: The project name that was used
+
+    Returns:
+        Result with project metadata footer
+    """
+    return f"{result}\n\n<!-- Project: {project_name} -->"  # pragma: no cover

src/basic_memory/mcp/resources/project_info.py

@@ -2,14 +2,15 @@
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_get
 from basic_memory.schemas import ProjectInfoResponse
 
 
-@mcp.tool(
+@mcp.resource(
+    uri="memory://project_info",
     description="Get information and statistics about the current Basic Memory project.",
 )
 async def project_info() -> ProjectInfoResponse:
@@ -44,7 +45,8 @@ async def project_info() -> ProjectInfoResponse:
         print(f"Basic Memory version: {info.system.version}")
     """
     logger.info("Getting project info")
-    project_url = get_project_config().project_url
+    project_config = get_active_project()
+    project_url = project_config.project_url
 
     # Call the API endpoint
     response = await call_get(client, f"{project_url}/project/info")

src/basic_memory/mcp/server.py

@@ -15,6 +15,7 @@ from mcp.server.auth.settings import AuthSettings
 from basic_memory.config import app_config
 from basic_memory.services.initialization import initialize_app
 from basic_memory.mcp.auth_provider import BasicMemoryOAuthProvider
+from basic_memory.mcp.project_session import session
 from basic_memory.mcp.external_auth_provider import (
     create_github_provider,
     create_google_provider,
@@ -37,6 +38,10 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:  # pragma:
     """Manage application lifecycle with type-safe context"""
     # Initialize on startup
     watch_task = await initialize_app(app_config)
+
+    # Initialize project session with default project
+    session.initialize(app_config.default_project)
+
     try:
         yield AppContext(watch_task=watch_task)
     finally:

src/basic_memory/mcp/tools/__init__.py

@@ -14,14 +14,34 @@ from basic_memory.mcp.tools.read_note import read_note
 from basic_memory.mcp.tools.write_note import write_note
 from basic_memory.mcp.tools.search import search_notes
 from basic_memory.mcp.tools.canvas import canvas
+from basic_memory.mcp.tools.list_directory import list_directory
+from basic_memory.mcp.tools.edit_note import edit_note
+from basic_memory.mcp.tools.move_note import move_note
+from basic_memory.mcp.tools.project_management import (
+    list_projects,
+    switch_project,
+    get_current_project,
+    set_default_project,
+    create_project,
+    delete_project,
+)
 
 __all__ = [
     "build_context",
     "canvas",
+    "create_project",
     "delete_note",
+    "delete_project",
+    "edit_note",
+    "get_current_project",
+    "list_directory",
+    "list_projects",
+    "move_note",
     "read_content",
     "read_note",
     "recent_activity",
     "search_notes",
+    "set_default_project",
+    "switch_project",
     "write_note",
 ]

src/basic_memory/mcp/tools/build_context.py

@@ -4,10 +4,10 @@ from typing import Optional
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_get
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.base import TimeFrame
 from basic_memory.schemas.memory import (
     GraphContext,
@@ -36,6 +36,7 @@ async def build_context(
     page: int = 1,
     page_size: int = 10,
     max_related: int = 10,
+    project: Optional[str] = None,
 ) -> GraphContext:
     """Get context needed to continue a discussion.
 
@@ -50,6 +51,7 @@ async def build_context(
         page: Page number of results to return (default: 1)
         page_size: Number of results to return per page (default: 10)
         max_related: Maximum number of related results to return (default: 10)
+        project: Optional project name to build context from. If not provided, uses current active project.
 
     Returns:
         GraphContext containing:
@@ -69,11 +71,15 @@ async def build_context(
 
         # Research the history of a feature
         build_context("memory://features/knowledge-graph", timeframe="3 months ago")
+
+        # Build context from specific project
+        build_context("memory://specs/search", project="work-project")
     """
     logger.info(f"Building context from {url}")
     url = normalize_memory_url(url)
 
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     response = await call_get(
         client,

src/basic_memory/mcp/tools/canvas.py

@@ -4,14 +4,14 @@ This tool creates Obsidian canvas files (.canvas) using the JSON Canvas 1.0 spec
 """
 
 import json
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Optional
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_put
+from basic_memory.mcp.project_session import get_active_project
 
 
 @mcp.tool(
@@ -22,6 +22,7 @@ async def canvas(
     edges: List[Dict[str, Any]],
     title: str,
     folder: str,
+    project: Optional[str] = None,
 ) -> str:
     """Create an Obsidian canvas file with the provided nodes and edges.
 
@@ -35,6 +36,7 @@ async def canvas(
         edges: List of edge objects following JSON Canvas 1.0 spec
         title: The title of the canvas (will be saved as title.canvas)
         folder: The folder where the file should be saved
+        project: Optional project name to create canvas in. If not provided, uses current active project.
 
     Returns:
         A summary of the created canvas file
@@ -72,8 +74,16 @@ async def canvas(
       ]
     }
     ```
+
+    Examples:
+        # Create canvas in current project
+        canvas(nodes=[...], edges=[...], title="My Canvas", folder="diagrams")
+
+        # Create canvas in specific project
+        canvas(nodes=[...], edges=[...], title="My Canvas", folder="diagrams", project="work-project")
     """
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     # Ensure path has .canvas extension
     file_title = title if title.endswith(".canvas") else f"{title}.canvas"

src/basic_memory/mcp/tools/delete_note.py

@@ -1,18 +1,19 @@
-from basic_memory.config import get_project_config
+from typing import Optional
+
 from basic_memory.mcp.tools.utils import call_delete
-
-
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.async_client import client
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas import DeleteEntitiesResponse
 
 
 @mcp.tool(description="Delete a note by title or permalink")
-async def delete_note(identifier: str) -> bool:
+async def delete_note(identifier: str, project: Optional[str] = None) -> bool:
     """Delete a note from the knowledge base.
 
     Args:
         identifier: Note title or permalink
+        project: Optional project name to delete from. If not provided, uses current active project.
 
     Returns:
         True if note was deleted, False otherwise
@@ -23,8 +24,12 @@ async def delete_note(identifier: str) -> bool:
 
         # Delete by permalink
         delete_note("notes/project-planning")
+
+        # Delete from specific project
+        delete_note("notes/project-planning", project="work-project")
     """
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     response = await call_delete(client, f"{project_url}/knowledge/entities/{identifier}")
     result = DeleteEntitiesResponse.model_validate(response.json())

src/basic_memory/mcp/tools/edit_note.py

@@ -0,0 +1,297 @@
+"""Edit note tool for Basic Memory MCP server."""
+
+from typing import Optional
+
+from loguru import logger
+
+from basic_memory.mcp.async_client import client
+from basic_memory.mcp.project_session import get_active_project
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_patch
+from basic_memory.schemas import EntityResponse
+
+
+def _format_error_response(
+    error_message: str,
+    operation: str,
+    identifier: str,
+    find_text: Optional[str] = None,
+    expected_replacements: int = 1,
+) -> str:
+    """Format helpful error responses for edit_note failures that guide the AI to retry successfully."""
+
+    # Entity not found errors
+    if "Entity not found" in error_message or "entity not found" in error_message.lower():
+        return f"""# Edit Failed - Note Not Found
+
+The note with identifier '{identifier}' could not be found.
+
+## Suggestions to try:
+1. **Search for the note first**: Use `search_notes("{identifier.split("/")[-1]}")` to find similar notes
+2. **Try different identifier formats**:
+   - If you used a permalink like "folder/note-title", try just the title: "{identifier.split("/")[-1].replace("-", " ").title()}"
+   - If you used a title, try the permalink format: "{identifier.lower().replace(" ", "-")}"
+   - Use `read_note()` first to verify the note exists and get the correct identifiers
+
+## Alternative approach:
+Use `write_note()` to create the note first, then edit it."""
+
+    # Find/replace specific errors
+    if operation == "find_replace":
+        if "Text to replace not found" in error_message:
+            return f"""# Edit Failed - Text Not Found
+
+The text '{find_text}' was not found in the note '{identifier}'.
+
+## Suggestions to try:
+1. **Read the note first**: Use `read_note("{identifier}")` to see the current content
+2. **Check for exact matches**: The search is case-sensitive and must match exactly
+3. **Try a broader search**: Search for just part of the text you want to replace
+4. **Use expected_replacements=0**: If you want to verify the text doesn't exist
+
+## Alternative approaches:
+- Use `append` or `prepend` to add new content instead
+- Use `replace_section` if you're trying to update a specific section"""
+
+        if "Expected" in error_message and "occurrences" in error_message:
+            # Extract the actual count from error message if possible
+            import re
+
+            match = re.search(r"found (\d+)", error_message)
+            actual_count = match.group(1) if match else "a different number of"
+
+            return f"""# Edit Failed - Wrong Replacement Count
+
+Expected {expected_replacements} occurrences of '{find_text}' but found {actual_count}.
+
+## How to fix:
+1. **Read the note first**: Use `read_note("{identifier}")` to see how many times '{find_text}' appears
+2. **Update expected_replacements**: Set expected_replacements={actual_count} in your edit_note call
+3. **Be more specific**: If you only want to replace some occurrences, make your find_text more specific
+
+## Example:
+```
+edit_note("{identifier}", "find_replace", "new_text", find_text="{find_text}", expected_replacements={actual_count})
+```"""
+
+    # Section replacement errors
+    if operation == "replace_section" and "Multiple sections" in error_message:
+        return f"""# Edit Failed - Duplicate Section Headers
+
+Multiple sections found with the same header in note '{identifier}'.
+
+## How to fix:
+1. **Read the note first**: Use `read_note("{identifier}")` to see the document structure
+2. **Make headers unique**: Add more specific text to distinguish sections
+3. **Use append instead**: Add content at the end rather than replacing a specific section
+
+## Alternative approach:
+Use `find_replace` to update specific text within the duplicate sections."""
+
+    # Generic server/request errors
+    if (
+        "Invalid request" in error_message or "malformed" in error_message.lower()
+    ):  # pragma: no cover
+        return f"""# Edit Failed - Request Error
+
+There was a problem with the edit request to note '{identifier}': {error_message}.
+
+## Common causes and fixes:
+1. **Note doesn't exist**: Use `search_notes()` or `read_note()` to verify the note exists
+2. **Invalid identifier format**: Try different identifier formats (title vs permalink)
+3. **Empty or invalid content**: Check that your content is properly formatted
+4. **Server error**: Try the operation again, or use `read_note()` first to verify the note state
+
+## Troubleshooting steps:
+1. Verify the note exists: `read_note("{identifier}")`
+2. If not found, search for it: `search_notes("{identifier.split("/")[-1]}")`
+3. Try again with the correct identifier from the search results"""
+
+    # Fallback for other errors
+    return f"""# Edit Failed
+
+Error editing note '{identifier}': {error_message}
+
+## General troubleshooting:
+1. **Verify the note exists**: Use `read_note("{identifier}")` to check
+2. **Check your parameters**: Ensure all required parameters are provided correctly
+3. **Read the note content first**: Use `read_note()` to understand the current structure
+4. **Try a simpler operation**: Start with `append` if other operations fail
+
+## Need help?
+- Use `search_notes()` to find notes
+- Use `read_note()` to examine content before editing
+- Check that identifiers, section headers, and find_text match exactly"""
+
+
+@mcp.tool(
+    description="Edit an existing markdown note using various operations like append, prepend, find_replace, or replace_section.",
+)
+async def edit_note(
+    identifier: str,
+    operation: str,
+    content: str,
+    section: Optional[str] = None,
+    find_text: Optional[str] = None,
+    expected_replacements: int = 1,
+    project: Optional[str] = None,
+) -> str:
+    """Edit an existing markdown note in the knowledge base.
+
+    This tool allows you to make targeted changes to existing notes without rewriting the entire content.
+    It supports various operations for different editing scenarios.
+
+    Args:
+        identifier: The title, permalink, or memory:// URL of the note to edit
+        operation: The editing operation to perform:
+                  - "append": Add content to the end of the note
+                  - "prepend": Add content to the beginning of the note
+                  - "find_replace": Replace occurrences of find_text with content
+                  - "replace_section": Replace content under a specific markdown header
+        content: The content to add or use for replacement
+        section: For replace_section operation - the markdown header to replace content under (e.g., "## Notes", "### Implementation")
+        find_text: For find_replace operation - the text to find and replace
+        expected_replacements: For find_replace operation - the expected number of replacements (validation will fail if actual doesn't match)
+        project: Optional project name to delete from. If not provided, uses current active project.
+
+    Returns:
+        A markdown formatted summary of the edit operation and resulting semantic content
+
+    Examples:
+        # Add new content to end of note
+        edit_note("project-planning", "append", "\\n## New Requirements\\n- Feature X\\n- Feature Y")
+
+        # Add timestamp at beginning (frontmatter-aware)
+        edit_note("meeting-notes", "prepend", "## 2025-05-25 Update\\n- Progress update...\\n\\n")
+
+        # Update version number (single occurrence)
+        edit_note("config-spec", "find_replace", "v0.13.0", find_text="v0.12.0")
+
+        # Update version in multiple places with validation
+        edit_note("api-docs", "find_replace", "v2.1.0", find_text="v2.0.0", expected_replacements=3)
+
+        # Replace text that appears multiple times - validate count first
+        edit_note("docs/guide", "find_replace", "new-api", find_text="old-api", expected_replacements=5)
+
+        # Replace implementation section
+        edit_note("api-spec", "replace_section", "New implementation approach...\\n", section="## Implementation")
+
+        # Replace subsection with more specific header
+        edit_note("docs/setup", "replace_section", "Updated install steps\\n", section="### Installation")
+
+        # Using different identifier formats
+        edit_note("Meeting Notes", "append", "\\n- Follow up on action items")  # title
+        edit_note("docs/meeting-notes", "append", "\\n- Follow up tasks")       # permalink
+        edit_note("docs/Meeting Notes", "append", "\\n- Next steps")           # folder/title
+
+        # Add new section to document
+        edit_note("project-plan", "replace_section", "TBD - needs research\\n", section="## Future Work")
+
+        # Update status across document (expecting exactly 2 occurrences)
+        edit_note("status-report", "find_replace", "In Progress", find_text="Not Started", expected_replacements=2)
+
+        # Replace text in a file, specifying project name
+        edit_note("docs/guide", "find_replace", "new-api", find_text="old-api", project="my-project"))
+
+    """
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
+
+    logger.info("MCP tool call", tool="edit_note", identifier=identifier, operation=operation)
+
+    # Validate operation
+    valid_operations = ["append", "prepend", "find_replace", "replace_section"]
+    if operation not in valid_operations:
+        raise ValueError(
+            f"Invalid operation '{operation}'. Must be one of: {', '.join(valid_operations)}"
+        )
+
+    # Validate required parameters for specific operations
+    if operation == "find_replace" and not find_text:
+        raise ValueError("find_text parameter is required for find_replace operation")
+    if operation == "replace_section" and not section:
+        raise ValueError("section parameter is required for replace_section operation")
+
+    # Use the PATCH endpoint to edit the entity
+    try:
+        # Prepare the edit request data
+        edit_data = {
+            "operation": operation,
+            "content": content,
+        }
+
+        # Add optional parameters
+        if section:
+            edit_data["section"] = section
+        if find_text:
+            edit_data["find_text"] = find_text
+        if expected_replacements != 1:  # Only send if different from default
+            edit_data["expected_replacements"] = str(expected_replacements)
+
+        # Call the PATCH endpoint
+        url = f"{project_url}/knowledge/entities/{identifier}"
+        response = await call_patch(client, url, json=edit_data)
+        result = EntityResponse.model_validate(response.json())
+
+        # Format summary
+        summary = [
+            f"# Edited note ({operation})",
+            f"project: {active_project.name}",
+            f"file_path: {result.file_path}",
+            f"permalink: {result.permalink}",
+            f"checksum: {result.checksum[:8] if result.checksum else 'unknown'}",
+        ]
+
+        # Add operation-specific details
+        if operation == "append":
+            lines_added = len(content.split("\n"))
+            summary.append(f"operation: Added {lines_added} lines to end of note")
+        elif operation == "prepend":
+            lines_added = len(content.split("\n"))
+            summary.append(f"operation: Added {lines_added} lines to beginning of note")
+        elif operation == "find_replace":
+            # For find_replace, we can't easily count replacements from here
+            # since we don't have the original content, but the server handled it
+            summary.append("operation: Find and replace operation completed")
+        elif operation == "replace_section":
+            summary.append(f"operation: Replaced content under section '{section}'")
+
+        # Count observations by category (reuse logic from write_note)
+        categories = {}
+        if result.observations:
+            for obs in result.observations:
+                categories[obs.category] = categories.get(obs.category, 0) + 1
+
+            summary.append("\\n## Observations")
+            for category, count in sorted(categories.items()):
+                summary.append(f"- {category}: {count}")
+
+        # Count resolved/unresolved relations
+        unresolved = 0
+        resolved = 0
+        if result.relations:
+            unresolved = sum(1 for r in result.relations if not r.to_id)
+            resolved = len(result.relations) - unresolved
+
+            summary.append("\\n## Relations")
+            summary.append(f"- Resolved: {resolved}")
+            if unresolved:
+                summary.append(f"- Unresolved: {unresolved}")
+
+        logger.info(
+            "MCP tool response",
+            tool="edit_note",
+            operation=operation,
+            permalink=result.permalink,
+            observations_count=len(result.observations),
+            relations_count=len(result.relations),
+            status_code=response.status_code,
+        )
+
+        return "\n".join(summary)
+
+    except Exception as e:
+        logger.error(f"Error editing note: {e}")
+        return _format_error_response(
+            str(e), operation, identifier, find_text, expected_replacements
+        )

src/basic_memory/mcp/tools/list_directory.py

@@ -0,0 +1,154 @@
+"""List directory tool for Basic Memory MCP server."""
+
+from typing import Optional
+
+from loguru import logger
+
+from basic_memory.mcp.async_client import client
+from basic_memory.mcp.project_session import get_active_project
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_get
+
+
+@mcp.tool(
+    description="List directory contents with filtering and depth control.",
+)
+async def list_directory(
+    dir_name: str = "/",
+    depth: int = 1,
+    file_name_glob: Optional[str] = None,
+    project: Optional[str] = None,
+) -> str:
+    """List directory contents from the knowledge base with optional filtering.
+
+    This tool provides 'ls' functionality for browsing the knowledge base directory structure.
+    It can list immediate children or recursively explore subdirectories with depth control,
+    and supports glob pattern filtering for finding specific files.
+
+    Args:
+        dir_name: Directory path to list (default: root "/")
+                 Examples: "/", "/projects", "/research/ml"
+        depth: Recursion depth (1-10, default: 1 for immediate children only)
+               Higher values show subdirectory contents recursively
+        file_name_glob: Optional glob pattern for filtering file names
+                       Examples: "*.md", "*meeting*", "project_*"
+        project: Optional project name to delete from. If not provided, uses current active project.
+    Returns:
+        Formatted listing of directory contents with file metadata
+
+    Examples:
+        # List root directory contents
+        list_directory()
+
+        # List specific folder
+        list_directory(dir_name="/projects")
+
+        # Find all Python files
+        list_directory(file_name_glob="*.py")
+
+        # Deep exploration of research folder
+        list_directory(dir_name="/research", depth=3)
+
+        # Find meeting notes in projects folder
+        list_directory(dir_name="/projects", file_name_glob="*meeting*")
+
+        # Find meeting notes in a specific project
+        list_directory(dir_name="/projects", file_name_glob="*meeting*", project="work-project")
+    """
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
+
+    # Prepare query parameters
+    params = {
+        "dir_name": dir_name,
+        "depth": str(depth),
+    }
+    if file_name_glob:
+        params["file_name_glob"] = file_name_glob
+
+    logger.debug(f"Listing directory '{dir_name}' with depth={depth}, glob='{file_name_glob}'")
+
+    # Call the API endpoint
+    response = await call_get(
+        client,
+        f"{project_url}/directory/list",
+        params=params,
+    )
+
+    nodes = response.json()
+
+    if not nodes:
+        filter_desc = ""
+        if file_name_glob:
+            filter_desc = f" matching '{file_name_glob}'"
+        return f"No files found in directory '{dir_name}'{filter_desc}"
+
+    # Format the results
+    output_lines = []
+    if file_name_glob:
+        output_lines.append(f"Files in '{dir_name}' matching '{file_name_glob}' (depth {depth}):")
+    else:
+        output_lines.append(f"Contents of '{dir_name}' (depth {depth}):")
+    output_lines.append("")
+
+    # Group by type and sort
+    directories = [n for n in nodes if n["type"] == "directory"]
+    files = [n for n in nodes if n["type"] == "file"]
+
+    # Sort by name
+    directories.sort(key=lambda x: x["name"])
+    files.sort(key=lambda x: x["name"])
+
+    # Display directories first
+    for node in directories:
+        path_display = node["directory_path"]
+        output_lines.append(f"📁 {node['name']:<30} {path_display}")
+
+    # Add separator if we have both directories and files
+    if directories and files:
+        output_lines.append("")
+
+    # Display files with metadata
+    for node in files:
+        path_display = node["directory_path"]
+        title = node.get("title", "")
+        updated = node.get("updated_at", "")
+
+        # Remove leading slash if present, requesting the file via read_note does not use the beginning slash'
+        if path_display.startswith("/"):
+            path_display = path_display[1:]
+
+        # Format date if available
+        date_str = ""
+        if updated:
+            try:
+                from datetime import datetime
+
+                dt = datetime.fromisoformat(updated.replace("Z", "+00:00"))
+                date_str = dt.strftime("%Y-%m-%d")
+            except Exception:  # pragma: no cover
+                date_str = updated[:10] if len(updated) >= 10 else ""
+
+        # Create formatted line
+        file_line = f"📄 {node['name']:<30} {path_display}"
+        if title and title != node["name"]:
+            file_line += f" | {title}"
+        if date_str:
+            file_line += f" | {date_str}"
+
+        output_lines.append(file_line)
+
+    # Add summary
+    output_lines.append("")
+    total_count = len(directories) + len(files)
+    summary_parts = []
+    if directories:
+        summary_parts.append(
+            f"{len(directories)} director{'y' if len(directories) == 1 else 'ies'}"
+        )
+    if files:
+        summary_parts.append(f"{len(files)} file{'s' if len(files) != 1 else ''}")
+
+    output_lines.append(f"Total: {total_count} items ({', '.join(summary_parts)})")
+
+    return "\n".join(output_lines)

src/basic_memory/mcp/tools/move_note.py

@@ -0,0 +1,87 @@
+"""Move note tool for Basic Memory MCP server."""
+
+from typing import Optional
+
+from loguru import logger
+
+from basic_memory.mcp.async_client import client
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_post
+from basic_memory.mcp.project_session import get_active_project
+from basic_memory.schemas import EntityResponse
+
+
+@mcp.tool(
+    description="Move a note to a new location, updating database and maintaining links.",
+)
+async def move_note(
+    identifier: str,
+    destination_path: str,
+    project: Optional[str] = None,
+) -> str:
+    """Move a note to a new file location within the same project.
+
+    Args:
+        identifier: Entity identifier (title, permalink, or memory:// URL)
+        destination_path: New path relative to project root (e.g., "work/meetings/2025-05-26.md")
+        project: Optional project name (defaults to current session project)
+
+    Returns:
+        Success message with move details
+
+    Examples:
+        - Move to new folder: move_note("My Note", "work/notes/my-note.md")
+        - Move by permalink: move_note("my-note-permalink", "archive/old-notes/my-note.md") 
+        - Specify project: move_note("My Note", "archive/my-note.md", project="work-project")
+
+    Note: This operation moves notes within the specified project only. Moving notes
+    between different projects is not currently supported.
+
+    The move operation:
+    - Updates the entity's file_path in the database
+    - Moves the physical file on the filesystem
+    - Optionally updates permalinks if configured
+    - Re-indexes the entity for search
+    - Maintains all observations and relations
+    """
+    logger.debug(f"Moving note: {identifier} to {destination_path}")
+
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
+
+    # Prepare move request
+    move_data = {
+        "identifier": identifier,
+        "destination_path": destination_path,
+        "project": active_project.name,
+    }
+
+    # Call the move API endpoint
+    url = f"{project_url}/knowledge/move"
+    response = await call_post(client, url, json=move_data)
+    result = EntityResponse.model_validate(response.json())
+
+    # 10. Build success message
+    result_lines = [
+        "✅ Note moved successfully",
+        "",
+        f"📁 **{identifier}** → **{result.file_path}**",
+        f"🔗 Permalink: {result.permalink}",
+        "📊 Database and search index updated",
+        "",
+        f"<!-- Project: {active_project.name} -->",
+    ]
+
+    # Return the response text which contains the formatted success message
+    result = "\n".join(result_lines)
+
+    # Log the operation
+    logger.info(
+        "Move note completed",
+        identifier=identifier,
+        destination_path=destination_path,
+        project=active_project.name,
+        status_code=response.status_code,
+    )
+
+    return result

src/basic_memory/mcp/tools/project_management.py

@@ -0,0 +1,300 @@
+"""Project management tools for Basic Memory MCP server.
+
+These tools allow users to switch between projects, list available projects,
+and manage project context during conversations.
+"""
+
+from fastmcp import Context
+from loguru import logger
+
+from basic_memory.config import get_project_config
+from basic_memory.mcp.async_client import client
+from basic_memory.mcp.project_session import session, add_project_metadata
+from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_get, call_put, call_post, call_delete
+from basic_memory.schemas import ProjectInfoResponse
+from basic_memory.schemas.project_info import ProjectList, ProjectStatusResponse, ProjectInfoRequest
+
+
+@mcp.tool()
+async def list_projects(ctx: Context | None = None) -> str:
+    """List all available projects with their status.
+
+    Shows all Basic Memory projects that are available, indicating which one
+    is currently active and which is the default.
+
+    Returns:
+        Formatted list of projects with status indicators
+
+    Example:
+        list_projects()
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info("Listing all available projects")
+
+    # Get projects from API
+    response = await call_get(client, "/projects/projects")
+    project_list = ProjectList.model_validate(response.json())
+
+    current = session.get_current_project()
+
+    result = "Available projects:\n"
+
+    for project in project_list.projects:
+        indicators = []
+        if project.name == current:
+            indicators.append("current")
+        if project.is_default:
+            indicators.append("default")
+
+        if indicators:
+            result += f"• {project.name} ({', '.join(indicators)})\n"
+        else:
+            result += f"• {project.name}\n"
+
+    return add_project_metadata(result, current)
+
+
+@mcp.tool()
+async def switch_project(project_name: str, ctx: Context | None = None) -> str:
+    """Switch to a different project context.
+
+    Changes the active project context for all subsequent tool calls.
+    Shows a project summary after switching successfully.
+
+    Args:
+        project_name: Name of the project to switch to
+
+    Returns:
+        Confirmation message with project summary
+
+    Example:
+        switch_project("work-notes")
+        switch_project("personal-journal")
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info(f"Switching to project: {project_name}")
+
+    current_project = session.get_current_project()
+    try:
+        # Validate project exists by getting project list
+        response = await call_get(client, "/projects/projects")
+        project_list = ProjectList.model_validate(response.json())
+
+        # Check if project exists
+        project_exists = any(p.name == project_name for p in project_list.projects)
+        if not project_exists:
+            available_projects = [p.name for p in project_list.projects]
+            return f"Error: Project '{project_name}' not found. Available projects: {', '.join(available_projects)}"
+
+        # Switch to the project
+        session.set_current_project(project_name)
+        current_project = session.get_current_project()
+        project_config = get_project_config(current_project)
+
+        # Get project info to show summary
+        try:
+            response = await call_get(client, f"{project_config.project_url}/project/info")
+            project_info = ProjectInfoResponse.model_validate(response.json())
+
+            result = f"✓ Switched to {project_name} project\n\n"
+            result += "Project Summary:\n"
+            result += f"• {project_info.statistics.total_entities} entities\n"
+            result += f"• {project_info.statistics.total_observations} observations\n"
+            result += f"• {project_info.statistics.total_relations} relations\n"
+
+        except Exception as e:
+            # If we can't get project info, still confirm the switch
+            logger.warning(f"Could not get project info for {project_name}: {e}")
+            result = f"✓ Switched to {project_name} project\n\n"
+            result += "Project summary unavailable.\n"
+
+        return add_project_metadata(result, project_name)
+
+    except Exception as e:
+        logger.error(f"Error switching to project {project_name}: {e}")
+        # Revert to previous project on error
+        session.set_current_project(current_project)
+        raise e
+
+
+@mcp.tool()
+async def get_current_project(ctx: Context | None = None) -> str:
+    """Show the currently active project and basic stats.
+
+    Displays which project is currently active and provides basic information
+    about it.
+
+    Returns:
+        Current project name and basic statistics
+
+    Example:
+        get_current_project()
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info("Getting current project information")
+
+    current_project = session.get_current_project()
+    project_config = get_project_config(current_project)
+    result = f"Current project: {current_project}\n\n"
+
+    # get project stats
+    response = await call_get(client, f"{project_config.project_url}/project/info")
+    project_info = ProjectInfoResponse.model_validate(response.json())
+
+    result += f"• {project_info.statistics.total_entities} entities\n"
+    result += f"• {project_info.statistics.total_observations} observations\n"
+    result += f"• {project_info.statistics.total_relations} relations\n"
+
+    default_project = session.get_default_project()
+    if current_project != default_project:
+        result += f"• Default project: {default_project}\n"
+
+    return add_project_metadata(result, current_project)
+
+
+@mcp.tool()
+async def set_default_project(project_name: str, ctx: Context | None = None) -> str:
+    """Set default project in config. Requires restart to take effect.
+
+    Updates the configuration to use a different default project. This change
+    only takes effect after restarting the Basic Memory server.
+
+    Args:
+        project_name: Name of the project to set as default
+
+    Returns:
+        Confirmation message about config update
+
+    Example:
+        set_default_project("work-notes")
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info(f"Setting default project to: {project_name}")
+
+    # Call API to set default project
+    response = await call_put(client, f"/projects/{project_name}/default")
+    status_response = ProjectStatusResponse.model_validate(response.json())
+
+    result = f"✓ {status_response.message}\n\n"
+    result += "Restart Basic Memory for this change to take effect:\n"
+    result += "basic-memory mcp\n"
+
+    if status_response.old_project:
+        result += f"\nPrevious default: {status_response.old_project.name}\n"
+
+    return add_project_metadata(result, session.get_current_project())
+
+
+@mcp.tool()
+async def create_project(
+    project_name: str, project_path: str, set_default: bool = False, ctx: Context | None = None
+) -> str:
+    """Create a new Basic Memory project.
+
+    Creates a new project with the specified name and path. The project directory
+    will be created if it doesn't exist. Optionally sets the new project as default.
+
+    Args:
+        project_name: Name for the new project (must be unique)
+        project_path: File system path where the project will be stored
+        set_default: Whether to set this project as the default (optional, defaults to False)
+
+    Returns:
+        Confirmation message with project details
+
+    Example:
+        create_project("my-research", "~/Documents/research")
+        create_project("work-notes", "/home/user/work", set_default=True)
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info(f"Creating project: {project_name} at {project_path}")
+
+    # Create the project request
+    project_request = ProjectInfoRequest(
+        name=project_name, path=project_path, set_default=set_default
+    )
+
+    # Call API to create project
+    response = await call_post(client, "/projects/projects", json=project_request.model_dump())
+    status_response = ProjectStatusResponse.model_validate(response.json())
+
+    result = f"✓ {status_response.message}\n\n"
+
+    if status_response.new_project:
+        result += "Project Details:\n"
+        result += f"• Name: {status_response.new_project.name}\n"
+        result += f"• Path: {status_response.new_project.path}\n"
+
+        if set_default:
+            result += "• Set as default project\n"
+
+    result += "\nProject is now available for use.\n"
+
+    # If project was set as default, update session
+    if set_default:
+        session.set_current_project(project_name)
+
+    return add_project_metadata(result, session.get_current_project())
+
+
+@mcp.tool()
+async def delete_project(project_name: str, ctx: Context | None = None) -> str:
+    """Delete a Basic Memory project.
+
+    Removes a project from the configuration and database. This does NOT delete
+    the actual files on disk - only removes the project from Basic Memory's
+    configuration and database records.
+
+    Args:
+        project_name: Name of the project to delete
+
+    Returns:
+        Confirmation message about project deletion
+
+    Example:
+        delete_project("old-project")
+
+    Warning:
+        This action cannot be undone. The project will need to be re-added
+        to access its content through Basic Memory again.
+    """
+    if ctx:  # pragma: no cover
+        await ctx.info(f"Deleting project: {project_name}")
+
+    current_project = session.get_current_project()
+
+    # Check if trying to delete current project
+    if project_name == current_project:
+        raise ValueError(
+            f"Cannot delete the currently active project '{project_name}'. Switch to a different project first."
+        )
+
+    # Get project info before deletion to validate it exists
+    response = await call_get(client, "/projects/projects")
+    project_list = ProjectList.model_validate(response.json())
+
+    # Check if project exists
+    project_exists = any(p.name == project_name for p in project_list.projects)
+    if not project_exists:
+        available_projects = [p.name for p in project_list.projects]
+        raise ValueError(
+            f"Project '{project_name}' not found. Available projects: {', '.join(available_projects)}"
+        )
+
+    # Call API to delete project
+    response = await call_delete(client, f"/projects/{project_name}")
+    status_response = ProjectStatusResponse.model_validate(response.json())
+
+    result = f"✓ {status_response.message}\n\n"
+
+    if status_response.old_project:
+        result += "Removed project details:\n"
+        result += f"• Name: {status_response.old_project.name}\n"
+        if hasattr(status_response.old_project, "path"):
+            result += f"• Path: {status_response.old_project.path}\n"
+
+    result += "Files remain on disk but project is no longer tracked by Basic Memory.\n"
+    result += "Re-add the project to access its content again.\n"
+
+    return add_project_metadata(result, session.get_current_project())

src/basic_memory/mcp/tools/read_content.py

@@ -5,18 +5,19 @@ supporting various file types including text, images, and other binary files.
 Files are read directly without any knowledge graph processing.
 """
 
-from loguru import logger
+from typing import Optional
+import base64
+import io
+
+from loguru import logger
+from PIL import Image as PILImage
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.tools.utils import call_get
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.memory import memory_url_path
 
-import base64
-import io
-from PIL import Image as PILImage
-
 
 def calculate_target_params(content_length):
     """Calculate initial quality and size based on input file size"""
@@ -145,7 +146,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_content(path: str) -> dict:
+async def read_content(path: str, project: Optional[str] = None) -> dict:
     """Read a file's raw content by path or permalink.
 
     This tool provides direct access to file content in the knowledge base,
@@ -159,6 +160,7 @@ async def read_content(path: str) -> dict:
             - A regular file path (docs/example.md)
             - A memory URL (memory://docs/example)
             - A permalink (docs/example)
+        project: Optional project name to read from. If not provided, uses current active project.
 
     Returns:
         A dictionary with the file content and metadata:
@@ -176,10 +178,14 @@ async def read_content(path: str) -> dict:
 
         # Read using memory URL
         content = await read_file("memory://docs/architecture")
+
+        # Read from specific project
+        content = await read_content("docs/example.md", project="work-project")
     """
     logger.info("Reading file", path=path)
 
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     url = memory_url_path(path)
     response = await call_get(client, f"{project_url}/resource/{url}")

src/basic_memory/mcp/tools/read_note.py

@@ -1,21 +1,24 @@
 """Read note tool for Basic Memory MCP server."""
 
 from textwrap import dedent
+from typing import Optional
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.search import search_notes
 from basic_memory.mcp.tools.utils import call_get
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.memory import memory_url_path
 
 
 @mcp.tool(
     description="Read a markdown note by title or permalink.",
 )
-async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
+async def read_note(
+    identifier: str, page: int = 1, page_size: int = 10, project: Optional[str] = None
+) -> str:
     """Read a markdown note from the knowledge base.
 
     This tool finds and retrieves a note by its title, permalink, or content search,
@@ -27,6 +30,7 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
                    Can be a full memory:// URL, a permalink, a title, or search text
         page: Page number for paginated results (default: 1)
         page_size: Number of items per page (default: 10)
+        project: Optional project name to read from. If not provided, uses current active project.
 
     Returns:
         The full markdown content of the note if found, or helpful guidance if not found.
@@ -43,9 +47,13 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
 
         # Read with pagination
         read_note("Project Updates", page=2, page_size=5)
+
+        # Read from specific project
+        read_note("Meeting Notes", project="work-project")
     """
 
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     # Get the file via REST API - first try direct permalink lookup
     entity_path = memory_url_path(identifier)
@@ -66,7 +74,7 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
 
     # Fallback 1: Try title search via API
     logger.info(f"Search title for: {identifier}")
-    title_results = await search_notes(query=identifier, search_type="title")
+    title_results = await search_notes(query=identifier, search_type="title", project=project)
 
     if title_results and title_results.results:
         result = title_results.results[0]  # Get the first/best match
@@ -90,7 +98,7 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
 
     # Fallback 2: Text search as a last resort
     logger.info(f"Title search failed, trying text search for: {identifier}")
-    text_results = await search_notes(query=identifier, search_type="text")
+    text_results = await search_notes(query=identifier, search_type="text", project=project)
 
     # We didn't find a direct match, construct a helpful error message
     if not text_results or not text_results.results:

src/basic_memory/mcp/tools/recent_activity.py

@@ -1,13 +1,13 @@
 """Recent activity tool for Basic Memory MCP server."""
 
-from typing import List, Union
+from typing import List, Union, Optional
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_get
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.base import TimeFrame
 from basic_memory.schemas.memory import GraphContext
 from basic_memory.schemas.search import SearchItemType
@@ -32,6 +32,7 @@ async def recent_activity(
     page: int = 1,
     page_size: int = 10,
     max_related: int = 10,
+    project: Optional[str] = None,
 ) -> GraphContext:
     """Get recent activity across the knowledge base.
 
@@ -52,6 +53,7 @@ async def recent_activity(
         page: Page number of results to return (default: 1)
         page_size: Number of results to return per page (default: 10)
         max_related: Maximum number of related results to return (default: 10)
+        project: Optional project name to get activity from. If not provided, uses current active project.
 
     Returns:
         GraphContext containing:
@@ -75,6 +77,9 @@ async def recent_activity(
         # Look back further with more context
         recent_activity(type="entity", depth=2, timeframe="2 weeks ago")
 
+        # Get activity from specific project
+        recent_activity(type="entity", project="work-project")
+
     Notes:
         - Higher depth values (>3) may impact performance with large result sets
         - For focused queries, consider using build_context with a specific URI
@@ -115,7 +120,8 @@ async def recent_activity(
         # Add validated types to params
         params["type"] = [t.value for t in validated_types]  # pyright: ignore
 
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     response = await call_get(
         client,

src/basic_memory/mcp/tools/search.py

@@ -4,10 +4,10 @@ from typing import List, Optional
 
 from loguru import logger
 
-from basic_memory.config import get_project_config
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_post
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.search import SearchItemType, SearchQuery, SearchResponse
 
 
@@ -22,6 +22,7 @@ async def search_notes(
     types: Optional[List[str]] = None,
     entity_types: Optional[List[str]] = None,
     after_date: Optional[str] = None,
+    project: Optional[str] = None,
 ) -> SearchResponse:
     """Search across all content in the knowledge base.
 
@@ -37,6 +38,7 @@ async def search_notes(
         types: Optional list of note types to search (e.g., ["note", "person"])
         entity_types: Optional list of entity types to filter by (e.g., ["entity", "observation"])
         after_date: Optional date filter for recent content (e.g., "1 week", "2d")
+        project: Optional project name to search in. If not provided, uses current active project.
 
     Returns:
         SearchResponse with results and pagination info
@@ -80,6 +82,9 @@ async def search_notes(
             query="docs/meeting-*",
             search_type="permalink"
         )
+
+        # Search in specific project
+        results = await search_notes("meeting notes", project="work-project")
     """
     # Create a SearchQuery object based on the parameters
     search_query = SearchQuery()
@@ -104,7 +109,8 @@ async def search_notes(
     if after_date:
         search_query.after_date = after_date
 
-    project_url = get_project_config().project_url
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     logger.info(f"Searching for {search_query}")
     response = await call_post(

src/basic_memory/mcp/tools/utils.py

@@ -5,6 +5,7 @@ to the Basic Memory API, with improved error handling and logging.
 """
 
 import typing
+from typing import Optional
 
 from httpx import Response, URL, AsyncClient, HTTPStatusError
 from httpx._client import UseClientDefault, USE_CLIENT_DEFAULT
@@ -23,7 +24,9 @@ from loguru import logger
 from mcp.server.fastmcp.exceptions import ToolError
 
 
-def get_error_message(status_code: int, url: URL | str, method: str) -> str:
+def get_error_message(
+    status_code: int, url: URL | str, method: str, msg: Optional[str] = None
+) -> str:
     """Get a friendly error message based on the HTTP status code.
 
     Args:
@@ -103,6 +106,7 @@ async def call_get(
         ToolError: If the request fails with an appropriate error message
     """
     logger.debug(f"Calling GET '{url}' params: '{params}'")
+    error_message = None
     try:
         response = await client.get(
             url,
@@ -120,7 +124,12 @@ async def call_get(
 
         # Handle different status codes differently
         status_code = response.status_code
-        error_message = get_error_message(status_code, url, "GET")
+        # get the message if available
+        response_data = response.json()
+        if isinstance(response_data, dict) and "detail" in response_data:
+            error_message = response_data["detail"]
+        else:
+            error_message = get_error_message(status_code, url, "PUT")
 
         # Log at appropriate level based on status code
         if 400 <= status_code < 500:
@@ -138,8 +147,6 @@ async def call_get(
         return response  # This line will never execute, but it satisfies the type checker  # pragma: no cover
 
     except HTTPStatusError as e:
-        status_code = e.response.status_code
-        error_message = get_error_message(status_code, url, "GET")
         raise ToolError(error_message) from e
 
 
@@ -183,6 +190,8 @@ async def call_put(
         ToolError: If the request fails with an appropriate error message
     """
     logger.debug(f"Calling PUT '{url}'")
+    error_message = None
+
     try:
         response = await client.put(
             url,
@@ -204,7 +213,13 @@ async def call_put(
 
         # Handle different status codes differently
         status_code = response.status_code
-        error_message = get_error_message(status_code, url, "PUT")
+
+        # get the message if available
+        response_data = response.json()
+        if isinstance(response_data, dict) and "detail" in response_data:
+            error_message = response_data["detail"]  # pragma: no cover
+        else:
+            error_message = get_error_message(status_code, url, "PUT")
 
         # Log at appropriate level based on status code
         if 400 <= status_code < 500:
@@ -221,9 +236,110 @@ async def call_put(
         response.raise_for_status()  # Will always raise since we're in the error case
         return response  # This line will never execute, but it satisfies the type checker  # pragma: no cover
 
+    except HTTPStatusError as e:
+        raise ToolError(error_message) from e
+
+
+async def call_patch(
+    client: AsyncClient,
+    url: URL | str,
+    *,
+    content: RequestContent | None = None,
+    data: RequestData | None = None,
+    files: RequestFiles | None = None,
+    json: typing.Any | None = None,
+    params: QueryParamTypes | None = None,
+    headers: HeaderTypes | None = None,
+    cookies: CookieTypes | None = None,
+    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
+    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    extensions: RequestExtensions | None = None,
+) -> Response:
+    """Make a PATCH request and handle errors appropriately.
+
+    Args:
+        client: The HTTPX AsyncClient to use
+        url: The URL to request
+        content: Request content
+        data: Form data
+        files: Files to upload
+        json: JSON data
+        params: Query parameters
+        headers: HTTP headers
+        cookies: HTTP cookies
+        auth: Authentication
+        follow_redirects: Whether to follow redirects
+        timeout: Request timeout
+        extensions: HTTPX extensions
+
+    Returns:
+        The HTTP response
+
+    Raises:
+        ToolError: If the request fails with an appropriate error message
+    """
+    logger.debug(f"Calling PATCH '{url}'")
+    try:
+        response = await client.patch(
+            url,
+            content=content,
+            data=data,
+            files=files,
+            json=json,
+            params=params,
+            headers=headers,
+            cookies=cookies,
+            auth=auth,
+            follow_redirects=follow_redirects,
+            timeout=timeout,
+            extensions=extensions,
+        )
+
+        if response.is_success:
+            return response
+
+        # Handle different status codes differently
+        status_code = response.status_code
+
+        # Try to extract specific error message from response body
+        try:
+            response_data = response.json()
+            if isinstance(response_data, dict) and "detail" in response_data:
+                error_message = response_data["detail"]
+            else:
+                error_message = get_error_message(status_code, url, "PATCH")  # pragma: no cover
+        except Exception:  # pragma: no cover
+            error_message = get_error_message(status_code, url, "PATCH")  # pragma: no cover
+
+        # Log at appropriate level based on status code
+        if 400 <= status_code < 500:
+            # Client errors: log as info except for 429 (Too Many Requests)
+            if status_code == 429:  # pragma: no cover
+                logger.warning(f"Rate limit exceeded: PATCH {url}: {error_message}")
+            else:
+                logger.info(f"Client error: PATCH {url}: {error_message}")
+        else:  # pragma: no cover
+            # Server errors: log as error
+            logger.error(f"Server error: PATCH {url}: {error_message}")  # pragma: no cover
+
+        # Raise a tool error with the friendly message
+        response.raise_for_status()  # Will always raise since we're in the error case
+        return response  # This line will never execute, but it satisfies the type checker  # pragma: no cover
+
     except HTTPStatusError as e:
         status_code = e.response.status_code
-        error_message = get_error_message(status_code, url, "PUT")
+
+        # Try to extract specific error message from response body
+        try:
+            response_data = e.response.json()
+            if isinstance(response_data, dict) and "detail" in response_data:
+                error_message = response_data["detail"]
+            else:
+                error_message = get_error_message(status_code, url, "PATCH")  # pragma: no cover
+        except Exception:  # pragma: no cover
+            error_message = get_error_message(status_code, url, "PATCH")  # pragma: no cover
+
         raise ToolError(error_message) from e
 
 
@@ -267,6 +383,7 @@ async def call_post(
         ToolError: If the request fails with an appropriate error message
     """
     logger.debug(f"Calling POST '{url}'")
+    error_message = None
     try:
         response = await client.post(
             url=url,
@@ -289,7 +406,12 @@ async def call_post(
 
         # Handle different status codes differently
         status_code = response.status_code
-        error_message = get_error_message(status_code, url, "POST")
+        # get the message if available
+        response_data = response.json()
+        if isinstance(response_data, dict) and "detail" in response_data:
+            error_message = response_data["detail"]
+        else:
+            error_message = get_error_message(status_code, url, "POST")
 
         # Log at appropriate level based on status code
         if 400 <= status_code < 500:
@@ -307,8 +429,6 @@ async def call_post(
         return response  # This line will never execute, but it satisfies the type checker  # pragma: no cover
 
     except HTTPStatusError as e:
-        status_code = e.response.status_code
-        error_message = get_error_message(status_code, url, "POST")
         raise ToolError(error_message) from e
 
 
@@ -344,6 +464,7 @@ async def call_delete(
         ToolError: If the request fails with an appropriate error message
     """
     logger.debug(f"Calling DELETE '{url}'")
+    error_message = None
     try:
         response = await client.delete(
             url=url,
@@ -361,7 +482,12 @@ async def call_delete(
 
         # Handle different status codes differently
         status_code = response.status_code
-        error_message = get_error_message(status_code, url, "DELETE")
+        # get the message if available
+        response_data = response.json()
+        if isinstance(response_data, dict) and "detail" in response_data:
+            error_message = response_data["detail"]  # pragma: no cover
+        else:
+            error_message = get_error_message(status_code, url, "DELETE")
 
         # Log at appropriate level based on status code
         if 400 <= status_code < 500:
@@ -379,6 +505,4 @@ async def call_delete(
         return response  # This line will never execute, but it satisfies the type checker  # pragma: no cover
 
     except HTTPStatusError as e:
-        status_code = e.response.status_code
-        error_message = get_error_message(status_code, url, "DELETE")
         raise ToolError(error_message) from e

src/basic_memory/mcp/tools/write_note.py

@@ -1,16 +1,16 @@
 """Write note tool for Basic Memory MCP server."""
 
-from typing import List, Union
+from typing import List, Union, Optional
 
 from loguru import logger
 
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_put
+from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas import EntityResponse
 from basic_memory.schemas.base import Entity
 from basic_memory.utils import parse_tags
-from basic_memory.config import get_project_config
 
 # Define TagType as a Union that can accept either a string or a list of strings or None
 TagType = Union[List[str], str, None]
@@ -27,6 +27,7 @@ async def write_note(
     content: str,
     folder: str,
     tags=None,  # Remove type hint completely to avoid schema issues
+    project: Optional[str] = None,
 ) -> str:
     """Write a markdown note to the knowledge base.
 
@@ -56,6 +57,7 @@ async def write_note(
         folder: the folder where the file should be saved
         tags: Tags to categorize the note. Can be a list of strings, a comma-separated string, or None.
               Note: If passing from external MCP clients, use a string format (e.g. "tag1,tag2,tag3")
+        project: Optional project name to write to. If not provided, uses current active project.
 
     Returns:
         A markdown formatted summary of the semantic content, including:
@@ -65,12 +67,12 @@ async def write_note(
         - Relation counts (resolved/unresolved)
         - Tags if present
     """
-    logger.info("MCP tool call", tool="write_note", folder=folder, title=title, tags=tags)
+    logger.info(f"MCP tool call tool=write_note folder={folder}, title={title}, tags={tags}")
 
     # Process tags using the helper function
     tag_list = parse_tags(tags)
     # Create the entity request
-    metadata = {"tags": [f"#{tag}" for tag in tag_list]} if tag_list else None
+    metadata = {"tags": tag_list} if tag_list else None
     entity = Entity(
         title=title,
         folder=folder,
@@ -79,11 +81,11 @@ async def write_note(
         content=content,
         entity_metadata=metadata,
     )
-    project_url = get_project_config().project_url
-    print(f"project_url: {project_url}")
+    active_project = get_active_project(project)
+    project_url = active_project.project_url
 
     # Create or update via knowledge API
-    logger.debug("Creating entity via API", permalink=entity.permalink)
+    logger.debug(f"Creating entity via API permalink={entity.permalink}")
     url = f"{project_url}/knowledge/entities/{entity.permalink}"
     response = await call_put(client, url, json=entity.model_dump())
     result = EntityResponse.model_validate(response.json())
@@ -125,15 +127,6 @@ async def write_note(
 
     # Log the response with structured data
     logger.info(
-        "MCP tool response",
-        tool="write_note",
-        action=action,
-        permalink=result.permalink,
-        observations_count=len(result.observations),
-        relations_count=len(result.relations),
-        resolved_relations=resolved,
-        unresolved_relations=unresolved,
-        status_code=response.status_code,
+        f"MCP tool response: tool=write_note action={action} permalink={result.permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved} status_code={response.status_code}"
     )
-
     return "\n".join(summary)

src/basic_memory/repository/search_repository.py

@@ -130,26 +130,35 @@ class SearchRepository:
         For FTS5:
         - Special characters and phrases need to be quoted
         - Terms with spaces or special chars need quotes
-        - Boolean operators (AND, OR, NOT) and parentheses are preserved
+        - Boolean operators (AND, OR, NOT) are preserved for complex queries
         """
         if "*" in term:
             return term
 
-        # Check for boolean operators - if present, return the term as is
-        boolean_operators = [" AND ", " OR ", " NOT ", "(", ")"]
+        # Check for explicit boolean operators - if present, return the term as is
+        boolean_operators = [" AND ", " OR ", " NOT "]
         if any(op in f" {term} " for op in boolean_operators):
             return term
 
-        # List of special characters that need quoting (excluding *)
+        # List of FTS5 special characters that need escaping/quoting
         special_chars = ["/", "-", ".", " ", "(", ")", "[", "]", '"', "'"]
 
         # Check if term contains any special characters
         needs_quotes = any(c in term for c in special_chars)
 
         if needs_quotes:
-            # If the term already contains quotes, escape them and add a wildcard
-            term = term.replace('"', '""')
-            term = f'"{term}"*'
+            # Escape any existing quotes by doubling them
+            escaped_term = term.replace('"', '""')
+            # Quote the entire term to handle special characters safely
+            if is_prefix and not ("/" in term and term.endswith(".md")):
+                # For search terms (not file paths), add prefix matching
+                term = f'"{escaped_term}"*'
+            else:
+                # For file paths, use exact matching
+                term = f'"{escaped_term}"'
+        elif is_prefix:
+            # Only add wildcard for simple terms without special characters
+            term = f"{term}*"
 
         return term
 
@@ -172,9 +181,8 @@ class SearchRepository:
 
         # Handle text search for title and content
         if search_text:
-            has_boolean = any(
-                op in f" {search_text} " for op in [" AND ", " OR ", " NOT ", "(", ")"]
-            )
+            # Check for explicit boolean operators - only detect them in proper boolean contexts
+            has_boolean = any(op in f" {search_text} " for op in [" AND ", " OR ", " NOT "])
 
             if has_boolean:
                 # If boolean operators are present, use the raw query
@@ -189,9 +197,9 @@ class SearchRepository:
 
         # Handle title match search
         if title:
-            title_text = self._prepare_search_term(title.strip())
-            params["text"] = title_text
-            conditions.append("title MATCH :text")
+            title_text = self._prepare_search_term(title.strip(), is_prefix=False)
+            params["title_text"] = title_text
+            conditions.append("title MATCH :title_text")
 
         # Handle permalink exact search
         if permalink:

src/basic_memory/schemas/project_info.py

@@ -107,7 +107,7 @@ class ProjectInfoResponse(BaseModel):
     system: SystemStatus = Field(description="System and service status information")
 
 
-class ProjectSwitchRequest(BaseModel):
+class ProjectInfoRequest(BaseModel):
     """Request model for switching projects."""
 
     name: str = Field(..., description="Name of the project to switch to")
@@ -177,27 +177,12 @@ class ProjectWatchStatus(BaseModel):
     )
 
 
-class ProjectStatusResponse(BaseModel):
-    """Response model for switching projects."""
-
-    message: str = Field(..., description="Status message about the project switch")
-    status: str = Field(..., description="Status of the switch (success or error)")
-    default: bool = Field(..., description="True if the project was set as the default")
-    old_project: Optional[ProjectWatchStatus] = Field(
-        None, description="Information about the project being switched from"
-    )
-    new_project: Optional[ProjectWatchStatus] = Field(
-        None, description="Information about the project being switched to"
-    )
-
-
 class ProjectItem(BaseModel):
     """Simple representation of a project."""
 
     name: str
     path: str
-    is_default: bool
-    is_current: bool
+    is_default: bool = False
 
 
 class ProjectList(BaseModel):
@@ -205,4 +190,17 @@ class ProjectList(BaseModel):
 
     projects: List[ProjectItem]
     default_project: str
-    current_project: str
+
+
+class ProjectStatusResponse(BaseModel):
+    """Response model for switching projects."""
+
+    message: str = Field(..., description="Status message about the project switch")
+    status: str = Field(..., description="Status of the switch (success or error)")
+    default: bool = Field(..., description="True if the project was set as the default")
+    old_project: Optional[ProjectItem] = Field(
+        None, description="Information about the project being switched from"
+    )
+    new_project: Optional[ProjectItem] = Field(
+        None, description="Information about the project being switched to"
+    )

src/basic_memory/schemas/request.py

@@ -1,9 +1,9 @@
 """Request schemas for interacting with the knowledge graph."""
 
-from typing import List, Optional, Annotated
+from typing import List, Optional, Annotated, Literal
 from annotated_types import MaxLen, MinLen
 
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
 
 from basic_memory.schemas.base import (
     Relation,
@@ -56,3 +56,57 @@ class GetEntitiesRequest(BaseModel):
 
 class CreateRelationsRequest(BaseModel):
     relations: List[Relation]
+
+
+class EditEntityRequest(BaseModel):
+    """Request schema for editing an existing entity's content.
+
+    This allows for targeted edits without requiring the full entity content.
+    Supports various operation types for different editing scenarios.
+    """
+
+    operation: Literal["append", "prepend", "find_replace", "replace_section"]
+    content: str
+    section: Optional[str] = None
+    find_text: Optional[str] = None
+    expected_replacements: int = 1
+
+    @field_validator("section")
+    @classmethod
+    def validate_section_for_replace_section(cls, v, info):
+        """Ensure section is provided for replace_section operation."""
+        if info.data.get("operation") == "replace_section" and not v:
+            raise ValueError("section parameter is required for replace_section operation")
+        return v
+
+    @field_validator("find_text")
+    @classmethod
+    def validate_find_text_for_find_replace(cls, v, info):
+        """Ensure find_text is provided for find_replace operation."""
+        if info.data.get("operation") == "find_replace" and not v:
+            raise ValueError("find_text parameter is required for find_replace operation")
+        return v
+
+
+class MoveEntityRequest(BaseModel):
+    """Request schema for moving an entity to a new file location.
+
+    This allows moving notes to different paths while maintaining project
+    consistency and optionally updating permalinks based on configuration.
+    """
+
+    identifier: Annotated[str, MinLen(1), MaxLen(200)]
+    destination_path: Annotated[str, MinLen(1), MaxLen(500)]
+    project: Optional[str] = None
+
+    @field_validator("destination_path")
+    @classmethod
+    def validate_destination_path(cls, v):
+        """Ensure destination path is relative and valid."""
+        if v.startswith("/"):
+            raise ValueError("destination_path must be relative, not absolute")
+        if ".." in v:
+            raise ValueError("destination_path cannot contain '..' path components")
+        if not v.strip():
+            raise ValueError("destination_path cannot be empty or whitespace only")
+        return v.strip()

src/basic_memory/services/directory_service.py

@@ -1,8 +1,9 @@
 """Directory service for managing file directories and tree structure."""
 
+import fnmatch
 import logging
 import os
-from typing import Dict
+from typing import Dict, List, Optional
 
 from basic_memory.repository import EntityRepository
 from basic_memory.schemas.directory import DirectoryNode
@@ -87,3 +88,80 @@ class DirectoryService:
 
         # Return the root node with its children
         return root_node
+
+    async def list_directory(
+        self,
+        dir_name: str = "/",
+        depth: int = 1,
+        file_name_glob: Optional[str] = None,
+    ) -> List[DirectoryNode]:
+        """List directory contents with filtering and depth control.
+
+        Args:
+            dir_name: Directory path to list (default: root "/")
+            depth: Recursion depth (1 = immediate children only)
+            file_name_glob: Glob pattern for filtering file names
+
+        Returns:
+            List of DirectoryNode objects matching the criteria
+        """
+        # Normalize directory path
+        if not dir_name.startswith("/"):
+            dir_name = f"/{dir_name}"
+        if dir_name != "/" and dir_name.endswith("/"):
+            dir_name = dir_name.rstrip("/")
+
+        # Get the full directory tree
+        root_tree = await self.get_directory_tree()
+
+        # Find the target directory node
+        target_node = self._find_directory_node(root_tree, dir_name)
+        if not target_node:
+            return []
+
+        # Collect nodes with depth and glob filtering
+        result = []
+        self._collect_nodes_recursive(target_node, result, depth, file_name_glob, 0)
+
+        return result
+
+    def _find_directory_node(
+        self, root: DirectoryNode, target_path: str
+    ) -> Optional[DirectoryNode]:
+        """Find a directory node by path in the tree."""
+        if root.directory_path == target_path:
+            return root
+
+        for child in root.children:
+            if child.type == "directory":
+                found = self._find_directory_node(child, target_path)
+                if found:
+                    return found
+
+        return None
+
+    def _collect_nodes_recursive(
+        self,
+        node: DirectoryNode,
+        result: List[DirectoryNode],
+        max_depth: int,
+        file_name_glob: Optional[str],
+        current_depth: int,
+    ) -> None:
+        """Recursively collect nodes with depth and glob filtering."""
+        if current_depth >= max_depth:
+            return
+
+        for child in node.children:
+            # Apply glob filtering
+            if file_name_glob and not fnmatch.fnmatch(child.name, file_name_glob):
+                continue
+
+            # Add the child to results
+            result.append(child)
+
+            # Recurse into subdirectories if we haven't reached max depth
+            if child.type == "directory" and current_depth < max_depth:
+                self._collect_nodes_recursive(
+                    child, result, max_depth, file_name_glob, current_depth + 1
+                )

src/basic_memory/services/entity_service.py

@@ -4,9 +4,12 @@ from pathlib import Path
 from typing import List, Optional, Sequence, Tuple, Union
 
 import frontmatter
+import yaml
 from loguru import logger
 from sqlalchemy.exc import IntegrityError
 
+from basic_memory.config import ProjectConfig, BasicMemoryConfig
+from basic_memory.file_utils import has_frontmatter, parse_frontmatter, remove_frontmatter
 from basic_memory.markdown import EntityMarkdown
 from basic_memory.markdown.entity_parser import EntityParser
 from basic_memory.markdown.utils import entity_model_from_markdown, schema_to_markdown
@@ -114,8 +117,29 @@ class EntityService(BaseService[EntityModel]):
                 f"file for entity {schema.folder}/{schema.title} already exists: {file_path}"
             )
 
-        # Get unique permalink
-        permalink = await self.resolve_permalink(schema.permalink or file_path)
+        # Parse content frontmatter to check for user-specified permalink
+        content_markdown = None
+        if schema.content and has_frontmatter(schema.content):
+            content_frontmatter = parse_frontmatter(schema.content)
+            if "permalink" in content_frontmatter:
+                # Create a minimal EntityMarkdown object for permalink resolution
+                from basic_memory.markdown.schemas import EntityFrontmatter
+
+                frontmatter_metadata = {
+                    "title": schema.title,
+                    "type": schema.entity_type,
+                    "permalink": content_frontmatter["permalink"],
+                }
+                frontmatter_obj = EntityFrontmatter(metadata=frontmatter_metadata)
+                content_markdown = EntityMarkdown(
+                    frontmatter=frontmatter_obj,
+                    content="",  # content not needed for permalink resolution
+                    observations=[],
+                    relations=[],
+                )
+
+        # Get unique permalink (prioritizing content frontmatter)
+        permalink = await self.resolve_permalink(file_path, content_markdown)
         schema._permalink = permalink
 
         post = await schema_to_markdown(schema)
@@ -148,12 +172,47 @@ class EntityService(BaseService[EntityModel]):
         # Read existing frontmatter from the file if it exists
         existing_markdown = await self.entity_parser.parse_file(file_path)
 
+        # Parse content frontmatter to check for user-specified permalink
+        content_markdown = None
+        if schema.content and has_frontmatter(schema.content):
+            content_frontmatter = parse_frontmatter(schema.content)
+            if "permalink" in content_frontmatter:
+                # Create a minimal EntityMarkdown object for permalink resolution
+                from basic_memory.markdown.schemas import EntityFrontmatter
+
+                frontmatter_metadata = {
+                    "title": schema.title,
+                    "type": schema.entity_type,
+                    "permalink": content_frontmatter["permalink"],
+                }
+                frontmatter_obj = EntityFrontmatter(metadata=frontmatter_metadata)
+                content_markdown = EntityMarkdown(
+                    frontmatter=frontmatter_obj,
+                    content="",  # content not needed for permalink resolution
+                    observations=[],
+                    relations=[],
+                )
+
+        # Check if we need to update the permalink based on content frontmatter
+        new_permalink = entity.permalink  # Default to existing
+        if content_markdown and content_markdown.frontmatter.permalink:
+            # Resolve permalink with the new content frontmatter
+            resolved_permalink = await self.resolve_permalink(file_path, content_markdown)
+            if resolved_permalink != entity.permalink:
+                new_permalink = resolved_permalink
+                # Update the schema to use the new permalink
+                schema._permalink = new_permalink
+
         # Create post with new content from schema
         post = await schema_to_markdown(schema)
 
         # Merge new metadata with existing metadata
         existing_markdown.frontmatter.metadata.update(post.metadata)
 
+        # Ensure the permalink in the metadata is the resolved one
+        if new_permalink != entity.permalink:
+            existing_markdown.frontmatter.metadata["permalink"] = new_permalink
+
         # Create a new post with merged metadata
         merged_post = frontmatter.Post(post.content, **existing_markdown.frontmatter.metadata)
 
@@ -325,3 +384,319 @@ class EntityService(BaseService[EntityModel]):
                 continue
 
         return await self.repository.get_by_file_path(path)
+
+    async def edit_entity(
+        self,
+        identifier: str,
+        operation: str,
+        content: str,
+        section: Optional[str] = None,
+        find_text: Optional[str] = None,
+        expected_replacements: int = 1,
+    ) -> EntityModel:
+        """Edit an existing entity's content using various operations.
+
+        Args:
+            identifier: Entity identifier (permalink, title, etc.)
+            operation: The editing operation (append, prepend, find_replace, replace_section)
+            content: The content to add or use for replacement
+            section: For replace_section operation - the markdown header
+            find_text: For find_replace operation - the text to find and replace
+            expected_replacements: For find_replace operation - expected number of replacements (default: 1)
+
+        Returns:
+            The updated entity model
+
+        Raises:
+            EntityNotFoundError: If the entity cannot be found
+            ValueError: If required parameters are missing for the operation or replacement count doesn't match expected
+        """
+        logger.debug(f"Editing entity: {identifier}, operation: {operation}")
+
+        # Find the entity using the link resolver
+        entity = await self.link_resolver.resolve_link(identifier)
+        if not entity:
+            raise EntityNotFoundError(f"Entity not found: {identifier}")
+
+        # Read the current file content
+        file_path = Path(entity.file_path)
+        current_content, _ = await self.file_service.read_file(file_path)
+
+        # Apply the edit operation
+        new_content = self.apply_edit_operation(
+            current_content, operation, content, section, find_text, expected_replacements
+        )
+
+        # Write the updated content back to the file
+        checksum = await self.file_service.write_file(file_path, new_content)
+
+        # Parse the updated file to get new observations/relations
+        entity_markdown = await self.entity_parser.parse_file(file_path)
+
+        # Update entity and its relationships
+        entity = await self.update_entity_and_observations(file_path, entity_markdown)
+        await self.update_entity_relations(str(file_path), entity_markdown)
+
+        # Set final checksum to match file
+        entity = await self.repository.update(entity.id, {"checksum": checksum})
+
+        return entity
+
+    def apply_edit_operation(
+        self,
+        current_content: str,
+        operation: str,
+        content: str,
+        section: Optional[str] = None,
+        find_text: Optional[str] = None,
+        expected_replacements: int = 1,
+    ) -> str:
+        """Apply the specified edit operation to the current content."""
+
+        if operation == "append":
+            # Ensure proper spacing
+            if current_content and not current_content.endswith("\n"):
+                return current_content + "\n" + content
+            return current_content + content  # pragma: no cover
+
+        elif operation == "prepend":
+            # Handle frontmatter-aware prepending
+            return self._prepend_after_frontmatter(current_content, content)
+
+        elif operation == "find_replace":
+            if not find_text:
+                raise ValueError("find_text is required for find_replace operation")
+            if not find_text.strip():
+                raise ValueError("find_text cannot be empty or whitespace only")
+
+            # Count actual occurrences
+            actual_count = current_content.count(find_text)
+
+            # Validate count matches expected
+            if actual_count != expected_replacements:
+                if actual_count == 0:
+                    raise ValueError(f"Text to replace not found: '{find_text}'")
+                else:
+                    raise ValueError(
+                        f"Expected {expected_replacements} occurrences of '{find_text}', "
+                        f"but found {actual_count}"
+                    )
+
+            return current_content.replace(find_text, content)
+
+        elif operation == "replace_section":
+            if not section:
+                raise ValueError("section is required for replace_section operation")
+            if not section.strip():
+                raise ValueError("section cannot be empty or whitespace only")
+            return self.replace_section_content(current_content, section, content)
+
+        else:
+            raise ValueError(f"Unsupported operation: {operation}")
+
+    def replace_section_content(
+        self, current_content: str, section_header: str, new_content: str
+    ) -> str:
+        """Replace content under a specific markdown section header.
+
+        This method uses a simple, safe approach: when replacing a section, it only
+        replaces the immediate content under that header until it encounters the next
+        header of ANY level. This means:
+
+        - Replacing "# Header" replaces content until "## Subsection" (preserves subsections)
+        - Replacing "## Section" replaces content until "### Subsection" (preserves subsections)
+        - More predictable and safer than trying to consume entire hierarchies
+
+        Args:
+            current_content: The current markdown content
+            section_header: The section header to find and replace (e.g., "## Section Name")
+            new_content: The new content to replace the section with
+
+        Returns:
+            The updated content with the section replaced
+
+        Raises:
+            ValueError: If multiple sections with the same header are found
+        """
+        # Normalize the section header (ensure it starts with #)
+        if not section_header.startswith("#"):
+            section_header = "## " + section_header
+
+        # First pass: count matching sections to check for duplicates
+        lines = current_content.split("\n")
+        matching_sections = []
+
+        for i, line in enumerate(lines):
+            if line.strip() == section_header.strip():
+                matching_sections.append(i)
+
+        # Handle multiple sections error
+        if len(matching_sections) > 1:
+            raise ValueError(
+                f"Multiple sections found with header '{section_header}'. "
+                f"Section replacement requires unique headers."
+            )
+
+        # If no section found, append it
+        if len(matching_sections) == 0:
+            logger.info(f"Section '{section_header}' not found, appending to end of document")
+            separator = "\n\n" if current_content and not current_content.endswith("\n\n") else ""
+            return current_content + separator + section_header + "\n" + new_content
+
+        # Replace the single matching section
+        result_lines = []
+        section_line_idx = matching_sections[0]
+
+        i = 0
+        while i < len(lines):
+            line = lines[i]
+
+            # Check if this is our target section header
+            if i == section_line_idx:
+                # Add the section header and new content
+                result_lines.append(line)
+                result_lines.append(new_content)
+                i += 1
+
+                # Skip the original section content until next header or end
+                while i < len(lines):
+                    next_line = lines[i]
+                    # Stop consuming when we hit any header (preserve subsections)
+                    if next_line.startswith("#"):
+                        # We found another header - continue processing from here
+                        break
+                    i += 1
+                # Continue processing from the next header (don't increment i again)
+                continue
+
+            # Add all other lines (including subsequent sections)
+            result_lines.append(line)
+            i += 1
+
+        return "\n".join(result_lines)
+
+    def _prepend_after_frontmatter(self, current_content: str, content: str) -> str:
+        """Prepend content after frontmatter, preserving frontmatter structure."""
+
+        # Check if file has frontmatter
+        if has_frontmatter(current_content):
+            try:
+                # Parse and separate frontmatter from body
+                frontmatter_data = parse_frontmatter(current_content)
+                body_content = remove_frontmatter(current_content)
+
+                # Prepend content to the body
+                if content and not content.endswith("\n"):
+                    new_body = content + "\n" + body_content
+                else:
+                    new_body = content + body_content
+
+                # Reconstruct file with frontmatter + prepended body
+                yaml_fm = yaml.dump(frontmatter_data, sort_keys=False, allow_unicode=True)
+                return f"---\n{yaml_fm}---\n\n{new_body.strip()}"
+
+            except Exception as e:  # pragma: no cover
+                logger.warning(
+                    f"Failed to parse frontmatter during prepend: {e}"
+                )  # pragma: no cover
+                # Fall back to simple prepend if frontmatter parsing fails  # pragma: no cover
+
+        # No frontmatter or parsing failed - do simple prepend  # pragma: no cover
+        if content and not content.endswith("\n"):  # pragma: no cover
+            return content + "\n" + current_content  # pragma: no cover
+        return content + current_content  # pragma: no cover
+
+    async def move_entity(
+        self,
+        identifier: str,
+        destination_path: str,
+        project_config: ProjectConfig,
+        app_config: BasicMemoryConfig,
+    ) -> EntityModel:
+        """Move entity to new location with database consistency.
+
+        Args:
+            identifier: Entity identifier (title, permalink, or memory:// URL)
+            destination_path: New path relative to project root
+            project_config: Project configuration for file operations
+            app_config: App configuration for permalink update settings
+
+        Returns:
+            Success message with move details
+
+        Raises:
+            EntityNotFoundError: If the entity cannot be found
+            ValueError: If move operation fails due to validation or filesystem errors
+        """
+        logger.debug(f"Moving entity: {identifier} to {destination_path}")
+
+        # 1. Resolve identifier to entity
+        entity = await self.link_resolver.resolve_link(identifier)
+        if not entity:
+            raise EntityNotFoundError(f"Entity not found: {identifier}")
+
+        current_path = entity.file_path
+        old_permalink = entity.permalink
+
+        # 2. Validate destination path format first
+        if not destination_path or destination_path.startswith("/") or not destination_path.strip():
+            raise ValueError(f"Invalid destination path: {destination_path}")
+
+        # 3. Validate paths
+        source_file = project_config.home / current_path
+        destination_file = project_config.home / destination_path
+
+        # Validate source exists
+        if not source_file.exists():
+            raise ValueError(f"Source file not found: {current_path}")
+
+        # Check if destination already exists
+        if destination_file.exists():
+            raise ValueError(f"Destination already exists: {destination_path}")
+
+        try:
+            # 4. Create destination directory if needed
+            destination_file.parent.mkdir(parents=True, exist_ok=True)
+
+            # 5. Move physical file
+            source_file.rename(destination_file)
+            logger.info(f"Moved file: {current_path} -> {destination_path}")
+
+            # 6. Prepare database updates
+            updates = {"file_path": destination_path}
+
+            # 7. Update permalink if configured
+            if app_config.update_permalinks_on_move:
+                # Generate new permalink from destination path
+                new_permalink = await self.resolve_permalink(destination_path)
+
+                # Update frontmatter with new permalink
+                await self.file_service.update_frontmatter(
+                    destination_path, {"permalink": new_permalink}
+                )
+
+                updates["permalink"] = new_permalink
+                logger.info(f"Updated permalink: {old_permalink} -> {new_permalink}")
+
+            # 8. Recalculate checksum
+            new_checksum = await self.file_service.compute_checksum(destination_path)
+            updates["checksum"] = new_checksum
+
+            # 9. Update database
+            updated_entity = await self.repository.update(entity.id, updates)
+            if not updated_entity:
+                raise ValueError(f"Failed to update entity in database: {entity.id}")
+
+            return updated_entity
+
+        except Exception as e:
+            # Rollback: try to restore original file location if move succeeded
+            if destination_file.exists() and not source_file.exists():
+                try:
+                    destination_file.rename(source_file)
+                    logger.info(f"Rolled back file move: {destination_path} -> {current_path}")
+                except Exception as rollback_error:  # pragma: no cover
+                    logger.error(f"Failed to rollback file move: {rollback_error}")
+
+            # Re-raise the original error with context
+            raise ValueError(f"Move failed: {str(e)}") from e

src/basic_memory/services/file_service.py

@@ -94,8 +94,8 @@ class FileService:
         """
         try:
             # Convert string to Path if needed
-            path_obj = Path(path) if isinstance(path, str) else path
-
+            path_obj = self.base_path / path if isinstance(path, str) else path
+            logger.debug(f"Checking file existence: path={path_obj}")
             if path_obj.is_absolute():
                 return path_obj.exists()
             else:
@@ -121,7 +121,7 @@ class FileService:
             FileOperationError: If write fails
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
 
         try:
@@ -140,7 +140,7 @@ class FileService:
 
             # Compute and return checksum
             checksum = await file_utils.compute_checksum(content)
-            logger.debug("File write completed", path=str(full_path), checksum=checksum)
+            logger.debug(f"File write completed path={full_path}, {checksum=}")
             return checksum
 
         except Exception as e:
@@ -164,7 +164,7 @@ class FileService:
             FileOperationError: If read fails
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
 
         try:
@@ -194,7 +194,7 @@ class FileService:
             path: Path to delete (Path or string)
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         full_path.unlink(missing_ok=True)
 
@@ -210,7 +210,7 @@ class FileService:
             Checksum of updated file
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         return await file_utils.update_frontmatter(full_path, updates)
 
@@ -227,7 +227,7 @@ class FileService:
             FileError: If checksum computation fails
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
 
         try:
@@ -253,7 +253,7 @@ class FileService:
             File statistics
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         # get file timestamps
         return full_path.stat()
@@ -268,7 +268,7 @@ class FileService:
             MIME type of the file
         """
         # Convert string to Path if needed
-        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj = self.base_path / path if isinstance(path, str) else path
         full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         # get file timestamps
         mime_type, _ = mimetypes.guess_type(full_path.name)

src/basic_memory/services/link_resolver.py

@@ -15,10 +15,10 @@ class LinkResolver:
 
     Uses a combination of exact matching and search-based resolution:
     1. Try exact permalink match (fastest)
-    2. Try permalink pattern match (for wildcards)
-    3. Try exact title match
-    4. Fall back to search for fuzzy matching
-    5. Generate new permalink if no match found
+    2. Try exact title match
+    3. Try exact file path match
+    4. Try file path with .md extension (for folder/title patterns)
+    5. Fall back to search for fuzzy matching
     """
 
     def __init__(self, entity_repository: EntityRepository, search_service: SearchService):
@@ -52,11 +52,19 @@ class LinkResolver:
             logger.debug(f"Found entity with path: {found_path.file_path}")
             return found_path
 
+        # 4. Try file path with .md extension if not already present
+        if not clean_text.endswith(".md") and "/" in clean_text:
+            file_path_with_md = f"{clean_text}.md"
+            found_path_md = await self.entity_repository.get_by_file_path(file_path_with_md)
+            if found_path_md:
+                logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
+                return found_path_md
+
         # search if indicated
         if use_search and "*" not in clean_text:
-            # 3. Fall back to search for fuzzy matching on title
+            # 5. Fall back to search for fuzzy matching on title (use text search for prefix matching)
             results = await self.search_service.search(
-                query=SearchQuery(title=clean_text, entity_types=[SearchItemType.ENTITY]),
+                query=SearchQuery(text=clean_text, entity_types=[SearchItemType.ENTITY]),
             )
 
             if results:

src/basic_memory/services/project_service.py

@@ -4,12 +4,13 @@ import json
 import os
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Sequence
 
 from loguru import logger
 from sqlalchemy import text
 
-from basic_memory.config import ConfigManager, config, app_config
+from basic_memory.config import config, app_config
+from basic_memory.models import Project
 from basic_memory.repository.project_repository import ProjectRepository
 from basic_memory.schemas import (
     ActivityMetrics,
@@ -18,15 +19,17 @@ from basic_memory.schemas import (
     SystemStatus,
 )
 from basic_memory.config import WATCH_STATUS_JSON
-
+from basic_memory.utils import generate_permalink
+from basic_memory.config import config_manager
 
 class ProjectService:
     """Service for managing Basic Memory projects."""
 
-    def __init__(self, repository: Optional[ProjectRepository] = None):
+    repository: ProjectRepository
+
+    def __init__(self, repository: ProjectRepository):
         """Initialize the project service."""
         super().__init__()
-        self.config_manager = ConfigManager()
         self.repository = repository
 
     @property
@@ -36,7 +39,7 @@ class ProjectService:
         Returns:
             Dict mapping project names to their file paths
         """
-        return self.config_manager.projects
+        return config_manager.projects
 
     @property
     def default_project(self) -> str:
@@ -45,7 +48,7 @@ class ProjectService:
         Returns:
             The name of the default project
         """
-        return self.config_manager.default_project
+        return config_manager.default_project
 
     @property
     def current_project(self) -> str:
@@ -54,7 +57,14 @@ class ProjectService:
         Returns:
             The name of the current project
         """
-        return os.environ.get("BASIC_MEMORY_PROJECT", self.config_manager.default_project)
+        return os.environ.get("BASIC_MEMORY_PROJECT", config_manager.default_project)
+
+    async def list_projects(self) -> Sequence[Project]:
+        return await self.repository.find_all()
+
+    async def get_project(self, name: str) -> Optional[Project]:
+        """Get the file path for a project by name."""
+        return await self.repository.get_by_name(name)
 
     async def add_project(self, name: str, path: str) -> None:
         """Add a new project to the configuration and database.
@@ -73,13 +83,13 @@ class ProjectService:
         resolved_path = os.path.abspath(os.path.expanduser(path))
 
         # First add to config file (this will validate the project doesn't exist)
-        self.config_manager.add_project(name, resolved_path)
+        project_config = config_manager.add_project(name, resolved_path)
 
         # Then add to database
         project_data = {
             "name": name,
             "path": resolved_path,
-            "permalink": name.lower().replace(" ", "-"),
+            "permalink": generate_permalink(project_config.name),
             "is_active": True,
             "is_default": False,
         }
@@ -100,7 +110,7 @@ class ProjectService:
             raise ValueError("Repository is required for remove_project")
 
         # First remove from config (this will validate the project exists and is not default)
-        self.config_manager.remove_project(name)
+        config_manager.remove_project(name)
 
         # Then remove from database
         project = await self.repository.get_by_name(name)
@@ -122,7 +132,7 @@ class ProjectService:
             raise ValueError("Repository is required for set_default_project")
 
         # First update config file (this will validate the project exists)
-        self.config_manager.set_default_project(name)
+        config_manager.set_default_project(name)
 
         # Then update database
         project = await self.repository.get_by_name(name)
@@ -150,7 +160,7 @@ class ProjectService:
         db_projects_by_name = {p.name: p for p in db_projects}
 
         # Get all projects from configuration
-        config_projects = self.config_manager.projects
+        config_projects = config_manager.projects
 
         # Add projects that exist in config but not in DB
         for name, path in config_projects.items():
@@ -161,7 +171,7 @@ class ProjectService:
                     "path": path,
                     "permalink": name.lower().replace(" ", "-"),
                     "is_active": True,
-                    "is_default": (name == self.config_manager.default_project),
+                    "is_default": (name == config_manager.default_project),
                 }
                 await self.repository.create(project_data)
 
@@ -169,16 +179,16 @@ class ProjectService:
         for name, project in db_projects_by_name.items():
             if name not in config_projects:
                 logger.info(f"Adding project '{name}' to configuration")
-                self.config_manager.add_project(name, project.path)
+                config_manager.add_project(name, project.path)
 
         # Make sure default project is synchronized
         db_default = next((p for p in db_projects if p.is_default), None)
-        config_default = self.config_manager.default_project
+        config_default = config_manager.default_project
 
         if db_default and db_default.name != config_default:
             # Update config to match DB default
             logger.info(f"Updating default project in config to '{db_default.name}'")
-            self.config_manager.set_default_project(db_default.name)
+            config_manager.set_default_project(db_default.name)
         elif not db_default and config_default in db_projects_by_name:
             # Update DB to match config default
             logger.info(f"Updating default project in database to '{config_default}'")
@@ -204,7 +214,7 @@ class ProjectService:
             raise ValueError("Repository is required for update_project")
 
         # Validate project exists in config
-        if name not in self.config_manager.projects:
+        if name not in config_manager.projects:
             raise ValueError(f"Project '{name}' not found in configuration")
 
         # Get project from database
@@ -218,10 +228,10 @@ class ProjectService:
             resolved_path = os.path.abspath(os.path.expanduser(updated_path))
 
             # Update in config
-            projects = self.config_manager.config.projects.copy()
+            projects = config_manager.config.projects.copy()
             projects[name] = resolved_path
-            self.config_manager.config.projects = projects
-            self.config_manager.save_config(self.config_manager.config)
+            config_manager.config.projects = projects
+            config_manager.save_config(config_manager.config)
 
             # Update in database
             project.path = resolved_path
@@ -242,7 +252,7 @@ class ProjectService:
             if active_projects:
                 new_default = active_projects[0]
                 await self.repository.set_as_default(new_default.id)
-                self.config_manager.set_default_project(new_default.name)
+                config_manager.set_default_project(new_default.name)
                 logger.info(
                     f"Changed default project to '{new_default.name}' as '{name}' was deactivated"
                 )
@@ -274,11 +284,11 @@ class ProjectService:
         db_projects_by_name = {p.name: p for p in db_projects}
 
         # Get default project info
-        default_project = self.config_manager.default_project
+        default_project = config_manager.default_project
 
         # Convert config projects to include database info
         enhanced_projects = {}
-        for name, path in self.config_manager.projects.items():
+        for name, path in config_manager.projects.items():
             db_project = db_projects_by_name.get(name)
             enhanced_projects[name] = {
                 "path": path,
@@ -535,4 +545,4 @@ class ProjectService:
             database_size=db_size_readable,
             watch_status=watch_status,
             timestamp=datetime.now(),
-        )
+        )

src/basic_memory/services/search_service.py

@@ -1,5 +1,6 @@
 """Service for search operations."""
 
+import ast
 from datetime import datetime
 from typing import List, Optional, Set
 
@@ -117,6 +118,38 @@ class SearchService:
 
         return variants
 
+    def _extract_entity_tags(self, entity: Entity) -> List[str]:
+        """Extract tags from entity metadata for search indexing.
+
+        Handles multiple tag formats:
+        - List format: ["tag1", "tag2"]
+        - String format: "['tag1', 'tag2']" or "[tag1, tag2]"
+        - Empty: [] or "[]"
+
+        Returns a list of tag strings for search indexing.
+        """
+        if not entity.entity_metadata or "tags" not in entity.entity_metadata:
+            return []
+
+        tags = entity.entity_metadata["tags"]
+
+        # Handle list format (preferred)
+        if isinstance(tags, list):
+            return [str(tag) for tag in tags if tag]
+
+        # Handle string format (legacy)
+        if isinstance(tags, str):
+            try:
+                # Parse string representation of list
+                parsed_tags = ast.literal_eval(tags)
+                if isinstance(parsed_tags, list):
+                    return [str(tag) for tag in parsed_tags if tag]
+            except (ValueError, SyntaxError):
+                # If parsing fails, treat as single tag
+                return [tags] if tags.strip() else []
+
+        return [] # pragma: no cover 
+
     async def index_entity(
         self,
         entity: Entity,
@@ -201,6 +234,11 @@ class SearchService:
 
         content_stems.extend(self._generate_variants(entity.file_path))
 
+        # Add entity tags from frontmatter to search content
+        entity_tags = self._extract_entity_tags(entity)
+        if entity_tags:
+            content_stems.extend(entity_tags)
+
         entity_content_stems = "\n".join(p for p in content_stems if p and p.strip())
 
         # Index entity
@@ -286,3 +324,32 @@ class SearchService:
     async def delete_by_entity_id(self, entity_id: int):
         """Delete an item from the search index."""
         await self.repository.delete_by_entity_id(entity_id)
+
+    async def handle_delete(self, entity: Entity):
+        """Handle complete entity deletion from search index including observations and relations.
+
+        This replicates the logic from sync_service.handle_delete() to properly clean up
+        all search index entries for an entity and its related data.
+        """
+        logger.debug(
+            f"Cleaning up search index for entity_id={entity.id}, file_path={entity.file_path}, "
+            f"observations={len(entity.observations)}, relations={len(entity.outgoing_relations)}"
+        )
+
+        # Clean up search index - same logic as sync_service.handle_delete()
+        permalinks = (
+            [entity.permalink]
+            + [o.permalink for o in entity.observations]
+            + [r.permalink for r in entity.outgoing_relations]
+        )
+
+        logger.debug(
+            f"Deleting search index entries for entity_id={entity.id}, "
+            f"index_entries={len(permalinks)}"
+        )
+
+        for permalink in permalinks:
+            if permalink:
+                await self.delete_by_permalink(permalink)
+            else:
+                await self.delete_by_entity_id(entity.id)

src/basic_memory/sync/sync_service.py

@@ -379,7 +379,7 @@ class SyncService:
             updates = {"file_path": new_path}
 
             # If configured, also update permalink to match new path
-            if self.app_config.update_permalinks_on_move:
+            if self.app_config.update_permalinks_on_move and self.file_service.is_markdown(new_path):
                 # generate new permalink value
                 new_permalink = await self.entity_service.resolve_permalink(new_path)
 
@@ -505,4 +505,4 @@ class SyncService:
             f"duration_ms={duration_ms}"
         )
 
-        return result
+        return result