feat: add --format json to CLI tool commands (#552)

Signed-off-by: phernandez <paul@basicmachines.co>
Signed-off-by: bm-clawd <clawd@basicmemory.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: bm-clawd <clawd@basicmemory.com>

Author: 3 people

src/basic_memory/cli/commands/tool.py

@@ -12,6 +12,13 @@
 from basic_memory.cli.commands.command_utils import run_with_cleanup
 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.clients import KnowledgeClient, ResourceClient
+from basic_memory.mcp.project_context import get_active_project
+from basic_memory.mcp.tools.utils import call_get
+from basic_memory.schemas.base import Entity, TimeFrame
+from basic_memory.schemas.memory import GraphContext, MemoryUrl, memory_url_path
+from basic_memory.schemas.search import SearchItemType
 
 # Import prompts
 from basic_memory.mcp.prompts.continue_conversation import (
@@ -25,14 +32,128 @@
 from basic_memory.mcp.tools import recent_activity as mcp_recent_activity
 from basic_memory.mcp.tools import search_notes as mcp_search
 from basic_memory.mcp.tools import write_note as mcp_write_note
-from basic_memory.schemas.base import TimeFrame
-from basic_memory.schemas.memory import MemoryUrl
-from basic_memory.schemas.search import SearchItemType
 
 tool_app = typer.Typer()
 app.add_typer(tool_app, name="tool", help="Access to MCP tools via CLI")
 
 
+# --- JSON output helpers ---
+# These async functions bypass the MCP tool (which returns formatted strings)
+# and use API clients directly to return structured data for --format json.
+
+
+async def _write_note_json(
+    title: str, content: str, folder: str, project_name: 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)
+
+    # Resolve the entity to get metadata back
+    async with get_client() as client:
+        active_project = await get_active_project(client, project_name)
+        knowledge_client = KnowledgeClient(client, active_project.external_id)
+
+        entity = Entity(title=title, directory=folder)
+        if not entity.permalink:
+            raise ValueError(f"Could not generate permalink for title={title}, folder={folder}")
+        entity_id = await knowledge_client.resolve_entity(entity.permalink)
+        entity = await knowledge_client.get_entity(entity_id)
+
+        return {
+            "title": entity.title,
+            "permalink": entity.permalink,
+            "content": content,
+            "file_path": entity.file_path,
+        }
+
+
+async def _read_note_json(
+    identifier: str, project_name: Optional[str], page: int, page_size: int
+) -> dict:
+    """Read a note and return structured JSON with content and metadata."""
+    async with get_client() 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)
+
+        # Try direct resolution first (works for permalinks and memory URLs)
+        entity_path = memory_url_path(identifier)
+        entity_id = None
+        try:
+            entity_id = await knowledge_client.resolve_entity(entity_path)
+        except Exception:
+            logger.info(f"Direct lookup failed for '{entity_path}', trying title search")
+
+        # Fallback: title search (handles plain titles like "My Note")
+        if entity_id is None:
+            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
+            )
+            if title_results and hasattr(title_results, "results") and title_results.results:
+                result = title_results.results[0]
+                if result.permalink:
+                    entity_id = await knowledge_client.resolve_entity(result.permalink)
+
+        if entity_id is None:
+            raise ValueError(f"Could not find note matching: {identifier}")
+
+        entity = await knowledge_client.get_entity(entity_id)
+        response = await resource_client.read(entity_id, page=page, page_size=page_size)
+
+        return {
+            "title": entity.title,
+            "permalink": entity.permalink,
+            "content": response.text,
+            "file_path": entity.file_path,
+        }
+
+
+async def _recent_activity_json(
+    type: Optional[List[SearchItemType]],
+    depth: Optional[int],
+    timeframe: Optional[TimeFrame],
+    project_name: Optional[str] = None,
+    page: int = 1,
+    page_size: int = 50,
+) -> list:
+    """Get recent activity and return structured JSON list."""
+    async with get_client() as client:
+        # Build query params matching the MCP tool's logic
+        params: dict = {"page": page, "page_size": page_size, "max_related": 10}
+        if depth:
+            params["depth"] = depth
+        if timeframe:
+            params["timeframe"] = timeframe
+        if type:
+            params["type"] = [t.value for t in type]
+
+        active_project = await get_active_project(client, project_name)
+        response = await call_get(
+            client,
+            f"/v2/projects/{active_project.external_id}/memory/recent",
+            params=params,
+        )
+        activity_data = GraphContext.model_validate(response.json())
+
+        # Extract entity results
+        results = []
+        for result in activity_data.results:
+            pr = result.primary_result
+            if pr.type == "entity":
+                results.append(
+                    {
+                        "title": pr.title,
+                        "permalink": pr.permalink,
+                        "file_path": pr.file_path,
+                        "created_at": str(pr.created_at) if pr.created_at else None,
+                    }
+                )
+        return results
+
+
 @tool_app.command()
 def write_note(
     title: Annotated[str, typer.Option(help="The title of the note")],
@@ -52,6 +173,7 @@ def write_note(
     tags: Annotated[
         Optional[List[str]], typer.Option(help="A list of tags to apply to the note")
     ] = None,
+    format: str = typer.Option("text", "--format", help="Output format: text or json"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
     ),
@@ -123,9 +245,20 @@ def write_note(
         # use the project name, or the default from the config
         project_name = project_name or config_manager.default_project
 
+        # content is validated non-None above (stdin or --content)
+        assert content is not None
+
         with force_routing(local=local, cloud=cloud):
-            note = run_with_cleanup(mcp_write_note.fn(title, content, folder, project_name, tags))
-        rprint(note)
+            if format == "json":
+                result = run_with_cleanup(
+                    _write_note_json(title, content, folder, project_name, 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)
+                )
+                rprint(note)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -147,6 +280,7 @@ def read_note(
     ] = None,
     page: int = 1,
     page_size: int = 10,
+    format: str = typer.Option("text", "--format", help="Output format: text or json"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
     ),
@@ -173,8 +307,14 @@ def read_note(
         project_name = project_name or config_manager.default_project
 
         with force_routing(local=local, cloud=cloud):
-            note = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size))
-        rprint(note)
+            if format == "json":
+                result = run_with_cleanup(
+                    _read_note_json(identifier, project_name, page, page_size)
+                )
+                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))
+                rprint(note)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -197,6 +337,7 @@ def build_context(
     page: int = 1,
     page_size: int = 10,
     max_related: int = 10,
+    format: str = typer.Option("json", "--format", help="Output format: text or json"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
     ),
@@ -234,9 +375,6 @@ def build_context(
                     max_related=max_related,
                 )
             )
