feat: CLI refactoring + workspace-aware cloud project listing

Refactor CLI commands to use typed ProjectClient instead of raw HTTP calls,
and add workspace metadata to cloud project listings so users can distinguish
personal vs organization projects.

Key changes:
- 🔧 CLI commands now use ProjectClient typed API clients instead of
  call_get/call_post with manual URL construction
- 🏢 Cloud project listings include workspace_name, workspace_type, and
  workspace_tenant_id for each cloud-sourced project
- Pass config.default_workspace when fetching cloud projects via
  _fetch_cloud_projects() and CLI list_projects
- Add --workspace flag to `bm project list` for explicit workspace override
- Add "Workspace" column to CLI project list table
- Add `bm tool list-projects` and `bm tool list-workspaces` JSON commands
- Comprehensive tests for workspace passthrough, merge behavior, and CLI routing

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/cli/commands/cloud/project_sync.py

@@ -11,7 +11,6 @@ import typer
 from rich.console import Console
 
 from basic_memory.cli.app import cloud_app
-from basic_memory.cli.auth import CLIAuth
 from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
 from basic_memory.cli.commands.cloud.rclone_commands import (
     RcloneError,
@@ -25,8 +24,8 @@ from basic_memory.cli.commands.command_utils import run_with_cleanup
 from basic_memory.cli.commands.routing import force_routing
 from basic_memory.config import ConfigManager, ProjectEntry
 from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.tools.utils import call_get, call_post
-from basic_memory.schemas.project_info import ProjectItem, ProjectList
+from basic_memory.mcp.clients import ProjectClient
+from basic_memory.schemas.project_info import ProjectItem
 from basic_memory.utils import generate_permalink, normalize_project_path
 
 console = Console()
@@ -37,11 +36,9 @@ console = Console()
 
 def _has_cloud_credentials(config) -> bool:
     """Return whether cloud credentials are available (API key or OAuth token)."""
-    if config.cloud_api_key:
-        return True
+    from basic_memory.config import has_cloud_credentials
 
-    auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
-    return auth.load_tokens() is not None
+    return has_cloud_credentials(config)
 
 
 def _require_cloud_credentials(config) -> None:
@@ -57,8 +54,7 @@ def _require_cloud_credentials(config) -> None:
 async def _get_cloud_project(name: str) -> ProjectItem | None:
     """Fetch a project by name from the cloud API."""
     async with get_client() as client:
-        response = await call_get(client, "/v2/projects/")
-        projects_list = ProjectList.model_validate(response.json())
+        projects_list = await ProjectClient(client).list_projects()
         for proj in projects_list.projects:
             if generate_permalink(proj.name) == generate_permalink(name):
                 return proj
@@ -132,12 +128,9 @@ def sync_project_command(
 
                 async def _trigger_db_sync():
                     async with get_client() as client:
-                        response = await call_post(
-                            client,
-                            f"/v2/projects/{project_data.external_id}/sync?force_full=true",
-                            json={},
+                        return await ProjectClient(client).sync(
+                            project_data.external_id, force_full=True
                         )
-                        return response.json()
 
                 try:
                     with force_routing(cloud=True):
@@ -210,12 +203,9 @@ def bisync_project_command(
 
                 async def _trigger_db_sync():
                     async with get_client() as client:
-                        response = await call_post(
-                            client,
-                            f"/v2/projects/{project_data.external_id}/sync?force_full=true",
-                            json={},
+                        return await ProjectClient(client).sync(
+                            project_data.external_id, force_full=True
                         )
-                        return response.json()
 
                 try:
                     with force_routing(cloud=True):
@@ -329,9 +319,8 @@ def setup_project_sync(
     async def _verify_project_exists():
         """Verify the project exists on cloud by listing all projects."""
         async with get_client() as client:
-            response = await call_get(client, "/v2/projects/")
-            project_list = response.json()
-            project_names = [p["name"] for p in project_list["projects"]]
+            projects_list = await ProjectClient(client).list_projects()
+            project_names = [p.name for p in projects_list.projects]
             if name not in project_names:
                 raise ValueError(f"Project '{name}' not found on cloud")
             return True

src/basic_memory/cli/commands/cloud/upload.py

@@ -10,7 +10,6 @@ import httpx
 
 from basic_memory.ignore_utils import load_gitignore_patterns, should_ignore_path
 from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.tools.utils import call_put
 
 # Archive file extensions that should be skipped during upload
 ARCHIVE_EXTENSIONS = {".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar", ".tgz", ".tbz2"}
@@ -24,7 +23,7 @@ async def upload_path(
     dry_run: bool = False,
     *,
     client_cm_factory: Callable[[], AbstractAsyncContextManager[httpx.AsyncClient]] | None = None,
-    put_func=call_put,
+    put_func: Callable | None = None,
 ) -> bool:
     """
     Upload a file or directory to cloud project via WebDAV.
@@ -117,9 +116,20 @@ async def upload_path(
 
                     # Upload via HTTP PUT to WebDAV endpoint with mtime header
                     # Using X-OC-Mtime (ownCloud/Nextcloud standard)
-                    response = await put_func(
-                        client, remote_path, content=content, headers={"X-OC-Mtime": str(mtime)}
-                    )
+                    if put_func is not None:
+                        # Test injection path
+                        response = await put_func(
+                            client,
+                            remote_path,
+                            content=content,
+                            headers={"X-OC-Mtime": str(mtime)},
+                        )
+                    else:
+                        response = await client.put(
+                            remote_path,
+                            content=content,
+                            headers={"X-OC-Mtime": str(mtime)},
+                        )
                     response.raise_for_status()
 
         # Format total size based on magnitude

src/basic_memory/cli/commands/cloud/workspace.py

@@ -4,9 +4,13 @@ import typer
 from rich.console import Console
 from rich.table import Table
 
-from basic_memory.cli.app import cloud_app
 from basic_memory.cli.commands.command_utils import run_with_cleanup
-from basic_memory.mcp.project_context import get_available_workspaces
+from basic_memory.config import ConfigManager
+from basic_memory.mcp.project_context import (
+    _workspace_choices,
+    _workspace_matches_identifier,
+    get_available_workspaces,
+)
 
 console = Console()
 
@@ -33,18 +37,77 @@ def list_workspaces() -> None:
         console.print("[yellow]No accessible workspaces found.[/yellow]")
         return
 
+    config = ConfigManager().config
+    default_ws = config.default_workspace
+
     table = Table(title="Available Workspaces")
     table.add_column("Name", style="cyan")
     table.add_column("Type", style="blue")
     table.add_column("Role", style="green")
     table.add_column("Tenant ID", style="yellow")
+    table.add_column("Default", style="magenta")
 
     for workspace in workspaces:
+        is_default = "[X]" if workspace.tenant_id == default_ws else ""
         table.add_row(
             workspace.name,
             workspace.workspace_type,
             workspace.role,
             workspace.tenant_id,
+            is_default,
         )
 
     console.print(table)
+
+
+@workspace_app.command("set-default")
+def set_default_workspace(
+    identifier: str = typer.Argument(..., help="Workspace name or tenant_id to set as default"),
+) -> None:
+    """Set the default cloud workspace.
+
+    The default workspace is used as fallback when no per-project workspace
+    is configured. Resolves the identifier against available workspaces.
+
+    Examples:
+      bm cloud workspace set-default Personal
+      bm cloud workspace set-default 11111111-1111-1111-1111-111111111111
+    """
+
+    async def _list():
+        return await get_available_workspaces()
+
+    try:
+        workspaces = run_with_cleanup(_list())
+    except RuntimeError as exc:
+        console.print(f"[red]Error: {exc}[/red]")
+        raise typer.Exit(1)
+
+    if not workspaces:
+        console.print("[yellow]No accessible workspaces found.[/yellow]")
+        raise typer.Exit(1)
+
+    matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, identifier)]
+
+    if not matches:
+        console.print(f"[red]Error: Workspace '{identifier}' not found[/red]")
+        console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+        raise typer.Exit(1)
+
+    if len(matches) > 1:
+        console.print(
+            f"[red]Error: Workspace name '{identifier}' matches multiple workspaces. "
+            f"Use tenant_id instead.[/red]"
+        )
+        console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+        raise typer.Exit(1)
+
+    selected = matches[0]
+    config_manager = ConfigManager()
+    config = config_manager.config
+    config.default_workspace = selected.tenant_id
+    config_manager.save_config(config)
+
+    console.print(
+        f"[green]Default workspace set to '{selected.name}' ({selected.tenant_id})[/green]"
+    )

src/basic_memory/cli/commands/command_utils.py

@@ -11,9 +11,8 @@ from rich.console import Console
 from basic_memory import db
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.tools.utils import call_post, call_get
+from basic_memory.mcp.clients import ProjectClient
 from basic_memory.mcp.project_context import get_active_project
-from basic_memory.schemas import ProjectInfoResponse
 
 console = Console()
 
@@ -61,16 +60,12 @@ async def run_sync(
     try:
         async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
-            url = f"/v2/projects/{project_item.external_id}/sync"
-            params = []
-            if force_full:
-                params.append("force_full=true")
-            if not run_in_background:
-                params.append("run_in_background=false")
-            if params:
-                url += "?" + "&".join(params)
-            response = await call_post(client, url)
-            data = response.json()
+            project_client = ProjectClient(client)
+            data = await project_client.sync(
+                project_item.external_id,
+                force_full=force_full,
+                run_in_background=run_in_background,
+            )
             # Background mode returns {"message": "..."}, foreground returns SyncReportResponse
             if "message" in data:
                 console.print(f"[green]{data['message']}[/green]")
@@ -94,8 +89,7 @@ async def get_project_info(project: str):
     try:
         async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
-            response = await call_get(client, f"/v2/projects/{project_item.external_id}/info")
-            return ProjectInfoResponse.model_validate(response.json())
+            return await ProjectClient(client).get_info(project_item.external_id)
     except (ToolError, ValueError) as e:
         error_text = str(e)
         if "internal proxy error" in error_text.lower() and "not found in configuration" in (

src/basic_memory/cli/commands/doctor.py

@@ -19,7 +19,6 @@ from basic_memory.markdown.markdown_processor import MarkdownProcessor
 from basic_memory.markdown.schemas import EntityFrontmatter, EntityMarkdown
 from basic_memory.mcp.async_client import get_client
 from basic_memory.mcp.clients import KnowledgeClient, ProjectClient, SearchClient
-from basic_memory.mcp.tools.utils import call_post
 from basic_memory.schemas.base import Entity
 from basic_memory.schemas.project_info import ProjectInfoRequest
 from basic_memory.schemas.search import SearchQuery
@@ -98,11 +97,10 @@ async def run_doctor() -> None:
                 await processor.write_file(manual_path, manual_markdown)
                 console.print("[green]OK[/green] Manual file written")
 
-                sync_response = await call_post(
-                    client,
-                    f"/v2/projects/{project_id}/sync?force_full=true&run_in_background=false",
+                sync_data = await project_client.sync(
+                    project_id, force_full=True, run_in_background=False
                 )
-                sync_report = SyncReportResponse.model_validate(sync_response.json())
+                sync_report = SyncReportResponse.model_validate(sync_data)
                 if sync_report.total == 0:
                     raise ValueError("Sync did not detect any changes")
 
@@ -118,8 +116,7 @@ async def run_doctor() -> None:
 
                 console.print("[green]OK[/green] Search confirmed manual file")
 
-                status_response = await call_post(client, f"/v2/projects/{project_id}/status")
-                status_report = SyncReportResponse.model_validate(status_response.json())
+                status_report = await project_client.get_status(project_id)
                 if status_report.total != 0:
                     raise ValueError("Project status not clean after sync")
 

src/basic_memory/cli/commands/project.py

@@ -25,9 +25,8 @@ from basic_memory.cli.commands.command_utils import get_project_info, run_with_c
 from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
 from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
 from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.tools.utils import call_delete, call_get, call_patch, call_post, call_put
-from basic_memory.schemas.project_info import ProjectItem, ProjectList, ProjectStatusResponse
-from basic_memory.schemas.v2 import ProjectResolveResponse
+from basic_memory.mcp.clients import ProjectClient
+from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.utils import generate_permalink, normalize_project_path
 
 console = Console()
@@ -49,6 +48,7 @@ def format_path(path: str) -> str:
 def list_projects(
     local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
     cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
+    workspace: str = typer.Option(None, "--workspace", help="Cloud workspace name or tenant_id"),
 ) -> None:
     """List Basic Memory projects from local and (when available) cloud."""
     try:
@@ -57,20 +57,22 @@ def list_projects(
         console.print(f"[red]Error: {e}[/red]")
         raise typer.Exit(1)
 
-    async def _list_projects():
-        async with get_client() as client:
-            response = await call_get(client, "/v2/projects/")
-            return ProjectList.model_validate(response.json())
+    async def _list_projects(ws: str | None = None):
+        async with get_client(workspace=ws) as client:
+            return await ProjectClient(client).list_projects()
 
     try:
         config = ConfigManager().config
+        # Use explicit workspace, fall back to config default
+        effective_workspace = workspace or config.default_workspace
+
         local_result: ProjectList | None = None
         cloud_result: ProjectList | None = None
         cloud_error: Exception | None = None
 
         if cloud:
             with force_routing(cloud=True):
-                cloud_result = run_with_cleanup(_list_projects())
+                cloud_result = run_with_cleanup(_list_projects(effective_workspace))
         elif local:
             with force_routing(local=True):
                 local_result = run_with_cleanup(_list_projects())
@@ -82,14 +84,33 @@ def list_projects(
             if _has_cloud_credentials(config):
                 try:
                     with force_routing(cloud=True):
-                        cloud_result = run_with_cleanup(_list_projects())
+                        cloud_result = run_with_cleanup(_list_projects(effective_workspace))
                 except Exception as exc:  # pragma: no cover
                     cloud_error = exc
 
+        # Resolve workspace name for cloud projects (best-effort)
+        cloud_ws_name: str | None = None
+        cloud_ws_type: str | None = None
+        if cloud_result and effective_workspace:
+            try:
+                from basic_memory.mcp.project_context import get_available_workspaces
+
+                workspaces = run_with_cleanup(get_available_workspaces())
+                matched = next(
+                    (ws for ws in workspaces if ws.tenant_id == effective_workspace),
+                    None,
+                )
+                if matched:
+                    cloud_ws_name = matched.name
+                    cloud_ws_type = matched.workspace_type
+            except Exception:
+                pass
+
         table = Table(title="Basic Memory Projects")
         table.add_column("Name", style="cyan")
         table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
         table.add_column("Cloud Path", style="green")
+        table.add_column("Workspace", style="green")
         table.add_column("CLI Route", style="blue")
         table.add_column("MCP (stdio)", style="blue")
         table.add_column("Sync", style="green")
@@ -151,10 +172,16 @@ def list_projects(
             has_sync = "[X]" if entry and entry.local_sync_path else ""
             mcp_stdio_target = "local" if local_project is not None else "n/a"
 
+            # Show workspace name (type) for cloud-sourced projects
+            ws_label = ""
+            if cloud_project is not None and cloud_ws_name:
+                ws_label = f"{cloud_ws_name} ({cloud_ws_type})" if cloud_ws_type else cloud_ws_name
+
             row = [
                 project_name,
                 local_path,
                 cloud_path,
+                ws_label,
                 cli_route,
                 mcp_stdio_target,
                 has_sync,
@@ -229,8 +256,7 @@ def add_project(
                     "local_sync_path": local_sync_path,
                     "set_default": set_default,
                 }
-                response = await call_post(client, "/v2/projects/", json=data)
-                return ProjectStatusResponse.model_validate(response.json())
+                return await ProjectClient(client).create_project(data)
     else:
         # Local mode: path is required
         if path is None:
@@ -243,8 +269,7 @@ def add_project(
         async def _add_project():
             async with get_client() as client:
                 data = {"name": name, "path": resolved_path, "set_default": set_default}
-                response = await call_post(client, "/v2/projects/", json=data)
-                return ProjectStatusResponse.model_validate(response.json())
+                return await ProjectClient(client).create_project(data)
 
     try:
         with force_routing(local=local, cloud=cloud):
@@ -302,19 +327,13 @@ def remove_project(
 
     async def _remove_project():
         async with get_client() as client:
+            project_client = ProjectClient(client)
             # Convert name to permalink for efficient resolution
             project_permalink = generate_permalink(name)
-
-            # Use v2 project resolver to find project ID by permalink
-            resolve_data = {"identifier": project_permalink}
-            response = await call_post(client, "/v2/projects/resolve", json=resolve_data)
-            target_project = response.json()
-
-            # Use v2 API with project ID
-            response = await call_delete(
-                client, f"/v2/projects/{target_project['external_id']}?delete_notes={delete_notes}"
+            target_project = await project_client.resolve_project(project_permalink)
+            return await project_client.delete_project(
+                target_project.external_id, delete_notes=delete_notes
             )
-            return ProjectStatusResponse.model_validate(response.json())
 
     try:
         # Get config to check for local sync path and bisync state
@@ -387,19 +406,11 @@ def set_default_project(
 
     async def _set_default():
         async with get_client() as client:
+            project_client = ProjectClient(client)
             # Convert name to permalink for efficient resolution
             project_permalink = generate_permalink(name)
-
-            # Use v2 project resolver to find project ID by permalink
-            resolve_data = {"identifier": project_permalink}
-            response = await call_post(client, "/v2/projects/resolve", json=resolve_data)
-            target_project = response.json()
-
-            # Use v2 API with project ID
-            response = await call_put(
-                client, f"/v2/projects/{target_project['external_id']}/default"
-            )
-            return ProjectStatusResponse.model_validate(response.json())
+            target_project = await project_client.resolve_project(project_permalink)
+            return await project_client.set_default(target_project.external_id)
 
     try:
         with force_routing(local=local):
@@ -410,31 +421,6 @@ def set_default_project(
         raise typer.Exit(1)
 
 
-@project_app.command("sync-config")
-def synchronize_projects(
-    local: bool = typer.Option(
-        False, "--local", help="Force local API routing (required in cloud mode)"
-    ),
-) -> None:
-    """Synchronize project config between configuration file and database.
-
-    In cloud mode, use --local to sync local configuration.
-    """
-
-    async def _sync_config():
-        async with get_client() as client:
-            response = await call_post(client, "/v2/projects/config/sync")
-            return ProjectStatusResponse.model_validate(response.json())
-
-    try:
-        with force_routing(local=local):
-            result = run_with_cleanup(_sync_config())
-        console.print(f"[green]{result.message}[/green]")
-    except Exception as e:  # pragma: no cover
-        console.print(f"[red]Error synchronizing projects: {str(e)}[/red]")
-        raise typer.Exit(1)
-
-
 @project_app.command("move")
 def move_project(
     name: str = typer.Argument(..., help="Name of the project to move"),
@@ -450,17 +436,11 @@ def move_project(
 
     async def _move_project():
         async with get_client() as client:
-            data = {"path": resolved_path}
-            resolve_response = await call_post(
-                client,
-                "/v2/projects/resolve",
-                json={"identifier": name},
+            project_client = ProjectClient(client)
+            project_info = await project_client.resolve_project(name)
+            return await project_client.update_project(
+                project_info.external_id, {"path": resolved_path}
             )
-            project_info = ProjectResolveResponse.model_validate(resolve_response.json())
-            response = await call_patch(
-                client, f"/v2/projects/{project_info.external_id}", json=data
-            )
-            return ProjectStatusResponse.model_validate(response.json())
 
     try:
         with force_routing(local=True):
@@ -489,17 +469,24 @@ def move_project(
 @project_app.command("set-cloud")
 def set_cloud(
     name: str = typer.Argument(..., help="Name of the project to route through cloud"),
+    workspace: str = typer.Option(
+        None,
+        "--workspace",
+        help="Cloud workspace name or tenant_id to associate with this project",
+    ),
 ) -> None:
     """Set a project to cloud mode (route through cloud API).
 
     Requires either an API key or an active OAuth session.
 
-    Examples:
-      bm cloud api-key save bmc_abc123...   # save API key, then:
-      bm project set-cloud research    # route "research" through cloud
+    Use --workspace to associate a specific workspace with this project.
+    If omitted, uses the default workspace (if set) or auto-selects when
+    only one workspace is available.
 
-      bm cloud login                   # OAuth login, then:
-      bm project set-cloud research    # route "research" through cloud
+    Examples:
+      bm project set-cloud research --workspace Personal
+      bm project set-cloud research --workspace 11111111-...
+      bm project set-cloud research   # uses default workspace
     """
 
     config_manager = ConfigManager()
@@ -522,10 +509,54 @@ def set_cloud(
         console.print("[dim]Run 'bm cloud api-key save <key>' or 'bm cloud login' first[/dim]")
         raise typer.Exit(1)
 
+    # --- Resolve workspace to tenant_id ---
+    resolved_workspace_id: str | None = None
+
+    if workspace is not None:
+        # Explicit --workspace: resolve to tenant_id via cloud lookup
+        from basic_memory.mcp.project_context import (
+            get_available_workspaces,
+            _workspace_matches_identifier,
+            _workspace_choices,
+        )
+
+        workspaces = run_with_cleanup(get_available_workspaces())
+        matches = [ws for ws in workspaces if _workspace_matches_identifier(ws, workspace)]
+        if not matches:
+            console.print(f"[red]Error: Workspace '{workspace}' not found[/red]")
+            if workspaces:
+                console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+            raise typer.Exit(1)
+        if len(matches) > 1:
+            console.print(
+                f"[red]Error: Workspace name '{workspace}' matches multiple workspaces. "
+                f"Use tenant_id instead.[/red]"
+            )
+            console.print(f"[dim]Available:\n{_workspace_choices(workspaces)}[/dim]")
+            raise typer.Exit(1)
+        resolved_workspace_id = matches[0].tenant_id
+    elif config.default_workspace:
+        # Fall back to global default
+        resolved_workspace_id = config.default_workspace
+    else:
+        # Try auto-select if single workspace
+        try:
+            from basic_memory.mcp.project_context import get_available_workspaces
+
+            workspaces = run_with_cleanup(get_available_workspaces())
+            if len(workspaces) == 1:
+                resolved_workspace_id = workspaces[0].tenant_id
+        except Exception:
+            pass  # Workspace resolution is optional at set-cloud time
+
     config.set_project_mode(name, ProjectMode.CLOUD)
+    if resolved_workspace_id:
+        config.projects[name].workspace_id = resolved_workspace_id
     config_manager.save_config(config)
 
     console.print(f"[green]Project '{name}' set to cloud mode[/green]")
+    if resolved_workspace_id:
+        console.print(f"[dim]Workspace: {resolved_workspace_id}[/dim]")
     console.print("[dim]MCP tools and CLI commands for this project will route through cloud[/dim]")
 
 
@@ -535,6 +566,8 @@ def set_local(
 ) -> None:
     """Revert a project to local mode (use in-process ASGI transport).
 
+    Clears any associated cloud workspace.
+
     Example:
       bm project set-local research
     """
@@ -547,6 +580,7 @@ def set_local(
         raise typer.Exit(1)
 
     config.set_project_mode(name, ProjectMode.LOCAL)
+    config.projects[name].workspace_id = None
     config_manager.save_config(config)
 
     console.print(f"[green]Project '{name}' set to local mode[/green]")
@@ -612,8 +646,7 @@ def ls_project_command(
         # Get project info
         async def _get_project():
             async with get_client() as client:
-                response = await call_get(client, "/v2/projects/")
-                projects_list = ProjectList.model_validate(response.json())
+                projects_list = await ProjectClient(client).list_projects()
                 for proj in projects_list.projects:
                     if generate_permalink(proj.name) == generate_permalink(name):
                         return proj

src/basic_memory/cli/commands/status.py

@@ -14,7 +14,7 @@ from basic_memory.cli.app import app
 from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.async_client import get_client
-from basic_memory.mcp.tools.utils import call_post
+from basic_memory.mcp.clients import ProjectClient
 from basic_memory.schemas import SyncReportResponse
 from basic_memory.mcp.project_context import get_active_project
 
@@ -149,8 +149,7 @@ async def run_status(project: Optional[str] = None, verbose: bool = False):  # p
     try:
         async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
-            response = await call_post(client, f"/v2/projects/{project_item.external_id}/status")
-            sync_report = SyncReportResponse.model_validate(response.json())
+            sync_report = await ProjectClient(client).get_status(project_item.external_id)
 
             display_changes(project_item.name, "Status", sync_report, verbose)
 

src/basic_memory/cli/commands/tool.py

@@ -17,6 +17,8 @@ from basic_memory.cli.commands.routing import force_routing, validate_routing_fl
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.tools import build_context as mcp_build_context
 from basic_memory.mcp.tools import edit_note as mcp_edit_note
+from basic_memory.mcp.tools import list_memory_projects as mcp_list_projects
+from basic_memory.mcp.tools import list_workspaces as mcp_list_workspaces
 from basic_memory.mcp.tools import read_note as mcp_read_note
 from basic_memory.mcp.tools import recent_activity as mcp_recent_activity
 from basic_memory.mcp.tools import schema_diff as mcp_schema_diff
@@ -545,6 +547,72 @@ def search_notes(
         raise
 
 
+# --- list-projects ---
+
+
+@tool_app.command("list-projects")
+def list_projects(
+    local: bool = typer.Option(
+        False, "--local", help="Force local API routing (ignore cloud mode)"
+    ),
+    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
+):
+    """List all available projects with their status (JSON output).
+
+    Examples:
+
+    bm tool list-projects
+    bm tool list-projects --local
+    """
+    try:
+        validate_routing_flags(local, cloud)
+
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(mcp_list_projects(output_format="json"))
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during list_projects: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+# --- list-workspaces ---
+
+
+@tool_app.command("list-workspaces")
+def list_workspaces(
+    local: bool = typer.Option(
+        False, "--local", help="Force local API routing (ignore cloud mode)"
+    ),
+    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
+):
+    """List available cloud workspaces (JSON output).
+
+    Examples:
+
+    bm tool list-workspaces
+    bm tool list-workspaces --cloud
+    """
+    try:
+        validate_routing_flags(local, cloud)
+
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(mcp_list_workspaces(output_format="json"))
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during list_workspaces: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
 # --- schema-validate ---