Add MCP output_format json mode across memory tools

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-21 13:47:35 +00:00 · 2026-02-18 19:30:49 -06:00
21 changed files with 1202 additions and 242 deletions
@@ -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
@@ -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: __
@@ -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)
@@ -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)
@@ -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)
@@ -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")
@@ -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)
@@ -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
            )
@@ -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)
@@ -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:
@@ -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 ""}")
            ```

            """)
@@ -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:
@@ -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

@@ -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,
    )

@@ -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)
@@ -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
@@ -0,0 +1,300 @@
+"""Integration tests for MCP `output_format="json"` responses."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from fastmcp import Client
+
+
+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(["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
+        assert first_payload["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_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"])
@@ -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
        )


@@ -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)
@@ -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",
    ],
 }

@@ -0,0 +1,273 @@
+"""Tests for text/json output mode behavior on MCP tools used by openclaw-basic-memory."""
+
+from __future__ import annotations
+
+import pytest
+
+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(["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
+    assert create_json_again["path"] == project_path
+    assert create_json_again["created"] is False
+    assert create_json_again["already_exists"] is True
+
+
+@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_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