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 JSON output mode for BM MCP tools (default text) (#585)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
Signed-off-by: Paul Hernandez <60959+phernandez@users.noreply.github.com>
This commit is contained in:
Paul Hernandez
2026-02-18 22:18:29 -06:00
committed by GitHub
parent f2683291e4
commit 0cb3f95d67
22 changed files with 1452 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12 -9
View File
@@ -418,19 +418,19 @@ basic-memory tool edit-note docs/setup --operation append --content $'\n- Added
**Content Management:**
```
write_note(title, content, folder, tags) - Create or update notes
read_note(identifier, page, page_size) - Read notes by title or permalink
write_note(title, content, folder, tags, output_format="text"|"json") - Create or update notes
read_note(identifier, page, page_size, output_format="text"|"json") - Read notes by title or permalink
read_content(path) - Read raw file content (text, images, binaries)
view_note(identifier) - View notes as formatted artifacts
edit_note(identifier, operation, content) - Edit notes incrementally
move_note(identifier, destination_path) - Move notes with database consistency
delete_note(identifier) - Delete notes from knowledge base
edit_note(identifier, operation, content, output_format="text"|"json") - Edit notes incrementally
move_note(identifier, destination_path, output_format="text"|"json") - Move notes with database consistency
delete_note(identifier, output_format="text"|"json") - Delete notes from knowledge base
```
**Knowledge Graph Navigation:**
```
build_context(url, depth, timeframe) - Navigate knowledge graph via memory:// URLs
recent_activity(type, depth, timeframe) - Find recently updated information
build_context(url, depth, timeframe, output_format="json"|"text") - Navigate knowledge graph via memory:// URLs
recent_activity(type, depth, timeframe, output_format="text"|"json") - Find recently updated information
list_directory(dir_name, depth) - Browse directory contents with filtering
```
@@ -443,12 +443,15 @@ search_by_metadata(filters, limit, offset, project) - Structured frontmatter sea
**Project Management:**
```
list_memory_projects() - List all available projects
create_memory_project(project_name, project_path) - Create new projects
list_memory_projects(output_format="text"|"json") - List all available projects
create_memory_project(project_name, project_path, output_format="text"|"json") - Create new projects
get_current_project() - Show current project stats
sync_status() - Check synchronization status
```
`output_format` defaults to `"text"` for these tools, preserving current human-readable responses.
`build_context` defaults to `"json"` and can be switched to `"text"` when compact markdown output is preferred.
**Cloud Discovery (opt-in):**
```
cloud_info() - Show optional Cloud overview and setup guidance
docs/mcp-ui-bakeoff-instructions.md
+15 -7
View File
@@ -83,18 +83,26 @@ Manual check:
---
### 2) ASCII / ANSI TUI Output
### 2) Text / JSON Output Modes
Tools:
- `search_notes(output_format="ascii" | "ansi")`
- `read_note(output_format="ascii" | "ansi")`
- `search_notes(output_format="text" | "json")`
- `read_note(output_format="text" | "json")`
- `write_note(output_format="text" | "json")`
- `edit_note(output_format="text" | "json")`
- `recent_activity(output_format="text" | "json")`
- `list_memory_projects(output_format="text" | "json")`
- `create_memory_project(output_format="text" | "json")`
- `delete_note(output_format="text" | "json")`
- `move_note(output_format="text" | "json")`
- `build_context(output_format="json" | "text")`
Expect:
- ASCII table for search, header + content preview for note.
- ANSI variants include color escape codes.
- `text` mode preserves existing human-readable responses.
- `json` mode returns structured dict/list payloads for machine-readable clients.
Automated:
- `uv run pytest test-int/mcp/test_output_format_ascii_integration.py`
- `uv run pytest test-int/mcp/test_output_format_json_integration.py`
---
@@ -125,6 +133,6 @@ Fill in after running:
- Tool‑UI (React): __
- MCP‑UI SDK (embedded): __
- ASCII/ANSI: __
- Text/JSON modes: __
Decision + rationale: __
docs/post-v0.18.0-test-plan.md
+2 -2
View File
@@ -180,7 +180,7 @@ Key finding: **FastEmbed (384-d local ONNX) matches or exceeds OpenAI (1536-d) q
### Existing coverage anchor points
- `tests/mcp/test_tool_contracts.py`
- `test-int/mcp/test_output_format_ascii_integration.py`
- `test-int/mcp/test_output_format_json_integration.py`
- `test-int/mcp/test_ui_sdk_integration.py`
### Gaps to close — DONE
@@ -288,7 +288,7 @@ Run after automated tests pass.
- Routing: verify success/failure paths with and without API key.
- Permalink routing: read/write/search notes across projects with colliding titles.
- Permalink routing: verify memory URL routing correctness.
- UI/TUI: call `search_notes` and `read_note` with UI variants and `output_format=ascii|ansi`.
- UI/TUI: call `search_notes` and `read_note` with UI variants and `output_format=text|json`.
- UI/TUI: verify payload/resource format and metadata completeness.
## Implementation Backlog (Ordered)
src/basic_memory/cli/commands/tool.py
+12 -6
View File
@@ -160,11 +160,14 @@ async def _read_note_json(
search_type="title",
project=project_name,
workspace=workspace,
output_format="json",
)
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)
results = title_results.get("results", []) if isinstance(title_results, dict) else []
if results:
result = results[0]
permalink = result.get("permalink")
if permalink:
entity_id = await knowledge_client.resolve_entity(permalink)
if entity_id is None:
raise ValueError(f"Could not find note matching: {identifier}")
@@ -635,10 +638,13 @@ def build_context(
page=page,
page_size=page_size,
max_related=max_related,
output_format="text" if format == "text" else "json",
)
)
# build_context now returns a slimmed dict (already serializable)
print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
if format == "json":
print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
else:
print(result)
except ValueError as e:
typer.echo(f"Error: {e}", err=True)
raise typer.Exit(1)
src/basic_memory/mcp/tools/build_context.py
+10 -9
View File
@@ -1,6 +1,6 @@
"""Build context tool for Basic Memory MCP server."""
from typing import Optional
from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
@@ -190,7 +190,7 @@ def _format_context_markdown(graph: GraphContext, project: str) -> str:
Format options:
- "json" (default): Slimmed JSON with redundant fields removed
- "markdown": Compact markdown text for LLM consumption
- "text": Compact markdown text for LLM consumption
""",
)
async def build_context(
@@ -202,7 +202,7 @@ async def build_context(
page: int = 1,
page_size: int = 10,
max_related: int = 10,
format: str = "json",
output_format: Literal["json", "text"] = "json",
context: Context | None = None,
) -> dict | str:
"""Get context needed to continue a discussion within a specific project.
@@ -225,12 +225,13 @@ async def build_context(
page: Page number of results to return (default: 1)
page_size: Number of results to return per page (default: 10)
max_related: Maximum number of related results to return (default: 10)
format: Response format - "json" for slimmed JSON dict, "markdown" for compact text
output_format: Response format - "json" for slimmed JSON dict,
"text" for compact markdown text
context: Optional FastMCP context for performance caching.
Returns:
dict (format="json"): Slimmed JSON with redundant fields removed
str (format="markdown"): Compact markdown representation
dict (output_format="json"): Slimmed JSON with redundant fields removed
str (output_format="text"): Compact markdown representation
Examples:
# Continue a specific discussion
@@ -239,8 +240,8 @@ async def build_context(
# Get deeper context about a component
build_context("work-docs", "memory://components/memory-service", depth=2)
# Get markdown output for compact context
build_context("research", "memory://specs/search", format="markdown")
# Get text output for compact context
build_context("research", "memory://specs/search", output_format="text")
Raises:
ToolError: If project doesn't exist or depth parameter is invalid
@@ -276,7 +277,7 @@ async def build_context(
max_related=max_related,
)
if format == "markdown":
if output_format == "text":
return _format_context_markdown(graph, active_project.name)
return _slim_context(graph)
src/basic_memory/mcp/tools/chatgpt_tools.py
+29 -8
View File
@@ -13,22 +13,41 @@ from fastmcp import Context
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.search import search_notes
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.schemas.search import SearchResponse
from basic_memory.config import ConfigManager
from basic_memory.schemas.search import SearchResponse, SearchResult
def _format_search_results_for_chatgpt(results: SearchResponse) -> List[Dict[str, Any]]:
def _format_search_results_for_chatgpt(
results: SearchResponse | list[SearchResult] | list[dict[str, Any]] | dict[str, Any],
) -> List[Dict[str, Any]]:
"""Format search results according to ChatGPT's expected schema.
Returns a list of result objects with id, title, and url fields.
"""
if isinstance(results, SearchResponse):
raw_results: list[SearchResult] | list[dict[str, Any]] = results.results
elif isinstance(results, dict):
nested_results = results.get("results")
raw_results = nested_results if isinstance(nested_results, list) else []
else:
raw_results = results
formatted_results = []
for result in results.results:
for result in raw_results:
if isinstance(result, SearchResult):
title = result.title
permalink = result.permalink
elif isinstance(result, dict):
title = result.get("title")
permalink = result.get("permalink")
else:
raise TypeError(f"Unexpected result type: {type(result).__name__}")
formatted_result = {
"id": result.permalink or f"doc-{len(formatted_results)}",
"title": result.title if result.title and result.title.strip() else "Untitled",
"url": result.permalink or "",
"id": permalink or f"doc-{len(formatted_results)}",
"title": title if isinstance(title, str) and title.strip() else "Untitled",
"url": permalink or "",
}
formatted_results.append(formatted_result)
@@ -102,6 +121,7 @@ async def search(
page=1,
page_size=10, # Reasonable default for ChatGPT consumption
search_type="text", # Default to full-text search
output_format="json",
context=context,
)
@@ -115,10 +135,11 @@ async def search(
}
else:
# Format successful results for ChatGPT
formatted_results = _format_search_results_for_chatgpt(results)
raw_results = results.get("results", []) if isinstance(results, dict) else []
formatted_results = _format_search_results_for_chatgpt(raw_results)
search_results = {
"results": formatted_results,
"total_count": len(results.results), # Use actual count from results
"total_count": len(raw_results), # Use actual count from results
"query": query,
}
logger.info(f"Search completed: {len(formatted_results)} results returned")
src/basic_memory/mcp/tools/delete_note.py
+69 -2
View File
@@ -1,5 +1,5 @@
from textwrap import dedent
from typing import Optional
from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
@@ -152,8 +152,9 @@ async def delete_note(
is_directory: bool = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> bool | str:
) -> bool | str | dict:
"""Delete a note or directory from the knowledge base.
Permanently removes a note or directory from the specified project. For single notes,
@@ -174,6 +175,8 @@ async def delete_note(
(without file extensions). Defaults to False.
project: Project name to delete from. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
output_format: "text" preserves existing behavior (bool/string). "json"
returns machine-readable deletion metadata.
context: Optional FastMCP context for performance caching.
Returns:
@@ -231,6 +234,15 @@ async def delete_note(
if is_directory:
try:
result = await knowledge_client.delete_directory(identifier)
if output_format == "json":
return {
"deleted": result.failed_deletes == 0,
"is_directory": True,
"identifier": identifier,
"total_files": result.total_files,
"successful_deletes": result.successful_deletes,
"failed_deletes": result.failed_deletes,
}
# Build success message for directory delete
result_lines = [
@@ -269,6 +281,16 @@ async def delete_note(
except Exception as e: # pragma: no cover
logger.error(f"Directory delete failed for '{identifier}': {e}")
if output_format == "json":
return {
"deleted": False,
"is_directory": True,
"identifier": identifier,
"total_files": 0,
"successful_deletes": 0,
"failed_deletes": 0,
"error": str(e),
}
return f"""# Directory Delete Failed
Error deleting directory '{identifier}': {str(e)}
@@ -288,18 +310,41 @@ delete_note("path/to/file.md")
```"""
# Handle single note deletes
note_title = None
note_permalink = None
note_file_path = None
try:
# Resolve identifier to entity ID
entity_id = await knowledge_client.resolve_entity(identifier)
if output_format == "json":
entity = await knowledge_client.get_entity(entity_id)
note_title = entity.title
note_permalink = entity.permalink
note_file_path = entity.file_path
except ToolError as e:
# If entity not found, return False (note doesn't exist)
if "Entity not found" in str(e) or "not found" in str(e).lower():
logger.warning(f"Note not found for deletion: {identifier}")
if output_format == "json":
return {
"deleted": False,
"title": None,
"permalink": None,
"file_path": None,
}
return False
# For other resolution errors, return formatted error message
logger.error( # pragma: no cover
f"Delete failed for '{identifier}': {e}, project: {active_project.name}"
)
if output_format == "json":
return {
"deleted": False,
"title": None,
"permalink": None,
"file_path": None,
"error": str(e),
}
return _format_delete_error_response( # pragma: no cover
active_project.name, str(e), identifier
)
@@ -312,14 +357,36 @@ delete_note("path/to/file.md")
logger.info(
f"Successfully deleted note: {identifier} in project: {active_project.name}"
)
if output_format == "json":
return {
"deleted": True,
"title": note_title,
"permalink": note_permalink,
"file_path": note_file_path,
}
return True
else:
logger.warning( # pragma: no cover
f"Delete operation completed but note was not deleted: {identifier}"
)
if output_format == "json":
return {
"deleted": False,
"title": note_title,
"permalink": note_permalink,
"file_path": note_file_path,
}
return False # pragma: no cover
except Exception as e: # pragma: no cover
logger.error(f"Delete failed for '{identifier}': {e}, project: {active_project.name}")
if output_format == "json":
return {
"deleted": False,
"title": note_title,
"permalink": note_permalink,
"file_path": note_file_path,
"error": str(e),
}
# Return formatted error message for better user experience
return _format_delete_error_response(active_project.name, str(e), identifier)
src/basic_memory/mcp/tools/edit_note.py
+25 -4
View File
@@ -1,6 +1,6 @@
"""Edit note tool for Basic Memory MCP server."""
from typing import Optional
from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
@@ -135,8 +135,9 @@ async def edit_note(
section: Optional[str] = None,
find_text: Optional[str] = None,
expected_replacements: int = 1,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str:
) -> str | dict:
"""Edit an existing markdown note in the knowledge base.
Makes targeted changes to existing notes without rewriting the entire content.
@@ -160,6 +161,8 @@ async def edit_note(
section: For replace_section operation - the markdown header to replace content under (e.g., "## Notes", "### Implementation")
find_text: For find_replace operation - the text to find and replace
expected_replacements: For find_replace operation - the expected number of replacements (validation will fail if actual doesn't match)
output_format: "text" returns the existing markdown summary. "json" returns
machine-readable edit metadata.
context: Optional FastMCP context for performance caching.
Returns:
@@ -285,7 +288,7 @@ async def edit_note(
for obs in result.observations:
categories[obs.category] = categories.get(obs.category, 0) + 1
summary.append("\\n## Observations")
summary.append("\n## Observations")
for category, count in sorted(categories.items()):
summary.append(f"- {category}: {count}")
@@ -296,7 +299,7 @@ async def edit_note(
unresolved = sum(1 for r in result.relations if not r.to_id)
resolved = len(result.relations) - unresolved
summary.append("\\n## Relations")
summary.append("\n## Relations")
summary.append(f"- Resolved: {resolved}")
if unresolved:
summary.append(f"- Unresolved: {unresolved}")
@@ -311,11 +314,29 @@ async def edit_note(
relations_count=len(result.relations),
)
if output_format == "json":
return {
"title": result.title,
"permalink": result.permalink,
"file_path": result.file_path,
"checksum": result.checksum,
"operation": operation,
}
summary_result = "\n".join(summary)
return add_project_metadata(summary_result, active_project.name)
except Exception as e:
logger.error(f"Error editing note: {e}")
if output_format == "json":
return {
"title": None,
"permalink": None,
"file_path": None,
"checksum": None,
"operation": operation,
"error": str(e),
}
return _format_error_response(
str(e), operation, identifier, find_text, expected_replacements, active_project.name
)
src/basic_memory/mcp/tools/move_note.py
+113 -22
View File
@@ -1,7 +1,7 @@
"""Move note tool for Basic Memory MCP server."""
from textwrap import dedent
from typing import Optional
from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
@@ -349,8 +349,9 @@ async def move_note(
is_directory: bool = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str:
) -> str | dict:
"""Move a note or directory to a new location within the same project.
Moves a note or directory from one location to another within the project,
@@ -369,6 +370,8 @@ async def move_note(
(without file extensions). Defaults to False.
project: Project name to move within. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
output_format: "text" returns existing markdown guidance/success text. "json"
returns machine-readable move metadata.
context: Optional FastMCP context for performance caching.
Returns:
@@ -425,6 +428,16 @@ async def move_note(
destination_path=destination_path,
project=active_project.name,
)
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"error": "SECURITY_VALIDATION_ERROR",
}
return f"""# Move Failed - Security Validation Error
The destination path '{destination_path}' is not allowed - paths must stay within project boundaries.
@@ -448,6 +461,19 @@ move_note("{identifier}", "notes/{destination_path.split("/")[-1] if "/" in dest
try:
result = await knowledge_client.move_directory(identifier, destination_path)
if output_format == "json":
return {
"moved": result.failed_moves == 0,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"is_directory": True,
"total_files": result.total_files,
"successful_moves": result.successful_moves,
"failed_moves": result.failed_moves,
}
# Build success message for directory move
result_lines = [
@@ -489,6 +515,17 @@ move_note("{identifier}", "notes/{destination_path.split("/")[-1] if "/" in dest
logger.error(
f"Directory move failed for '{identifier}' to '{destination_path}': {e}"
)
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"is_directory": True,
"error": str(e),
}
return f"""# Directory Move Failed
Error moving directory '{identifier}' to '{destination_path}': {str(e)}
@@ -513,6 +550,16 @@ move_note("path/to/file.md", "{destination_path}/file.md")
)
if cross_project_error:
logger.info(f"Detected cross-project move attempt: {identifier} -> {destination_path}")
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"error": "CROSS_PROJECT_MOVE_NOT_SUPPORTED",
}
return cross_project_error
# Import here to avoid circular import
@@ -521,22 +568,40 @@ move_note("path/to/file.md", "{destination_path}/file.md")
# Use typed KnowledgeClient for API calls
knowledge_client = KnowledgeClient(client, active_project.external_id)
# Get the source entity information for extension validation
# Resolve once and reuse the entity ID across extension validation and move.
source_ext = "md" # Default to .md if we can't determine source extension
resolved_entity_id: str | None = None
source_entity = None
async def _ensure_resolved_entity_id() -> str:
"""Resolve and cache the source entity ID for the duration of this move."""
nonlocal resolved_entity_id
if resolved_entity_id is None:
resolved_entity_id = await knowledge_client.resolve_entity(identifier)
return resolved_entity_id
try:
# Resolve identifier to entity ID
entity_id = await knowledge_client.resolve_entity(identifier)
# Fetch source entity information to get the current file extension
source_entity = await knowledge_client.get_entity(entity_id)
resolved_entity_id = await _ensure_resolved_entity_id()
source_entity = await knowledge_client.get_entity(resolved_entity_id)
if "." in source_entity.file_path:
source_ext = source_entity.file_path.split(".")[-1]
except Exception as e:
# If we can't fetch the source entity, default to .md extension
# If we can't fetch source metadata, continue with extension defaults.
logger.debug(f"Could not fetch source entity for extension check: {e}")
# Validate that destination path includes a file extension
if "." not in destination_path or not destination_path.split(".")[-1]:
logger.warning(f"Move failed - no file extension provided: {destination_path}")
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"error": "FILE_EXTENSION_REQUIRED",
}
return dedent(f"""
# Move Failed - File Extension Required
@@ -555,14 +620,15 @@ move_note("path/to/file.md", "{destination_path}/file.md")
All examples in Basic Memory expect file extensions to be explicitly provided.
""").strip()
# Get the source entity to check its file extension
try:
# Resolve identifier to entity ID (might already be cached from above)
entity_id = await knowledge_client.resolve_entity(identifier)
# Fetch source entity information
source_entity = await knowledge_client.get_entity(entity_id)
# Validate extension consistency when source metadata is available.
if source_entity is None:
try:
resolved_entity_id = await _ensure_resolved_entity_id()
source_entity = await knowledge_client.get_entity(resolved_entity_id)
except Exception as e:
logger.debug(f"Could not fetch source entity for extension check: {e}")
# Extract file extensions
if source_entity is not None:
source_ext = (
source_entity.file_path.split(".")[-1] if "." in source_entity.file_path else ""
)
@@ -573,6 +639,16 @@ move_note("path/to/file.md", "{destination_path}/file.md")
logger.warning(
f"Move failed - file extension mismatch: source={source_ext}, dest={dest_ext}"
)
if output_format == "json":
return {
"moved": False,
"title": source_entity.title,
"permalink": source_entity.permalink,
"file_path": source_entity.file_path,
"source": identifier,
"destination": destination_path,
"error": "FILE_EXTENSION_MISMATCH",
}
return dedent(f"""
# Move Failed - File Extension Mismatch
@@ -589,17 +665,22 @@ move_note("path/to/file.md", "{destination_path}/file.md")
move_note("{identifier}", "{destination_path.rsplit(".", 1)[0]}.{source_ext}")
```
""").strip()
except Exception as e:
# If we can't fetch the source entity, log it but continue
# This might happen if the identifier is not yet resolved
logger.debug(f"Could not fetch source entity for extension check: {e}")
try:
# Resolve identifier to entity ID for the move operation
entity_id = await knowledge_client.resolve_entity(identifier)
# Resolve identifier only if earlier checks could not.
resolved_entity_id = await _ensure_resolved_entity_id()
# Call the move API using KnowledgeClient
result = await knowledge_client.move_entity(entity_id, destination_path)
result = await knowledge_client.move_entity(resolved_entity_id, destination_path)
if output_format == "json":
return {
"moved": True,
"title": result.title,
"permalink": result.permalink,
"file_path": result.file_path,
"source": identifier,
"destination": destination_path,
}
# Build success message
result_lines = [
@@ -624,5 +705,15 @@ move_note("path/to/file.md", "{destination_path}/file.md")
except Exception as e:
logger.error(f"Move failed for '{identifier}' to '{destination_path}': {e}")
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"error": str(e),
}
# Return formatted error message for better user experience
return _format_move_error_response(str(e), identifier, destination_path)
src/basic_memory/mcp/tools/project_management.py
+97 -34
View File
@@ -5,6 +5,7 @@ and manage project context during conversations.
"""
import os
from typing import Literal
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
@@ -14,65 +15,73 @@ from basic_memory.utils import generate_permalink
@mcp.tool("list_memory_projects")
async def list_memory_projects(context: Context | None = None) -> str:
async def list_memory_projects(
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | dict:
"""List all available projects with their status.
Shows all Basic Memory projects that are available for MCP operations.
Use this tool to discover projects when you need to know which project to use.
Use this tool:
- At conversation start when project is unknown
- When user asks about available projects
- Before any operation requiring a project
After calling:
- Ask user which project to use
- Remember their choice for the session
Returns:
Formatted list of projects with session management guidance
Example:
list_memory_projects()
Args:
output_format: "text" returns the existing human-readable project list.
"json" returns structured project metadata.
context: Optional FastMCP context for progress/status logging.
"""
async with get_client() as client:
if context: # pragma: no cover
await context.info("Listing all available projects")
# Check if server is constrained to a specific project
constrained_project = os.environ.get("BASIC_MEMORY_MCP_PROJECT")
# Import here to avoid circular import
from basic_memory.mcp.clients import ProjectClient
# Use typed ProjectClient for API calls
project_client = ProjectClient(client)
project_list = await project_client.list_projects()
if output_format == "json":
projects = [
{
"name": project.name,
"path": project.path,
"is_default": project.is_default,
# Reserved for forward-compatible cloud/private project metadata.
# Local project list responses do not currently provide these values.
"is_private": False,
"display_name": None,
}
for project in project_list.projects
]
return {
"projects": projects,
"default_project": project_list.default_project,
"constrained_project": constrained_project,
}
if constrained_project:
result = f"Project: {constrained_project}\n\n"
result += "Note: This MCP server is constrained to a single project.\n"
result += "All operations will automatically use this project."
else:
# Show all projects with session guidance
result = "Available projects:\n"
return result
for project in project_list.projects:
result += f"• {project.name}\n"
result += "\n" + "─" * 40 + "\n"
result += "Next: Ask which project to use for this session.\n"
result += "Example: 'Which project should I use for this task?'\n\n"
result += "Session reminder: Track the selected project for all subsequent operations in this conversation.\n"
result += "The user can say 'switch to [project]' to change projects."
result = "Available projects:\n"
for project in project_list.projects:
result += f"• {project.name}\n"
result += "\n" + "─" * 40 + "\n"
result += "Next: Ask which project to use for this session.\n"
result += "Example: 'Which project should I use for this task?'\n\n"
result += "Session reminder: Track the selected project for all subsequent operations in this conversation.\n"
result += "The user can say 'switch to [project]' to change projects."
return result
@mcp.tool("create_memory_project")
async def create_memory_project(
project_name: str, project_path: str, set_default: bool = False, context: Context | None = None
) -> str:
project_name: str,
project_path: str,
set_default: bool = False,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | dict:
"""Create a new Basic Memory project.
Creates a new project with the specified name and path. The project directory
@@ -82,6 +91,9 @@ async def create_memory_project(
project_name: Name for the new project (must be unique)
project_path: File system path where the project will be stored
set_default: Whether to set this project as the default (optional, defaults to False)
output_format: "text" returns the existing human-readable result text.
"json" returns structured project creation metadata.
context: Optional FastMCP context for progress/status logging.
Returns:
Confirmation message with project details
@@ -94,6 +106,19 @@ async def create_memory_project(
# Check if server is constrained to a specific project
constrained_project = os.environ.get("BASIC_MEMORY_MCP_PROJECT")
if constrained_project:
if output_format == "json":
return {
"name": project_name,
"path": project_path,
"is_default": False,
"created": False,
"already_exists": False,
"error": "PROJECT_CONSTRAINED",
"message": (
f"Project creation disabled - MCP server is constrained to project "
f"'{constrained_project}'."
),
}
return f'# Error\n\nProject creation disabled - MCP server is constrained to project \'{constrained_project}\'.\nUse the CLI to create projects: `basic-memory project add "{project_name}" "{project_path}"`'
if context: # pragma: no cover
@@ -109,8 +134,46 @@ async def create_memory_project(
# Use typed ProjectClient for API calls
project_client = ProjectClient(client)
existing = await project_client.list_projects()
existing_match = next(
(p for p in existing.projects if p.name.casefold() == project_name.casefold()),
None,
)
if existing_match:
is_default = bool(
existing_match.is_default or existing.default_project == existing_match.name
)
if output_format == "json":
return {
"name": existing_match.name,
"path": existing_match.path,
"is_default": is_default,
"created": False,
"already_exists": True,
}
return (
f"✓ Project already exists: {existing_match.name}\n\n"
f"Project Details:\n"
f"• Name: {existing_match.name}\n"
f"• Path: {existing_match.path}\n"
f"{'• Set as default project\n' if is_default else ''}"
"\nProject is already available for use in tool calls.\n"
)
status_response = await project_client.create_project(project_request.model_dump())
if output_format == "json":
new_project = status_response.new_project
return {
"name": new_project.name if new_project else project_name,
"path": new_project.path if new_project else project_path,
"is_default": bool(
(new_project.is_default if new_project else False) or set_default
),
"created": True,
"already_exists": False,
}
result = f"✓ {status_response.message}\n\n"
if status_response.new_project:
src/basic_memory/mcp/tools/read_note.py
+147 -35
View File
@@ -3,12 +3,13 @@
from textwrap import dedent
from typing import Optional, Literal
import yaml
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.server import mcp
from basic_memory.mcp.formatting import format_note_preview_ascii
from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.memory import memory_url_path
from basic_memory.utils import validate_project_path
@@ -19,6 +20,41 @@ def _is_exact_title_match(identifier: str, title: str) -> bool:
return identifier.strip().casefold() == title.strip().casefold()
def _parse_opening_frontmatter(content: str) -> tuple[str, dict | None]:
"""Parse opening YAML frontmatter and return (body, frontmatter).
Mirrors CLI behavior: only parses a frontmatter block at the very top.
If parsing fails or frontmatter is not a mapping, returns body unchanged and None.
"""
original_content = content
if not content.startswith("---\n"):
return original_content, None
lines = content.splitlines(keepends=True)
closing_index = None
for i in range(1, len(lines)):
if lines[i].strip() == "---":
closing_index = i
break
if closing_index is None:
return original_content, None
fm_text = "".join(lines[1:closing_index])
try:
parsed = yaml.safe_load(fm_text)
except yaml.YAMLError:
return original_content, None
if parsed is None:
parsed = {}
if not isinstance(parsed, dict):
return original_content, None
body_content = "".join(lines[closing_index + 1 :])
return body_content, parsed
@mcp.tool(
description="Read a markdown note by title or permalink.",
# TODO: re-enable once MCP client rendering is working
@@ -30,9 +66,10 @@ async def read_note(
workspace: Optional[str] = None,
page: int = 1,
page_size: int = 10,
output_format: Literal["default", "ascii", "ansi"] = "default",
output_format: Literal["text", "json"] = "text",
include_frontmatter: bool = False,
context: Context | None = None,
) -> str:
) -> str | dict:
"""Return the raw markdown for a note, or guidance text if no match is found.
Finds and retrieves a note by its title, permalink, or content search,
@@ -56,8 +93,10 @@ async def read_note(
Can be a full memory:// URL, a permalink, a title, or search text
page: Page number for paginated results (default: 1)
page_size: Number of items per page (default: 10)
output_format: "default" returns markdown, "ascii" returns a plain text preview,
"ansi" returns a colorized preview for TUI clients.
output_format: "text" returns markdown content or guidance text.
"json" returns a structured object with title/permalink/file_path/content/frontmatter.
include_frontmatter: When output_format="json", whether content should include the
opening YAML frontmatter block.
context: Optional FastMCP context for performance caching.
Returns:
@@ -108,6 +147,15 @@ async def read_note(
processed_path=processed_path,
project=active_project.name,
)
if output_format == "json":
return {
"title": None,
"permalink": None,
"file_path": None,
"content": None,
"frontmatter": None,
"error": "SECURITY_VALIDATION_ERROR",
}
return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries"
# Get the file via REST API - first try direct identifier resolution
@@ -122,6 +170,45 @@ async def read_note(
knowledge_client = KnowledgeClient(client, active_project.external_id)
resource_client = ResourceClient(client, active_project.external_id)
async def _read_json_payload(entity_id: str) -> dict:
entity = await knowledge_client.get_entity(entity_id)
response = await resource_client.read(entity_id, page=page, page_size=page_size)
content_text = response.text
body_content, parsed_frontmatter = _parse_opening_frontmatter(content_text)
return {
"title": entity.title,
"permalink": entity.permalink,
"file_path": entity.file_path,
"content": content_text if include_frontmatter else body_content,
"frontmatter": parsed_frontmatter,
}
def _empty_json_payload() -> dict:
return {
"title": None,
"permalink": None,
"file_path": None,
"content": None,
"frontmatter": None,
}
def _search_results(payload: object) -> list[dict]:
if not isinstance(payload, dict):
return []
results = payload.get("results")
return results if isinstance(results, list) else []
def _result_title(item: dict) -> str:
return str(item.get("title") or "")
def _result_permalink(item: dict) -> Optional[str]:
value = item.get("permalink")
return str(value) if value else None
def _result_file_path(item: dict) -> Optional[str]:
value = item.get("file_path")
return str(value) if value else None
try:
# Try to resolve identifier to entity ID
entity_id = await knowledge_client.resolve_entity(entity_path, strict=True)
@@ -132,12 +219,8 @@ async def read_note(
# If successful, return the content
if response.status_code == 200:
logger.info("Returning read_note result from resource: {path}", path=entity_path)
if output_format in ("ascii", "ansi"):
return format_note_preview_ascii(
response.text,
identifier=identifier,
color=output_format == "ansi",
)
if output_format == "json":
return await _read_json_payload(entity_id)
return response.text
except Exception as e: # pragma: no cover
logger.info(f"Direct lookup failed for '{entity_path}': {e}")
@@ -150,44 +233,45 @@ async def read_note(
search_type="title",
project=active_project.name,
workspace=workspace,
output_format="json",
context=context,
)
# Handle both SearchResponse object and error strings
if title_results and hasattr(title_results, "results") and title_results.results:
title_candidates = _search_results(title_results)
if title_candidates:
# Trigger: direct resolution failed and title search returned candidates.
# Why: avoid returning unrelated notes when search yields only fuzzy matches.
# Outcome: fetch content only when a true exact title match exists.
result = next(
(
candidate
for candidate in title_results.results
if _is_exact_title_match(identifier, candidate.title)
for candidate in title_candidates
if _is_exact_title_match(identifier, _result_title(candidate))
),
None,
)
if not result:
logger.info(f"No exact title match found for: {identifier}")
elif result.permalink:
elif _result_permalink(result):
try:
# Resolve the permalink to entity ID
entity_id = await knowledge_client.resolve_entity(result.permalink, strict=True)
entity_id = await knowledge_client.resolve_entity(
_result_permalink(result) or "", strict=True
)
# Fetch content using the entity ID
response = await resource_client.read(entity_id, page=page, page_size=page_size)
if response.status_code == 200:
logger.info(f"Found note by exact title search: {result.permalink}")
if output_format in ("ascii", "ansi"):
return format_note_preview_ascii(
response.text,
identifier=identifier,
color=output_format == "ansi",
)
logger.info(
f"Found note by exact title search: {_result_permalink(result)}"
)
if output_format == "json":
return await _read_json_payload(entity_id)
return response.text
except Exception as e: # pragma: no cover
logger.info(
f"Failed to fetch content for found title match {result.permalink}: {e}"
f"Failed to fetch content for found title match {_result_permalink(result)}: {e}"
)
else:
logger.info(
@@ -201,17 +285,28 @@ async def read_note(
search_type="text",
project=active_project.name,
workspace=workspace,
output_format="json",
context=context,
)
# We didn't find a direct match, construct a helpful error message
# Handle both SearchResponse object and error strings
if not text_results or not hasattr(text_results, "results") or not text_results.results:
# No results at all
text_candidates = _search_results(text_results)
if not text_candidates:
if output_format == "json":
return _empty_json_payload()
return format_not_found_message(active_project.name, identifier)
else:
# We found some related results
return format_related_results(active_project.name, identifier, text_results.results[:5])
if output_format == "json":
payload = _empty_json_payload()
payload["related_results"] = [
{
"title": _result_title(result),
"permalink": _result_permalink(result),
"file_path": _result_file_path(result),
}
for result in text_candidates[:5]
]
return payload
return format_related_results(active_project.name, identifier, text_candidates[:5])
def format_not_found_message(project: str | None, identifier: str) -> str:
@@ -271,14 +366,31 @@ def format_related_results(project: str | None, identifier: str, results) -> str
""")
for i, result in enumerate(results):
title = result.get("title") if isinstance(result, dict) else getattr(result, "title", None)
permalink = (
result.get("permalink")
if isinstance(result, dict)
else getattr(result, "permalink", None)
)
result_type = (
result.get("type") if isinstance(result, dict) else getattr(result, "type", None)
)
normalized_type = (
result_type
if isinstance(result_type, str)
else str(getattr(result_type, "value", result_type))
if result_type is not None
else None
)
message += dedent(f"""
## {i + 1}. {result.title}
- **Type**: {result.type.value}
- **Permalink**: {result.permalink}
## {i + 1}. {title or "Untitled"}
- **Type**: {normalized_type or "entity"}
- **Permalink**: {permalink or "unknown"}
You can read this note with:
```
read_note(project="{project}", {result.permalink}")
read_note(project="{project}", identifier="{permalink or ""}")
```
""")
src/basic_memory/mcp/tools/recent_activity.py
+36 -2
View File
@@ -1,7 +1,7 @@
"""Recent activity tool for Basic Memory MCP server."""
from datetime import timezone
from typing import List, Union, Optional
from typing import List, Union, Optional, Literal
from loguru import logger
from fastmcp import Context
@@ -41,8 +41,9 @@ async def recent_activity(
timeframe: TimeFrame = "7d",
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str:
) -> str | list[dict]:
"""Get recent activity for a specific project or across all projects.
Project Resolution:
@@ -78,6 +79,8 @@ async def recent_activity(
project: Project name to query. Optional - server will resolve using the
hierarchy above. If unknown, use list_memory_projects() to discover
available projects.
output_format: "text" returns human-readable summary text. "json" returns
a flat list of recent items.
context: Optional FastMCP context for performance caching.
Returns:
@@ -186,6 +189,12 @@ async def recent_activity(
most_active_count = item_count
most_active_project = project_info.name
if output_format == "json":
rows: list[dict] = []
for project_name, project_activity in projects_activity.items():
rows.extend(_extract_recent_rows(project_activity.activity, project_name))
return rows
# Build summary stats
summary = ActivityStats(
total_projects=len(project_list.projects),
@@ -258,6 +267,9 @@ async def recent_activity(
)
activity_data = GraphContext.model_validate(response.json())
if output_format == "json":
return _extract_recent_rows(activity_data)
# Format project-specific mode output
return _format_project_output(resolved_project, activity_data, timeframe, type)
@@ -315,6 +327,28 @@ async def _get_project_activity(
)
def _extract_recent_rows(
activity_data: GraphContext, project_name: Optional[str] = None
) -> list[dict]:
"""Flatten GraphContext into a list of recent rows."""
rows: list[dict] = []
for result in activity_data.results:
primary = result.primary_result
row = {
"type": primary.type,
"title": primary.title,
"permalink": primary.permalink,
"file_path": primary.file_path,
"created_at": (
primary.created_at.isoformat() if getattr(primary, "created_at", None) else None
),
}
if project_name is not None:
row["project"] = project_name
rows.append(row)
return rows
def _format_discovery_output(
projects_activity: dict, summary: ActivityStats, timeframe: str, guidance: str
) -> str:
src/basic_memory/mcp/tools/search.py
+6 -11
View File
@@ -9,7 +9,6 @@ from fastmcp import Context
from basic_memory.config import ConfigManager
from basic_memory.mcp.container import get_container
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.formatting import format_search_results_ascii
from basic_memory.mcp.server import mcp
from basic_memory.schemas.search import (
SearchItemType,
@@ -254,7 +253,7 @@ async def search_notes(
page: int = 1,
page_size: int = 10,
search_type: str = "text",
output_format: Literal["default", "ascii", "ansi"] = "default",
output_format: Literal["text", "json"] = "text",
types: List[str] | None = None,
entity_types: List[str] | None = None,
after_date: Optional[str] = None,
@@ -263,7 +262,7 @@ async def search_notes(
status: Optional[str] = None,
min_similarity: Optional[float] = None,
context: Context | None = None,
) -> SearchResponse | str:
) -> SearchResponse | dict | str:
"""Search across all content in the knowledge base with comprehensive syntax support.
This tool searches the knowledge base using full-text search, pattern matching,
@@ -342,8 +341,8 @@ async def search_notes(
search_type: Type of search to perform, one of:
"text", "title", "permalink", "vector", "semantic", "hybrid" (default: "text";
text mode auto-upgrades to hybrid when semantic search is enabled)
output_format: "default" returns structured data, "ascii" returns a plain text table,
"ansi" returns a colorized table for TUI clients.
output_format: "text" preserves existing structured search response behavior.
"json" returns a machine-readable dictionary payload.
types: Optional list of note types to search (e.g., ["note", "person"])
entity_types: Optional list of entity types to filter by (e.g., ["entity", "observation"])
after_date: Optional date filter for recent content (e.g., "1 week", "2d", "2024-01-01")
@@ -497,12 +496,8 @@ async def search_notes(
# Don't treat this as an error, but the user might want guidance
# We return the empty result as normal - the user can decide if they need help
if output_format in ("ascii", "ansi"):
return format_search_results_ascii(
result,
query=query,
color=output_format == "ansi",
)
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
src/basic_memory/mcp/tools/ui_sdk.py
+3 -3
View File
@@ -42,7 +42,7 @@ async def search_notes_ui(
page=page,
page_size=page_size,
search_type=search_type,
output_format="default",
output_format="json",
types=types,
entity_types=entity_types,
after_date=after_date,
@@ -62,7 +62,7 @@ async def search_notes_ui(
"page": page,
"page_size": page_size,
},
"toolOutput": result.model_dump(),
"toolOutput": result,
}
try:
@@ -96,7 +96,7 @@ async def read_note_ui(
project=project,
page=page,
page_size=page_size,
output_format="default",
output_format="text",
context=context,
)
src/basic_memory/mcp/tools/write_note.py
+23 -2
View File
@@ -1,6 +1,6 @@
"""Write note tool for Basic Memory MCP server."""
from typing import List, Union, Optional
from typing import List, Union, Optional, Literal
from loguru import logger
@@ -26,8 +26,9 @@ async def write_note(
tags: list[str] | str | None = None,
note_type: str = "note",
metadata: dict | None = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str:
) -> str | dict:
"""Write a markdown note to the knowledge base.
Creates or updates a markdown note with semantic observations and relations.
@@ -72,6 +73,8 @@ async def write_note(
metadata: Optional dict of extra frontmatter fields merged into entity_metadata.
Useful for schema notes or any note that needs custom YAML frontmatter
beyond title/type/tags. Nested dicts are supported.
output_format: "text" returns the existing markdown summary. "json" returns
machine-readable metadata.
context: Optional FastMCP context for performance caching.
Returns:
@@ -145,6 +148,15 @@ async def write_note(
directory=directory,
project=active_project.name,
)
if output_format == "json":
return {
"title": title,
"permalink": None,
"file_path": None,
"checksum": None,
"action": "created",
"error": "SECURITY_VALIDATION_ERROR",
}
return f"# Error\n\nDirectory path '{directory}' is not allowed - paths must stay within project boundaries"
# Process tags using the helper function
@@ -246,5 +258,14 @@ async def write_note(
logger.info(
f"MCP tool response: tool=write_note project={active_project.name} action={action} permalink={result.permalink} observations_count={len(result.observations)} relations_count={len(result.relations)} resolved_relations={resolved} unresolved_relations={unresolved}"
)
if output_format == "json":
return {
"title": result.title,
"permalink": result.permalink,
"file_path": result.file_path,
"checksum": result.checksum,
"action": action.lower(),
}
summary_result = "\n".join(summary)
return add_project_metadata(summary_result, active_project.name)
test-int/mcp/test_output_format_ascii_integration.py
-78
View File
@@ -1,78 +0,0 @@
"""
Integration tests for ASCII/ANSI output formats in MCP tools.
"""
import pytest
from fastmcp import Client
@pytest.mark.asyncio
async def test_search_notes_ascii_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "ASCII Note",
"directory": "notes",
"content": "# ASCII Note\n\nThis is a note for ASCII output.",
"tags": "ascii,output",
},
)
search_result = await client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "ASCII",
"output_format": "ascii",
},
)
assert len(search_result.content) == 1
assert search_result.content[0].type == "text"
text = search_result.content[0].text
assert "Search results" in text
assert "ASCII Note" in text
assert "+" in text
@pytest.mark.asyncio
async def test_read_note_ascii_and_ansi_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Color Note",
"directory": "notes",
"content": "# Color Note\n\nThis note is for ANSI output.",
"tags": "ansi,output",
},
)
ascii_result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "Color Note",
"output_format": "ascii",
},
)
assert len(ascii_result.content) == 1
ascii_text = ascii_result.content[0].text
assert "Note preview" in ascii_text
assert "# Color Note" in ascii_text
ansi_result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "Color Note",
"output_format": "ansi",
},
)
ansi_text = ansi_result.content[0].text
assert "\x1b[" in ansi_text
test-int/mcp/test_output_format_json_integration.py
+398
View File
@@ -0,0 +1,398 @@
"""Integration tests for MCP `output_format="json"` responses."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastmcp import Client
from basic_memory.mcp.clients.knowledge import KnowledgeClient
def _json_content(tool_result) -> dict | list:
"""Parse a FastMCP tool result content block into JSON."""
assert len(tool_result.content) == 1
assert tool_result.content[0].type == "text"
return json.loads(tool_result.content[0].text) # pyright: ignore [reportAttributeAccessIssue]
@pytest.mark.asyncio
async def test_write_note_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
result = await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Write",
"directory": "json-int",
"content": "# JSON Integration Write\n\nBody",
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["title"] == "JSON Integration Write"
assert payload["action"] in ("created", "updated")
assert payload["permalink"]
assert payload["file_path"]
assert "checksum" in payload
@pytest.mark.asyncio
async def test_read_note_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Read",
"directory": "json-int",
"content": "# JSON Integration Read\n\nBody",
},
)
result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "json-int/json-integration-read",
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["title"] == "JSON Integration Read"
assert payload["permalink"]
assert payload["file_path"]
assert isinstance(payload["content"], str)
assert "frontmatter" in payload
@pytest.mark.asyncio
async def test_edit_note_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Edit",
"directory": "json-int",
"content": "# JSON Integration Edit\n\nBody",
},
)
result = await client.call_tool(
"edit_note",
{
"project": test_project.name,
"identifier": "json-int/json-integration-edit",
"operation": "append",
"content": "\n\nAppended",
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["title"] == "JSON Integration Edit"
assert payload["operation"] == "append"
assert payload["permalink"]
assert payload["file_path"]
assert "checksum" in payload
@pytest.mark.asyncio
async def test_recent_activity_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Recent",
"directory": "json-int",
"content": "# JSON Integration Recent\n\nBody",
},
)
result = await client.call_tool(
"recent_activity",
{
"project": test_project.name,
"timeframe": "7d",
"output_format": "json",
},
)
payload = _json_content(result)
assert isinstance(payload, list)
assert any(item.get("title") == "JSON Integration Recent" for item in payload)
for item in payload:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(
item.keys()
)
@pytest.mark.asyncio
async def test_recent_activity_json_output_for_relation_and_observation_types(
mcp_server, app, test_project
):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Type Source",
"directory": "json-int",
"content": (
"# JSON Integration Type Source\n\n"
"- [note] observation from source\n"
"- links_to [[JSON Integration Type Target]]"
),
},
)
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Type Target",
"directory": "json-int",
"content": "# JSON Integration Type Target\n\nBody",
},
)
relation_result = await client.call_tool(
"recent_activity",
{
"project": test_project.name,
"timeframe": "7d",
"type": "relation",
"output_format": "json",
},
)
relation_payload = _json_content(relation_result)
assert isinstance(relation_payload, list)
assert relation_payload
assert all(item.get("type") == "relation" for item in relation_payload)
for item in relation_payload:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(
item.keys()
)
observation_result = await client.call_tool(
"recent_activity",
{
"project": test_project.name,
"timeframe": "7d",
"type": "observation",
"output_format": "json",
},
)
observation_payload = _json_content(observation_result)
assert isinstance(observation_payload, list)
assert observation_payload
assert all(item.get("type") == "observation" for item in observation_payload)
for item in observation_payload:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(
item.keys()
)
@pytest.mark.asyncio
async def test_list_memory_projects_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
result = await client.call_tool(
"list_memory_projects",
{"output_format": "json"},
)
payload = _json_content(result)
assert isinstance(payload, dict)
assert "projects" in payload
assert any(project["name"] == test_project.name for project in payload["projects"])
assert "default_project" in payload
assert "constrained_project" in payload
@pytest.mark.asyncio
async def test_create_memory_project_json_output_is_idempotent(
mcp_server, app, test_project, tmp_path
):
async with Client(mcp_server) as client:
project_name = "json-int-created"
project_path = str(tmp_path.parent / (tmp_path.name + "-projects") / "json-int-created")
first = await client.call_tool(
"create_memory_project",
{
"project_name": project_name,
"project_path": project_path,
"output_format": "json",
},
)
first_payload = _json_content(first)
assert first_payload["name"] == project_name
# Normalize path separators for cross-platform compatibility.
assert Path(first_payload["path"]) == Path(project_path)
assert first_payload["created"] is True
assert first_payload["already_exists"] is False
second = await client.call_tool(
"create_memory_project",
{
"project_name": project_name,
"project_path": str(
tmp_path.parent / (tmp_path.name + "-projects") / "json-int-created-second"
),
"output_format": "json",
},
)
second_payload = _json_content(second)
assert second_payload["name"] == project_name
assert second_payload["created"] is False
assert second_payload["already_exists"] is True
@pytest.mark.asyncio
async def test_delete_note_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Delete",
"directory": "json-int",
"content": "# JSON Integration Delete\n\nBody",
},
)
result = await client.call_tool(
"delete_note",
{
"project": test_project.name,
"identifier": "json-int/json-integration-delete",
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["deleted"] is True
assert payload["title"] == "JSON Integration Delete"
assert payload["permalink"]
assert payload["file_path"]
@pytest.mark.asyncio
async def test_delete_note_directory_json_output_failure_is_structured(
mcp_server, app, test_project, monkeypatch
):
async def mock_delete_directory(self, directory: str):
raise RuntimeError("simulated directory delete failure")
monkeypatch.setattr(KnowledgeClient, "delete_directory", mock_delete_directory)
async with Client(mcp_server) as client:
result = await client.call_tool(
"delete_note",
{
"project": test_project.name,
"identifier": "json-int",
"is_directory": True,
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["deleted"] is False
assert payload["is_directory"] is True
assert payload["identifier"] == "json-int"
assert "simulated directory delete failure" in payload["error"]
@pytest.mark.asyncio
async def test_move_note_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Move",
"directory": "json-int",
"content": "# JSON Integration Move\n\nBody",
},
)
result = await client.call_tool(
"move_note",
{
"project": test_project.name,
"identifier": "json-int/json-integration-move",
"destination_path": "json-int/moved/json-integration-move.md",
"output_format": "json",
},
)
payload = _json_content(result)
assert payload["moved"] is True
assert payload["title"] == "JSON Integration Move"
assert payload["source"] == "json-int/json-integration-move"
assert payload["destination"] == "json-int/moved/json-integration-move.md"
assert payload["permalink"]
assert payload["file_path"]
@pytest.mark.asyncio
async def test_build_context_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Context",
"directory": "json-int",
"content": "# JSON Integration Context\n\nBody",
},
)
result = await client.call_tool(
"build_context",
{
"project": test_project.name,
"url": "memory://json-int/json-integration-context",
"output_format": "json",
},
)
payload = _json_content(result)
assert isinstance(payload, dict)
assert "results" in payload
assert "metadata" in payload
@pytest.mark.asyncio
async def test_search_notes_json_output(mcp_server, app, test_project):
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "JSON Integration Search",
"directory": "json-int",
"content": "# JSON Integration Search\n\nBody",
},
)
result = await client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "JSON Integration Search",
"output_format": "json",
},
)
payload = _json_content(result)
assert isinstance(payload, dict)
assert "results" in payload
assert isinstance(payload["results"], list)
assert any(item.get("title") == "JSON Integration Search" for item in payload["results"])
test-int/mcp/test_project_management_integration.py
+28 -18
View File
@@ -121,7 +121,7 @@ async def test_create_project_with_default_flag(mcp_server, app, test_project, t
@pytest.mark.asyncio
async def test_create_project_duplicate_name(mcp_server, app, test_project, tmp_path):
"""Test creating a project with duplicate name shows error."""
"""Test creating a project with duplicate name is idempotent."""
async with Client(mcp_server) as client:
# First create a project
@@ -135,25 +135,35 @@ async def test_create_project_duplicate_name(mcp_server, app, test_project, tmp_
},
)
# Try to create another project with same name
with pytest.raises(Exception) as exc_info:
await client.call_tool(
"create_memory_project",
{
"project_name": "duplicate-test",
"project_path": str(
tmp_path.parent / (tmp_path.name + "-projects") / "project-duplicate-test-2"
),
},
)
# Second create with same name should succeed idempotently
second_result = await client.call_tool(
"create_memory_project",
{
"project_name": "duplicate-test",
"project_path": str(
tmp_path.parent / (tmp_path.name + "-projects") / "project-duplicate-test-2"
),
},
)
second_text = second_result.content[0].text # pyright: ignore [reportAttributeAccessIssue]
assert "already exists" in second_text.lower()
assert "duplicate-test" in second_text
# Should show error about duplicate name
error_message = str(exc_info.value)
assert "create_memory_project" in error_message
# JSON mode should explicitly report already_exists=true
second_json = await client.call_tool(
"create_memory_project",
{
"project_name": "duplicate-test",
"project_path": str(
tmp_path.parent / (tmp_path.name + "-projects") / "project-duplicate-test-3"
),
"output_format": "json",
},
)
second_json_text = second_json.content[0].text # pyright: ignore [reportAttributeAccessIssue]
assert (
"duplicate-test" in error_message
or "already exists" in error_message
or "Invalid request" in error_message
'"already_exists":true' in second_json_text
or '"already_exists": true' in second_json_text
)
tests/mcp/test_tool_build_context.py
+5 -5
View File
@@ -152,12 +152,12 @@ async def test_build_context_string_depth_parameter(client, test_graph, test_pro
@pytest.mark.asyncio
async def test_build_context_markdown_format(client, test_graph, test_project):
"""Test that format='markdown' returns compact text."""
async def test_build_context_text_format(client, test_graph, test_project):
"""Test that output_format='text' returns compact text."""
result = await build_context.fn(
project=test_project.name,
url="memory://test/root",
format="markdown",
output_format="text",
)
assert isinstance(result, str)
@@ -176,7 +176,7 @@ async def test_build_context_markdown_pattern(client, test_graph, test_project):
result = await build_context.fn(
project=test_project.name,
url="memory://test/*",
format="markdown",
output_format="text",
)
assert isinstance(result, str)
@@ -193,7 +193,7 @@ async def test_build_context_markdown_not_found(client, test_project):
result = await build_context.fn(
project=test_project.name,
url="memory://test/does-not-exist",
format="markdown",
output_format="text",
)
assert isinstance(result, str)
tests/mcp/test_tool_contracts.py
+24 -7
View File
@@ -17,12 +17,12 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"page",
"page_size",
"max_related",
"format",
"output_format",
],
"canvas": ["nodes", "edges", "title", "directory", "project", "workspace"],
"cloud_info": [],
"create_memory_project": ["project_name", "project_path", "set_default"],
"delete_note": ["identifier", "is_directory", "project", "workspace"],
"create_memory_project": ["project_name", "project_path", "set_default", "output_format"],
"delete_note": ["identifier", "is_directory", "project", "workspace", "output_format"],
"delete_project": ["project_name"],
"edit_note": [
"identifier",
@@ -33,16 +33,32 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"section",
"find_text",
"expected_replacements",
"output_format",
],
"fetch": ["id"],
"list_directory": ["dir_name", "depth", "file_name_glob", "project", "workspace"],
"list_memory_projects": [],
"list_memory_projects": ["output_format"],
"list_workspaces": [],
"move_note": ["identifier", "destination_path", "is_directory", "project", "workspace"],
"move_note": [
"identifier",
"destination_path",
"is_directory",
"project",
"workspace",
"output_format",
],
"read_content": ["path", "project", "workspace"],
"read_note": ["identifier", "project", "workspace", "page", "page_size", "output_format"],
"read_note": [
"identifier",
"project",
"workspace",
"page",
"page_size",
"output_format",
"include_frontmatter",
],
"release_notes": [],
"recent_activity": ["type", "depth", "timeframe", "project", "workspace"],
"recent_activity": ["type", "depth", "timeframe", "project", "workspace", "output_format"],
"schema_diff": ["note_type", "project", "workspace"],
"schema_infer": ["note_type", "threshold", "project", "workspace"],
"schema_validate": ["note_type", "identifier", "project", "workspace"],
@@ -74,6 +90,7 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"tags",
"note_type",
"metadata",
"output_format",
],
}
tests/mcp/test_tool_json_output_modes.py
+362
View File
@@ -0,0 +1,362 @@
"""Tests for text/json output mode behavior on MCP tools used by openclaw-basic-memory."""
from __future__ import annotations
from pathlib import Path
import pytest
from basic_memory.mcp.clients.knowledge import KnowledgeClient
from basic_memory.mcp.tools import (
build_context,
create_memory_project,
delete_note,
edit_note,
list_memory_projects,
move_note,
read_note,
recent_activity,
write_note,
)
@pytest.mark.asyncio
async def test_write_note_text_and_json_modes(app, test_project):
text_result = await write_note.fn(
project=test_project.name,
title="Mode Write Note",
directory="mode-tests",
content="# Mode Write Note\n\ninitial",
output_format="text",
)
assert isinstance(text_result, str)
assert "note" in text_result.lower()
json_result = await write_note.fn(
project=test_project.name,
title="Mode Write Note",
directory="mode-tests",
content="# Mode Write Note\n\nupdated",
output_format="json",
)
assert isinstance(json_result, dict)
assert json_result["title"] == "Mode Write Note"
assert json_result["action"] in ("created", "updated")
assert json_result["permalink"]
assert json_result["file_path"]
assert "checksum" in json_result
@pytest.mark.asyncio
async def test_read_note_text_and_json_modes(app, test_project):
await write_note.fn(
project=test_project.name,
title="Mode Read Note",
directory="mode-tests",
content="# Mode Read Note\n\nbody",
)
text_result = await read_note.fn(
identifier="mode-tests/mode-read-note",
project=test_project.name,
output_format="text",
)
assert isinstance(text_result, str)
assert "Mode Read Note" in text_result
json_result = await read_note.fn(
identifier="mode-tests/mode-read-note",
project=test_project.name,
output_format="json",
)
assert isinstance(json_result, dict)
assert json_result["title"] == "Mode Read Note"
assert json_result["permalink"]
assert json_result["file_path"]
assert isinstance(json_result["content"], str)
assert "frontmatter" in json_result
missing_json = await read_note.fn(
identifier="mode-tests/missing-note",
project=test_project.name,
output_format="json",
)
assert isinstance(missing_json, dict)
assert set(["title", "permalink", "file_path", "content", "frontmatter"]).issubset(
missing_json.keys()
)
@pytest.mark.asyncio
async def test_edit_note_text_and_json_modes(app, test_project):
await write_note.fn(
project=test_project.name,
title="Mode Edit Note",
directory="mode-tests",
content="# Mode Edit Note\n\nstart",
)
text_result = await edit_note.fn(
identifier="mode-tests/mode-edit-note",
operation="append",
content="\n\ntext-append",
project=test_project.name,
output_format="text",
)
assert isinstance(text_result, str)
assert "Edited note" in text_result
json_result = await edit_note.fn(
identifier="mode-tests/mode-edit-note",
operation="append",
content="\n\njson-append",
project=test_project.name,
output_format="json",
)
assert isinstance(json_result, dict)
assert json_result["title"] == "Mode Edit Note"
assert json_result["operation"] == "append"
assert json_result["permalink"]
assert json_result["file_path"]
assert "checksum" in json_result
@pytest.mark.asyncio
async def test_recent_activity_text_and_json_modes(app, test_project):
await write_note.fn(
project=test_project.name,
title="Mode Activity Note",
directory="mode-tests",
content="# Mode Activity Note\n\nactivity",
)
text_result = await recent_activity.fn(
project=test_project.name,
timeframe="7d",
output_format="text",
)
assert isinstance(text_result, str)
assert "Recent Activity" in text_result
json_result = await recent_activity.fn(
project=test_project.name,
timeframe="7d",
output_format="json",
)
assert isinstance(json_result, list)
assert any(item.get("title") == "Mode Activity Note" for item in json_result)
for item in json_result:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(item.keys())
@pytest.mark.asyncio
async def test_recent_activity_json_preserves_relation_and_observation_types(app, test_project):
await write_note.fn(
project=test_project.name,
title="Activity Type Source",
directory="mode-tests",
content=(
"# Activity Type Source\n\n"
"- [note] observation from source\n"
"- links_to [[Activity Type Target]]"
),
)
await write_note.fn(
project=test_project.name,
title="Activity Type Target",
directory="mode-tests",
content="# Activity Type Target",
)
relation_json = await recent_activity.fn(
project=test_project.name,
type="relation",
timeframe="7d",
output_format="json",
)
assert isinstance(relation_json, list)
assert relation_json
assert all(item.get("type") == "relation" for item in relation_json)
for item in relation_json:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(item.keys())
observation_json = await recent_activity.fn(
project=test_project.name,
type="observation",
timeframe="7d",
output_format="json",
)
assert isinstance(observation_json, list)
assert observation_json
assert all(item.get("type") == "observation" for item in observation_json)
for item in observation_json:
assert set(["type", "title", "permalink", "file_path", "created_at"]).issubset(item.keys())
@pytest.mark.asyncio
async def test_list_and_create_project_text_and_json_modes(app, test_project, tmp_path):
list_text = await list_memory_projects.fn(output_format="text")
assert isinstance(list_text, str)
assert test_project.name in list_text
list_json = await list_memory_projects.fn(output_format="json")
assert isinstance(list_json, dict)
assert "projects" in list_json
assert any(project["name"] == test_project.name for project in list_json["projects"])
project_name = "mode-create-project"
project_path = str(tmp_path.parent / (tmp_path.name + "-projects") / "mode-create-project")
create_text = await create_memory_project.fn(
project_name=project_name,
project_path=project_path,
output_format="text",
)
assert isinstance(create_text, str)
assert "mode-create-project" in create_text
create_json_again = await create_memory_project.fn(
project_name=project_name,
project_path=project_path,
output_format="json",
)
assert isinstance(create_json_again, dict)
assert create_json_again["name"] == project_name
# Normalize path separators for cross-platform compatibility.
assert Path(create_json_again["path"]) == Path(project_path)
assert create_json_again["created"] is False
assert create_json_again["already_exists"] is True
default_project_name = "mode-default-project"
default_project_path = str(
tmp_path.parent / (tmp_path.name + "-projects") / "mode-default-project"
)
await create_memory_project.fn(
project_name=default_project_name,
project_path=default_project_path,
set_default=True,
output_format="text",
)
default_text_again = await create_memory_project.fn(
project_name=default_project_name,
project_path=default_project_path,
output_format="text",
)
assert "Set as default project\\n" not in default_text_again
assert "Set as default project\n" in default_text_again
@pytest.mark.asyncio
async def test_delete_note_text_and_json_modes(app, test_project):
await write_note.fn(
project=test_project.name,
title="Mode Delete Text",
directory="mode-tests",
content="# Mode Delete Text",
)
text_delete = await delete_note.fn(
identifier="mode-tests/mode-delete-text",
project=test_project.name,
output_format="text",
)
assert text_delete is True
await write_note.fn(
project=test_project.name,
title="Mode Delete Json",
directory="mode-tests",
content="# Mode Delete Json",
)
json_delete = await delete_note.fn(
identifier="mode-tests/mode-delete-json",
project=test_project.name,
output_format="json",
)
assert isinstance(json_delete, dict)
assert json_delete["deleted"] is True
assert json_delete["title"] == "Mode Delete Json"
assert json_delete["permalink"]
assert json_delete["file_path"]
@pytest.mark.asyncio
async def test_delete_directory_json_mode_returns_structured_error_on_failure(
app, test_project, monkeypatch
):
async def mock_delete_directory(self, directory: str):
raise RuntimeError("simulated directory delete failure")
monkeypatch.setattr(KnowledgeClient, "delete_directory", mock_delete_directory)
json_delete = await delete_note.fn(
identifier="mode-tests",
is_directory=True,
project=test_project.name,
output_format="json",
)
assert isinstance(json_delete, dict)
assert json_delete["deleted"] is False
assert json_delete["is_directory"] is True
assert json_delete["identifier"] == "mode-tests"
assert "simulated directory delete failure" in json_delete["error"]
@pytest.mark.asyncio
async def test_move_note_text_and_json_modes(app, test_project):
await write_note.fn(
project=test_project.name,
title="Mode Move Text",
directory="mode-tests",
content="# Mode Move Text",
)
text_move = await move_note.fn(
identifier="mode-tests/mode-move-text",
destination_path="mode-tests/moved/mode-move-text.md",
project=test_project.name,
output_format="text",
)
assert isinstance(text_move, str)
assert "moved" in text_move.lower()
await write_note.fn(
project=test_project.name,
title="Mode Move Json",
directory="mode-tests",
content="# Mode Move Json",
)
json_move = await move_note.fn(
identifier="mode-tests/mode-move-json",
destination_path="mode-tests/moved/mode-move-json.md",
project=test_project.name,
output_format="json",
)
assert isinstance(json_move, dict)
assert json_move["moved"] is True
assert json_move["title"] == "Mode Move Json"
assert json_move["source"] == "mode-tests/mode-move-json"
assert json_move["destination"] == "mode-tests/moved/mode-move-json.md"
assert json_move["permalink"]
assert json_move["file_path"]
@pytest.mark.asyncio
async def test_build_context_json_default_and_text_mode(client, test_graph, test_project):
json_result = await build_context.fn(
project=test_project.name,
url="memory://test/root",
)
assert isinstance(json_result, dict)
assert "results" in json_result
text_result = await build_context.fn(
project=test_project.name,
url="memory://test/root",
output_format="text",
)
assert isinstance(text_result, str)
assert "# Context:" in text_result
tests/mcp/test_tool_read_note.py
+36 -37
View File
@@ -5,7 +5,6 @@ from textwrap import dedent
import pytest
from basic_memory.mcp.tools import write_note, read_note
from basic_memory.schemas.search import SearchResponse, SearchResult, SearchItemType
from basic_memory.utils import normalize_newlines
@@ -68,30 +67,30 @@ async def test_read_note_returns_related_results_when_text_search_finds_matches(
async def fake_search_notes_fn(*, query, search_type, **kwargs):
if search_type == "title":
return SearchResponse(results=[], current_page=1, page_size=10)
return {"results": [], "current_page": 1, "page_size": 10}
return SearchResponse(
results=[
SearchResult(
title="Related One",
permalink="docs/related-one",
content="",
type=SearchItemType.ENTITY,
score=1.0,
file_path="docs/related-one.md",
),
SearchResult(
title="Related Two",
permalink="docs/related-two",
content="",
type=SearchItemType.ENTITY,
score=0.9,
file_path="docs/related-two.md",
),
return {
"results": [
{
"title": "Related One",
"permalink": "docs/related-one",
"content": "",
"type": "entity",
"score": 1.0,
"file_path": "docs/related-one.md",
},
{
"title": "Related Two",
"permalink": "docs/related-two",
"content": "",
"type": "entity",
"score": 0.9,
"file_path": "docs/related-two.md",
},
],
current_page=1,
page_size=10,
)
"current_page": 1,
"page_size": 10,
}
# Ensure direct resolution doesn't short-circuit the fallback logic.
class FailingKnowledgeClient(OriginalKnowledgeClient):
@@ -131,21 +130,21 @@ async def test_read_note_title_fallback_requires_exact_title_match(monkeypatch,
async def fake_search_notes_fn(*, query, search_type, **kwargs):
if search_type == "title":
return SearchResponse(
results=[
SearchResult(
title="Existing Note",
permalink="test/existing-note",
content="",
type=SearchItemType.ENTITY,
score=1.0,
file_path="test/Existing Note.md",
)
return {
"results": [
{
"title": "Existing Note",
"permalink": "test/existing-note",
"content": "",
"type": "entity",
"score": 1.0,
"file_path": "test/Existing Note.md",
}
],
current_page=1,
page_size=10,
)
return SearchResponse(results=[], current_page=1, page_size=10)
"current_page": 1,
"page_size": 10,
}
return {"results": [], "current_page": 1, "page_size": 10}
monkeypatch.setattr(clients_mod, "KnowledgeClient", StrictFailingKnowledgeClient)
monkeypatch.setattr(read_note_module.search_notes, "fn", fake_search_notes_fn)