-        # Use json module for more controlled serialization
-        import json
-
         context_dict = context.model_dump(exclude_none=True)
         print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str))
     except ValueError as e:
@@ -252,8 +390,17 @@ def build_context(
 @tool_app.command()
 def recent_activity(
     type: Annotated[Optional[List[SearchItemType]], typer.Option()] = None,
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
     depth: Optional[int] = 1,
     timeframe: Optional[TimeFrame] = "7d",
+    page: int = typer.Option(1, "--page", help="Page number for pagination (JSON format)"),
+    page_size: int = typer.Option(
+        50, "--page-size", help="Number of results per page (JSON format)"
+    ),
+    format: str = typer.Option("text", "--format", help="Output format: text or json"),
     local: bool = typer.Option(
         False, "--local", help="Force local API routing (ignore cloud mode)"
     ),
@@ -267,16 +414,33 @@ def recent_activity(
     try:
         validate_routing_flags(local, cloud)
 
+        # Resolve project from config for JSON mode
+        config_manager = ConfigManager()
+        project_name = None
+        if project is not None:
+            project_name, _ = config_manager.get_project(project)
+            if not project_name:
+                typer.echo(f"No project found named: {project}", err=True)
+                raise typer.Exit(1)
+        project_name = project_name or config_manager.default_project
+
         with force_routing(local=local, cloud=cloud):
-            result = run_with_cleanup(
-                mcp_recent_activity.fn(
-                    type=type,  # pyright: ignore [reportArgumentType]
-                    depth=depth,
-                    timeframe=timeframe,
+            if format == "json":
+                result = run_with_cleanup(
+                    _recent_activity_json(type, depth, timeframe, project_name, page, page_size)
                 )
-            )
-        # The tool now returns a formatted string directly
-        print(result)
+                print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
+            else:
+                result = run_with_cleanup(
+                    mcp_recent_activity.fn(
+                        type=type,  # pyright: ignore [reportArgumentType]
+                        depth=depth,
+                        timeframe=timeframe,
+                        project=project_name,
+                    )
+                )
+                # The tool returns a formatted string directly
+                print(result)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)