feat: add workspace selection and routing for MCP tools

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

Author: phernandez

src/basic_memory/cli/commands/__init__.py

@@ -1,7 +1,16 @@
 """CLI commands for basic-memory."""
 
 from . import status, db, doctor, import_memory_json, mcp, import_claude_conversations
-from . import import_claude_projects, import_chatgpt, tool, project, format, schema, watch
+from . import (
+    import_claude_projects,
+    import_chatgpt,
+    tool,
+    project,
+    format,
+    schema,
+    watch,
+    workspace,
+)
 
 __all__ = [
     "status",
@@ -17,4 +26,5 @@
     "format",
     "schema",
     "watch",
+    "workspace",
 ]

src/basic_memory/cli/commands/tool.py

@@ -99,14 +99,26 @@ def _parse_opening_frontmatter(content: str) -> tuple[str, dict[str, Any] | None
 
 
 async def _write_note_json(
-    title: str, content: str, folder: str, project_name: Optional[str], tags: Optional[List[str]]
+    title: str,
+    content: str,
+    folder: str,
+    project_name: Optional[str],
+    workspace: Optional[str],
+    tags: Optional[List[str]],
 ) -> dict:
     """Write a note and return structured JSON metadata."""
     # Use the MCP tool to create/update the entity (handles create-or-update logic)
-    await mcp_write_note.fn(title, content, folder, project_name, tags)
+    await mcp_write_note.fn(
+        title=title,
+        content=content,
+        directory=folder,
+        project=project_name,
+        workspace=workspace,
+        tags=tags,
+    )
 
     # Resolve the entity to get metadata back
-    async with get_client(project_name=project_name) as client:
+    async with get_client(project_name=project_name, workspace=workspace) as client:
         active_project = await get_active_project(client, project_name)
         knowledge_client = KnowledgeClient(client, active_project.external_id)
 
@@ -125,10 +137,14 @@ async def _write_note_json(
 
 
 async def _read_note_json(
-    identifier: str, project_name: Optional[str], page: int, page_size: int
+    identifier: str,
+    project_name: Optional[str],
+    workspace: Optional[str],
+    page: int,
+    page_size: int,
 ) -> dict:
     """Read a note and return structured JSON with content and metadata."""
-    async with get_client(project_name=project_name) as client:
+    async with get_client(project_name=project_name, workspace=workspace) as client:
         active_project = await get_active_project(client, project_name)
         knowledge_client = KnowledgeClient(client, active_project.external_id)
         resource_client = ResourceClient(client, active_project.external_id)
@@ -146,7 +162,10 @@ async def _read_note_json(
             from basic_memory.mcp.tools.search import search_notes as mcp_search_tool
 
             title_results = await mcp_search_tool.fn(
-                query=identifier, search_type="title", project=project_name
+                query=identifier,
+                search_type="title",
+                project=project_name,
+                workspace=workspace,
             )
             if title_results and hasattr(title_results, "results") and title_results.results:
                 result = title_results.results[0]
@@ -172,12 +191,13 @@ async def _edit_note_json(
     operation: str,
     content: str,
     project_name: Optional[str],
+    workspace: Optional[str],
     section: Optional[str],
     find_text: Optional[str],
     expected_replacements: int,
 ) -> dict:
     """Edit a note and return structured JSON metadata."""
-    async with get_client(project_name=project_name) as client:
+    async with get_client(project_name=project_name, workspace=workspace) as client:
         active_project = await get_active_project(client, project_name)
         knowledge_client = KnowledgeClient(client, active_project.external_id)
 
@@ -227,11 +247,12 @@ async def _recent_activity_json(
     depth: Optional[int],
     timeframe: Optional[TimeFrame],
     project_name: Optional[str] = None,
+    workspace: Optional[str] = None,
     page: int = 1,
     page_size: int = 50,
 ) -> list:
     """Get recent activity and return structured JSON list."""
-    async with get_client(project_name=project_name) as client:
+    async with get_client(project_name=project_name, workspace=workspace) as client:
         # Build query params matching the MCP tool's logic
         params: dict = {"page": page, "page_size": page_size, "max_related": 10}
         if depth:
@@ -275,6 +296,10 @@ def write_note(
             help="The project to write to. If not provided, the default project will be used."
         ),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     content: Annotated[
         Optional[str],
         typer.Option(
@@ -362,12 +387,19 @@ def write_note(
         with force_routing(local=local, cloud=cloud):
             if format == "json":
                 result = run_with_cleanup(
-                    _write_note_json(title, content, folder, project_name, tags)
+                    _write_note_json(title, content, folder, project_name, workspace, tags)
                 )
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
                 note = run_with_cleanup(
-                    mcp_write_note.fn(title, content, folder, project_name, tags)
+                    mcp_write_note.fn(
+                        title=title,
+                        content=content,
+                        directory=folder,
+                        project=project_name,
+                        workspace=workspace,
+                        tags=tags,
+                    )
                 )
                 rprint(note)
     except ValueError as e:
@@ -389,6 +421,10 @@ def read_note(
             help="The project to use for the note. If not provided, the default project will be used."
         ),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     page: int = 1,
     page_size: int = 10,
     format: str = typer.Option("text", "--format", help="Output format: text or json"),
@@ -429,15 +465,23 @@ def read_note(
         with force_routing(local=local, cloud=cloud):
             if format == "json":
                 result = run_with_cleanup(
-                    _read_note_json(identifier, project_name, page, page_size)
+                    _read_note_json(identifier, project_name, workspace, page, page_size)
                 )
                 stripped_content, parsed_frontmatter = _parse_opening_frontmatter(result["content"])
                 result["frontmatter"] = parsed_frontmatter
                 if strip_frontmatter:
                     result["content"] = stripped_content
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
-                note = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size))
+                note = run_with_cleanup(
+                    mcp_read_note.fn(
+                        identifier=identifier,
+                        project=project_name,
+                        workspace=workspace,
+                        page=page,
+                        page_size=page_size,
+                    )
+                )
                 if strip_frontmatter:
                     note, _ = _parse_opening_frontmatter(note)
                 rprint(note)
@@ -462,6 +506,10 @@ def edit_note(
             help="The project to edit. If not provided, the default project will be used."
         ),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     find_text: Annotated[
         Optional[str], typer.Option("--find-text", help="Text to find for find_replace operation")
     ] = None,
@@ -509,6 +557,7 @@ def edit_note(
                         operation=operation,
                         content=content,
                         project_name=project_name,
+                        workspace=workspace,
                         section=section,
                         find_text=find_text,
                         expected_replacements=expected_replacements,
@@ -522,6 +571,7 @@ def edit_note(
                         operation=operation,
                         content=content,
                         project=project_name,
+                        workspace=workspace,
                         section=section,
                         find_text=find_text,
                         expected_replacements=expected_replacements,
@@ -547,6 +597,10 @@ def build_context(
         Optional[str],
         typer.Option(help="The project to use. If not provided, the default project will be used."),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     depth: Optional[int] = 1,
     timeframe: Optional[TimeFrame] = "7d",
     page: int = 1,
@@ -582,6 +636,7 @@ def build_context(
             result = run_with_cleanup(
                 mcp_build_context.fn(
                     project=project_name,
+                    workspace=workspace,
                     url=url,
                     depth=depth,
                     timeframe=timeframe,
@@ -609,6 +664,10 @@ def recent_activity(
         Optional[str],
         typer.Option(help="The project to use. If not provided, the default project will be used."),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     depth: Optional[int] = 1,
     timeframe: Optional[TimeFrame] = "7d",
     page: int = typer.Option(1, "--page", help="Page number for pagination (JSON format)"),
@@ -642,7 +701,15 @@ def recent_activity(
         with force_routing(local=local, cloud=cloud):
             if format == "json":
                 result = run_with_cleanup(
-                    _recent_activity_json(type, depth, timeframe, project_name, page, page_size)
+                    _recent_activity_json(
+                        type=type,
+                        depth=depth,
+                        timeframe=timeframe,
+                        project_name=project_name,
+                        workspace=workspace,
+                        page=page,
+                        page_size=page_size,
+                    )
                 )
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
@@ -652,6 +719,7 @@ def recent_activity(
                         depth=depth,
                         timeframe=timeframe,
                         project=project_name,
+                        workspace=workspace,
                     )
                 )
                 # The tool returns a formatted string directly
@@ -682,6 +750,10 @@ def search_notes(
             help="The project to use for the note. If not provided, the default project will be used."
         ),
     ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
     after_date: Annotated[
         Optional[str],
         typer.Option("--after_date", help="Search results after date, eg. '2d', '1 week'"),
@@ -793,8 +865,9 @@ def search_notes(
         with force_routing(local=local, cloud=cloud):
             results = run_with_cleanup(
                 mcp_search.fn(
-                    query or "",
-                    project_name,
+                    query=query or "",
+                    project=project_name,
+                    workspace=workspace,
                     search_type=search_type,
                     page=page,
                     after_date=after_date,

src/basic_memory/cli/commands/workspace.py

@@ -0,0 +1,57 @@
+"""Workspace commands for Basic Memory cloud workspaces."""
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from basic_memory.cli.app import app
+from basic_memory.cli.commands.command_utils import run_with_cleanup
+from basic_memory.mcp.project_context import get_available_workspaces
+
+console = Console()
+
+workspace_app = typer.Typer(help="Manage cloud workspaces")
+app.add_typer(workspace_app, name="workspace")
+
+
+@workspace_app.command("list")
+def list_workspaces() -> None:
+    """List cloud workspaces available to the current OAuth session."""
+
+    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)
+    except Exception as exc:  # pragma: no cover
+        console.print(f"[red]Error listing workspaces: {exc}[/red]")
+        raise typer.Exit(1)
+
+    if not workspaces:
+        console.print("[yellow]No accessible workspaces found.[/yellow]")
+        return
+
+    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")
+
+    for workspace in workspaces:
+        table.add_row(
+            workspace.name,
+            workspace.workspace_type,
+            workspace.role,
+            workspace.tenant_id,
+        )
+
+    console.print(table)
+
+
+@app.command("workspaces")
+def workspaces_alias() -> None:
+    """Alias for `bm workspace list`."""
+    list_workspaces()

src/basic_memory/cli/main.py

@@ -28,6 +28,7 @@ def _version_only_invocation(argv: list[str]) -> bool:
         schema,
         status,
         tool,
+        workspace,
     )
 
 warnings.filterwarnings("ignore")  # pragma: no cover