admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Paul Hernandez
2026-02-08 14:56:29 -06:00
committed by GitHub
parent c46d7a6833
commit a47c9c021f
3 changed files with 641 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/cli/commands/tool.py
+182 -18
View File
@@ -12,6 +12,13 @@ from basic_memory.cli.app import app
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 read_note as mcp_read_note
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)