mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
feat: add --format json support to CLI tool commands
Addresses issue #553 - CLI JSON improvements for OpenClaw plugin integration Changes: - Added --format json flag to write-note, read-note, recent-activity - Added --page and --page-size options to recent-activity (fixes hardcoded pagination) - Added --project parameter to recent-activity - Added --format json to build-context for CLI consistency - Implemented JSON helpers that call API directly to avoid double round-trips - _write_note_json() returns entity metadata without second lookup - _read_note_json() handles titles, permalinks, and memory:// URLs correctly - _recent_activity_json() supports pagination parameters Tests: - Comprehensive test coverage in tests/cli/test_cli_json_format.py - Tests for JSON format output - Tests for pagination parameters - Tests for identifier handling (titles, permalinks, URLs) - Tests for project parameter interaction Resolves: #553 Co-authored-by: bm-clawd <bm-clawd@users.noreply.github.com> Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
This commit is contained in:
claude[bot]
@@ -33,6 +33,147 @@ tool_app = typer.Typer()
|
||||
app.add_typer(tool_app, name="tool", help="Access to MCP tools via CLI")
|
||||
|
||||
|
||||
# --- JSON Format Helpers ---
|
||||
# These helpers provide JSON output for OpenClaw plugin integration
|
||||
# They call the API directly to avoid double round-trips (see issue #553)
|
||||
|
||||
|
||||
async def _write_note_json(
|
||||
title: str,
|
||||
content: str | None,
|
||||
directory: str,
|
||||
project_name: str | None,
|
||||
tags: list[str] | None,
|
||||
) -> dict:
|
||||
"""Write a note and return JSON response with entity metadata.
|
||||
|
||||
Calls the API directly to avoid the double round-trip issue where
|
||||
the MCP tool writes via one API call, then we'd need another to
|
||||
get the structured data back.
|
||||
"""
|
||||
from basic_memory.mcp.async_client import get_client
|
||||
from basic_memory.mcp.clients import KnowledgeClient
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
from basic_memory.schemas.base import Entity
|
||||
from basic_memory.utils import parse_tags
|
||||
|
||||
async with get_client() as client:
|
||||
# Get active project
|
||||
active_project = await get_active_project(client, project_name, None)
|
||||
|
||||
# Normalize directory
|
||||
if directory == "/":
|
||||
directory = ""
|
||||
|
||||
# Process tags
|
||||
tag_list = parse_tags(tags)
|
||||
metadata = {"tags": tag_list} if tag_list else None
|
||||
|
||||
# Create entity
|
||||
entity = Entity(
|
||||
title=title,
|
||||
directory=directory,
|
||||
entity_type="note",
|
||||
content_type="text/markdown",
|
||||
content=content or "",
|
||||
entity_metadata=metadata,
|
||||
)
|
||||
|
||||
# Use typed client
|
||||
knowledge_client = KnowledgeClient(client, active_project.external_id)
|
||||
|
||||
# Try create first (optimistic)
|
||||
try:
|
||||
result = await knowledge_client.create_entity(entity.model_dump(), fast=False)
|
||||
except Exception as e:
|
||||
# If conflict, try update
|
||||
if "409" in str(e) or "conflict" in str(e).lower() or "already exists" in str(e).lower():
|
||||
if not entity.permalink:
|
||||
raise ValueError("Entity permalink required for updates")
|
||||
entity_id = await knowledge_client.resolve_entity(entity.permalink)
|
||||
result = await knowledge_client.update_entity(
|
||||
entity_id, entity.model_dump(), fast=False
|
||||
)
|
||||
else:
|
||||
raise
|
||||
|
||||
# Return the entity data as dict
|
||||
return result
|
||||
|
||||
|
||||
async def _read_note_json(identifier: str, project_name: str | None, page: int, page_size: int) -> dict:
|
||||
"""Read a note and return JSON response.
|
||||
|
||||
Handles plain titles, permalinks, and memory:// URLs via memory_url_path().
|
||||
"""
|
||||
from basic_memory.mcp.async_client import get_client
|
||||
from basic_memory.mcp.clients import KnowledgeClient
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
from basic_memory.schemas.memory import memory_url_path
|
||||
|
||||
async with get_client() as client:
|
||||
# Get active project
|
||||
active_project = await get_active_project(client, project_name, None)
|
||||
|
||||
# Process identifier (handles memory://, permalinks, plain titles)
|
||||
entity_path = memory_url_path(identifier)
|
||||
|
||||
# Use typed client
|
||||
knowledge_client = KnowledgeClient(client, active_project.external_id)
|
||||
|
||||
# Resolve and fetch
|
||||
entity_id = await knowledge_client.resolve_entity(entity_path)
|
||||
entity_data = await knowledge_client.get_entity(entity_id)
|
||||
|
||||
return entity_data
|
||||
|
||||
|
||||
async def _recent_activity_json(
|
||||
type: list[SearchItemType] | None,
|
||||
depth: int,
|
||||
timeframe: TimeFrame,
|
||||
project: str | None,
|
||||
page: int,
|
||||
page_size: int,
|
||||
) -> dict:
|
||||
"""Get recent activity and return JSON response.
|
||||
|
||||
Now supports pagination via page and page_size parameters (issue #553).
|
||||
"""
|
||||
from basic_memory.mcp.async_client import get_client
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
from basic_memory.mcp.tools.utils import call_get
|
||||
from basic_memory.schemas.memory import GraphContext
|
||||
|
||||
async with get_client() as client:
|
||||
# Get active project
|
||||
active_project = await get_active_project(client, project, None)
|
||||
|
||||
# Build params
|
||||
params = {
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"max_related": 10,
|
||||
}
|
||||
if depth:
|
||||
params["depth"] = depth
|
||||
if timeframe:
|
||||
params["timeframe"] = timeframe # type: ignore
|
||||
if type:
|
||||
params["type"] = [t.value for t in type] # type: ignore
|
||||
|
||||
# Call API
|
||||
response = await call_get(
|
||||
client,
|
||||
f"/v2/projects/{active_project.external_id}/memory/recent",
|
||||
params=params,
|
||||
)
|
||||
|
||||
# Return as dict
|
||||
activity_data = GraphContext.model_validate(response.json())
|
||||
return activity_data.model_dump(exclude_none=True)
|
||||
|
||||
|
||||
@tool_app.command()
|
||||
def write_note(
|
||||
title: Annotated[str, typer.Option(help="The title of the note")],
|
||||
@@ -52,6 +193,10 @@ def write_note(
|
||||
tags: Annotated[
|
||||
Optional[List[str]], typer.Option(help="A list of tags to apply to the note")
|
||||
] = None,
|
||||
format: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(help="Output format: 'text' (default) or 'json'"),
|
||||
] = None,
|
||||
local: bool = typer.Option(
|
||||
False, "--local", help="Force local API routing (ignore cloud mode)"
|
||||
),
|
||||
@@ -124,8 +269,14 @@ def write_note(
|
||||
project_name = project_name or config_manager.default_project
|
||||
|
||||
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":
|
||||
# For JSON output, bypass the MCP tool and call the API directly
|
||||
# to avoid the double round-trip mentioned in issue #553
|
||||
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:
|
||||
result = run_with_cleanup(mcp_write_note.fn(title, content, folder, project_name, tags))
|
||||
rprint(result)
|
||||
except ValueError as e:
|
||||
typer.echo(f"Error: {e}", err=True)
|
||||
raise typer.Exit(1)
|
||||
@@ -147,6 +298,10 @@ def read_note(
|
||||
] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 10,
|
||||
format: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(help="Output format: 'text' (default) or 'json'"),
|
||||
] = None,
|
||||
local: bool = typer.Option(
|
||||
False, "--local", help="Force local API routing (ignore cloud mode)"
|
||||
),
|
||||
@@ -173,8 +328,13 @@ 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":
|
||||
# For JSON output, use helper that returns structured data
|
||||
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:
|
||||
result = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size))
|
||||
rprint(result)
|
||||
except ValueError as e:
|
||||
typer.echo(f"Error: {e}", err=True)
|
||||
raise typer.Exit(1)
|
||||
@@ -197,6 +357,10 @@ def build_context(
|
||||
page: int = 1,
|
||||
page_size: int = 10,
|
||||
max_related: int = 10,
|
||||
format: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(help="Output format: 'json' (always outputs JSON, flag added for consistency)"),
|
||||
] = None,
|
||||
local: bool = typer.Option(
|
||||
False, "--local", help="Force local API routing (ignore cloud mode)"
|
||||
),
|
||||
@@ -254,6 +418,18 @@ def recent_activity(
|
||||
type: Annotated[Optional[List[SearchItemType]], typer.Option()] = None,
|
||||
depth: Optional[int] = 1,
|
||||
timeframe: Optional[TimeFrame] = "7d",
|
||||
project: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(
|
||||
help="The project to use. If not provided, the default project will be used."
|
||||
),
|
||||
] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
format: Annotated[
|
||||
Optional[str],
|
||||
typer.Option(help="Output format: 'text' (default) or 'json'"),
|
||||
] = None,
|
||||
local: bool = typer.Option(
|
||||
False, "--local", help="Force local API routing (ignore cloud mode)"
|
||||
),
|
||||
@@ -267,16 +443,43 @@ def recent_activity(
|
||||
try:
|
||||
validate_routing_flags(local, cloud)
|
||||
|
||||
# look for the project in the config
|
||||
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)
|
||||
|
||||
# use the project name, or the default from the config
|
||||
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":
|
||||
# For JSON output, use helper with pagination support
|
||||
result = run_with_cleanup(
|
||||
_recent_activity_json(
|
||||
type=type, # pyright: ignore [reportArgumentType]
|
||||
depth=depth or 1,
|
||||
timeframe=timeframe or "7d",
|
||||
project=project_name,
|
||||
page=page,
|
||||
page_size=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)
|
||||
|
||||
@@ -0,0 +1,282 @@
|
||||
"""Tests for CLI --format json functionality.
|
||||
|
||||
These tests verify that the --format json flag works correctly for
|
||||
write-note, read-note, recent-activity, and build-context commands.
|
||||
|
||||
See issue #553 for context on these improvements.
|
||||
"""
|
||||
|
||||
import json
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from basic_memory.cli.main import app as cli_app
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
class TestWriteNoteJsonFormat:
|
||||
"""Tests for write-note --format json."""
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._write_note_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_write_note_json_format(self, mock_config_manager, mock_write_json):
|
||||
"""Test write-note with --format json returns structured data."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the JSON helper to return entity data
|
||||
mock_entity_data = {
|
||||
"id": "123",
|
||||
"title": "Test Note",
|
||||
"permalink": "test-note",
|
||||
"file_path": "/tmp/test/test-note.md",
|
||||
"checksum": "abc123",
|
||||
}
|
||||
mock_write_json.return_value = mock_entity_data
|
||||
|
||||
# Run command
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
[
|
||||
"tool",
|
||||
"write-note",
|
||||
"--title",
|
||||
"Test Note",
|
||||
"--folder",
|
||||
"notes",
|
||||
"--content",
|
||||
"Test content",
|
||||
"--format",
|
||||
"json",
|
||||
],
|
||||
)
|
||||
|
||||
# Verify success
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify JSON output
|
||||
output = json.loads(result.output)
|
||||
assert output["title"] == "Test Note"
|
||||
assert output["permalink"] == "test-note"
|
||||
assert "checksum" in output
|
||||
|
||||
# Verify the JSON helper was called (not the MCP tool)
|
||||
mock_write_json.assert_called_once()
|
||||
|
||||
|
||||
class TestReadNoteJsonFormat:
|
||||
"""Tests for read-note --format json."""
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._read_note_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_read_note_json_format(self, mock_config_manager, mock_read_json):
|
||||
"""Test read-note with --format json returns structured data."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the JSON helper to return entity data
|
||||
mock_entity_data = {
|
||||
"id": "123",
|
||||
"title": "Test Note",
|
||||
"permalink": "test-note",
|
||||
"content": "Test content",
|
||||
}
|
||||
mock_read_json.return_value = mock_entity_data
|
||||
|
||||
# Run command
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
["tool", "read-note", "test-note", "--format", "json"],
|
||||
)
|
||||
|
||||
# Verify success
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify JSON output
|
||||
output = json.loads(result.output)
|
||||
assert output["title"] == "Test Note"
|
||||
assert output["content"] == "Test content"
|
||||
|
||||
# Verify the JSON helper was called
|
||||
mock_read_json.assert_called_once()
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._read_note_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_read_note_handles_plain_title(self, mock_config_manager, mock_read_json):
|
||||
"""Test read-note with plain title (not permalink) works with JSON format."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the JSON helper
|
||||
mock_entity_data = {"title": "My Note", "permalink": "my-note"}
|
||||
mock_read_json.return_value = mock_entity_data
|
||||
|
||||
# Run with plain title (should be handled by memory_url_path in helper)
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
["tool", "read-note", "My Note", "--format", "json"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
output = json.loads(result.output)
|
||||
assert output["title"] == "My Note"
|
||||
|
||||
|
||||
class TestRecentActivityJsonFormat:
|
||||
"""Tests for recent-activity --format json."""
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._recent_activity_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_recent_activity_json_format(self, mock_config_manager, mock_activity_json):
|
||||
"""Test recent-activity with --format json returns structured data."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the JSON helper to return activity data
|
||||
mock_activity_data = {
|
||||
"results": [
|
||||
{"primary_result": {"title": "Note 1", "type": "entity"}},
|
||||
{"primary_result": {"title": "Note 2", "type": "entity"}},
|
||||
],
|
||||
"metadata": {"total_results": 2},
|
||||
}
|
||||
mock_activity_json.return_value = mock_activity_data
|
||||
|
||||
# Run command
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
["tool", "recent-activity", "--format", "json"],
|
||||
)
|
||||
|
||||
# Verify success
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify JSON output
|
||||
output = json.loads(result.output)
|
||||
assert len(output["results"]) == 2
|
||||
assert output["metadata"]["total_results"] == 2
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._recent_activity_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_recent_activity_pagination_options(self, mock_config_manager, mock_activity_json):
|
||||
"""Test recent-activity exposes --page and --page-size options."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the JSON helper
|
||||
mock_activity_data = {"results": [], "metadata": {}}
|
||||
mock_activity_json.return_value = mock_activity_data
|
||||
|
||||
# Run command with pagination
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
[
|
||||
"tool",
|
||||
"recent-activity",
|
||||
"--format",
|
||||
"json",
|
||||
"--page",
|
||||
"2",
|
||||
"--page-size",
|
||||
"25",
|
||||
],
|
||||
)
|
||||
|
||||
# Verify success
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify pagination params were passed to helper
|
||||
call_args = mock_activity_json.call_args
|
||||
assert call_args.kwargs["page"] == 2
|
||||
assert call_args.kwargs["page_size"] == 25
|
||||
|
||||
@patch("basic_memory.cli.commands.tool._recent_activity_json")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_recent_activity_project_param(self, mock_config_manager, mock_activity_json):
|
||||
"""Test recent-activity --project works with --format json."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("my-project", "/tmp/my-proj")
|
||||
|
||||
# Mock the JSON helper
|
||||
mock_activity_data = {"results": [], "metadata": {}}
|
||||
mock_activity_json.return_value = mock_activity_data
|
||||
|
||||
# Run command with project
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
[
|
||||
"tool",
|
||||
"recent-activity",
|
||||
"--format",
|
||||
"json",
|
||||
"--project",
|
||||
"my-project",
|
||||
],
|
||||
)
|
||||
|
||||
# Verify success
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify project was resolved
|
||||
mock_config_manager.return_value.get_project.assert_called_with("my-project")
|
||||
|
||||
|
||||
class TestBuildContextFormatFlag:
|
||||
"""Tests for build-context --format json consistency."""
|
||||
|
||||
@patch("basic_memory.cli.commands.tool.mcp_build_context")
|
||||
@patch("basic_memory.cli.commands.tool.ConfigManager")
|
||||
def test_build_context_accepts_format_flag(self, mock_config_manager, mock_build_context):
|
||||
"""Test build-context accepts --format json for consistency."""
|
||||
# Mock config
|
||||
mock_config = MagicMock()
|
||||
mock_config.default_project = "test-project"
|
||||
mock_config_manager.return_value = mock_config
|
||||
mock_config_manager.return_value.get_project.return_value = ("test-project", "/tmp/test")
|
||||
|
||||
# Mock the MCP tool
|
||||
mock_context_result = MagicMock()
|
||||
mock_context_result.model_dump.return_value = {
|
||||
"primary_result": {"title": "Test"},
|
||||
"related_results": [],
|
||||
}
|
||||
mock_build_context.fn.return_value = mock_context_result
|
||||
|
||||
# Run command with format flag (build-context always outputs JSON)
|
||||
result = runner.invoke(
|
||||
cli_app,
|
||||
[
|
||||
"tool",
|
||||
"build-context",
|
||||
"memory://test",
|
||||
"--format",
|
||||
"json",
|
||||
],
|
||||
)
|
||||
|
||||
# Verify command accepts the flag without error
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Verify output is valid JSON
|
||||
output = json.loads(result.output)
|
||||
assert "primary_result" in output
|
||||
Reference in New Issue
Block a user