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

Add MCP output_format json mode across memory tools

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
phernandez
2026-02-18 19:30:49 -06:00
parent 0a36256f8a
commit 46b372c3e1
21 changed files with 1202 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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
+59 -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 = [
@@ -288,18 +300,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 +347,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
+23 -2
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:
@@ -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
+88 -2
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
@@ -537,6 +584,16 @@ move_note("path/to/file.md", "{destination_path}/file.md")
# 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
@@ -573,6 +630,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
@@ -600,6 +667,15 @@ move_note("path/to/file.md", "{destination_path}/file.md")
# Call the move API using KnowledgeClient
result = await knowledge_client.move_entity(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 +700,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
+95 -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,71 @@ 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,
"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 +89,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 +104,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 +132,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
+158 -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,56 @@ 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:
if isinstance(payload, dict):
results = payload.get("results")
return results if isinstance(results, list) else []
if hasattr(payload, "results"):
results = getattr(payload, "results")
return results if isinstance(results, list) else []
return []
def _result_title(item: object) -> str:
if isinstance(item, dict):
return str(item.get("title") or "")
return str(getattr(item, "title", "") or "")
def _result_permalink(item: object) -> Optional[str]:
if isinstance(item, dict):
value = item.get("permalink")
return str(value) if value else None
value = getattr(item, "permalink", None)
return str(value) if value else None
def _result_file_path(item: object) -> Optional[str]:
if isinstance(item, dict):
value = item.get("file_path")
return str(value) if value else None
value = getattr(item, "file_path", None)
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 +230,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 +244,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 +296,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 +377,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
+37 -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 entity 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_entity_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_entity_rows(activity_data)
# Format project-specific mode output
return _format_project_output(resolved_project, activity_data, timeframe, type)
@@ -315,6 +327,29 @@ async def _get_project_activity(
)
def _extract_recent_entity_rows(
activity_data: GraphContext, project_name: Optional[str] = None
) -> list[dict]:
"""Flatten GraphContext into a list of recent entity rows."""
rows: list[dict] = []
for result in activity_data.results:
primary = result.primary_result
if primary.type != "entity":
continue
row = {
"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)