feat: upgrade fastmcp 2.12.3 to 3.0.1 with tool annotations (#598)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

src/basic_memory/cli/commands/tool.py

@@ -104,7 +104,7 @@ async def _write_note_json(
 ) -> dict:
     """Write a note and return structured JSON metadata."""
     # Use the MCP tool to create/update the entity (handles create-or-update logic)
-    await mcp_write_note.fn(
+    await mcp_write_note(
         title=title,
         content=content,
         directory=folder,
@@ -155,7 +155,7 @@ async def _read_note_json(
         if entity_id is None:
             from basic_memory.mcp.tools.search import search_notes as mcp_search_tool
 
-            title_results = await mcp_search_tool.fn(
+            title_results = await mcp_search_tool(
                 query=identifier,
                 search_type="title",
                 project=project_name,
@@ -387,7 +387,7 @@ def write_note(
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
                 note = run_with_cleanup(
-                    mcp_write_note.fn(
+                    mcp_write_note(
                         title=title,
                         content=content,
                         directory=folder,
@@ -468,13 +468,15 @@ def read_note(
                     result["content"] = stripped_content
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
-                note = run_with_cleanup(
-                    mcp_read_note.fn(
-                        identifier=identifier,
-                        project=project_name,
-                        workspace=workspace,
-                        page=page,
-                        page_size=page_size,
+                note = str(
+                    run_with_cleanup(
+                        mcp_read_note(
+                            identifier=identifier,
+                            project=project_name,
+                            workspace=workspace,
+                            page=page,
+                            page_size=page_size,
+                        )
                     )
                 )
                 if strip_frontmatter:
@@ -560,16 +562,18 @@ def edit_note(
                 )
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
-                result = run_with_cleanup(
-                    mcp_edit_note.fn(
-                        identifier=identifier,
-                        operation=operation,
-                        content=content,
-                        project=project_name,
-                        workspace=workspace,
-                        section=section,
-                        find_text=find_text,
-                        expected_replacements=expected_replacements,
+                result = str(
+                    run_with_cleanup(
+                        mcp_edit_note(
+                            identifier=identifier,
+                            operation=operation,
+                            content=content,
+                            project=project_name,
+                            workspace=workspace,
+                            section=section,
+                            find_text=find_text,
+                            expected_replacements=expected_replacements,
+                        )
                     )
                 )
                 rprint(result)
@@ -629,7 +633,7 @@ def build_context(
 
         with force_routing(local=local, cloud=cloud):
             result = run_with_cleanup(
-                mcp_build_context.fn(
+                mcp_build_context(
                     project=project_name,
                     workspace=workspace,
                     url=url,
@@ -712,10 +716,10 @@ def recent_activity(
                 print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
             else:
                 result = run_with_cleanup(
-                    mcp_recent_activity.fn(
-                        type=type,  # pyright: ignore [reportArgumentType]
-                        depth=depth,
-                        timeframe=timeframe,
+                    mcp_recent_activity(
+                        type=type,  # pyright: ignore[reportArgumentType]
+                        depth=depth if depth is not None else 1,
+                        timeframe=timeframe if timeframe is not None else "7d",
                         project=project_name,
                         workspace=workspace,
                     )
@@ -862,11 +866,12 @@ def search_notes(
 
         with force_routing(local=local, cloud=cloud):
             results = run_with_cleanup(
-                mcp_search.fn(
+                mcp_search(
                     query=query or "",
                     project=project_name,
                     workspace=workspace,
                     search_type=search_type,
+                    output_format="json",
                     page=page,
                     after_date=after_date,
                     page_size=page_size,
@@ -881,8 +886,7 @@ def search_notes(
             print(results)
             raise typer.Exit(1)
 
-        results_dict = results.model_dump(exclude_none=True)
-        print(json.dumps(results_dict, indent=2, ensure_ascii=True, default=str))
+        print(json.dumps(results, indent=2, ensure_ascii=True, default=str))
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -916,7 +920,7 @@ def continue_conversation(
         with force_routing(local=local, cloud=cloud):
             # Prompt functions return formatted strings directly
             session = run_with_cleanup(
-                mcp_continue_conversation.fn(topic=topic, timeframe=timeframe)  # type: ignore[arg-type]
+                mcp_continue_conversation(topic=topic, timeframe=timeframe)  # type: ignore[arg-type]
             )
         rprint(session)
     except ValueError as e:

src/basic_memory/mcp/project_context.py

@@ -99,11 +99,9 @@ def _workspace_choices(workspaces: list[WorkspaceInfo]) -> str:
 async def get_available_workspaces(context: Optional[Context] = None) -> list[WorkspaceInfo]:
     """Load available cloud workspaces for the current authenticated user."""
     if context:
-        cached_workspaces = context.get_state("available_workspaces")
-        if isinstance(cached_workspaces, list) and all(
-            isinstance(item, WorkspaceInfo) for item in cached_workspaces
-        ):
-            return cached_workspaces
+        cached_raw = await context.get_state("available_workspaces")
+        if isinstance(cached_raw, list):
+            return [WorkspaceInfo.model_validate(item) for item in cached_raw]
 
     from basic_memory.mcp.async_client import get_cloud_control_plane_client
     from basic_memory.mcp.tools.utils import call_get
@@ -113,7 +111,10 @@ async def get_available_workspaces(context: Optional[Context] = None) -> list[Wo
         workspace_list = WorkspaceListResponse.model_validate(response.json())
 
     if context:
-        context.set_state("available_workspaces", workspace_list.workspaces)
+        await context.set_state(
+            "available_workspaces",
+            [ws.model_dump() for ws in workspace_list.workspaces],
+        )
 
     return workspace_list.workspaces
 
@@ -124,12 +125,12 @@ async def resolve_workspace_parameter(
 ) -> WorkspaceInfo:
     """Resolve workspace using explicit input, session cache, and cloud discovery."""
     if context:
-        cached_workspace = context.get_state("active_workspace")
-        if isinstance(cached_workspace, WorkspaceInfo) and (
-            workspace is None or _workspace_matches_identifier(cached_workspace, workspace)
-        ):
-            logger.debug(f"Using cached workspace from context: {cached_workspace.tenant_id}")
-            return cached_workspace
+        cached_raw = await context.get_state("active_workspace")
+        if isinstance(cached_raw, dict):
+            cached_workspace = WorkspaceInfo.model_validate(cached_raw)
+            if workspace is None or _workspace_matches_identifier(cached_workspace, workspace):
+                logger.debug(f"Using cached workspace from context: {cached_workspace.tenant_id}")
+                return cached_workspace
 
     workspaces = await get_available_workspaces(context=context)
     if not workspaces:
@@ -164,7 +165,7 @@ async def resolve_workspace_parameter(
         )
 
     if context:
-        context.set_state("active_workspace", selected_workspace)
+        await context.set_state("active_workspace", selected_workspace.model_dump())
         logger.debug(f"Cached workspace in context: {selected_workspace.tenant_id}")
 
     return selected_workspace
@@ -206,10 +207,12 @@ async def get_active_project(
 
     # Check if already cached in context
     if context:
-        cached_project = context.get_state("active_project")
-        if cached_project and cached_project.name == project:
-            logger.debug(f"Using cached project from context: {project}")
-            return cached_project
+        cached_raw = await context.get_state("active_project")
+        if isinstance(cached_raw, dict):
+            cached_project = ProjectItem.model_validate(cached_raw)
+            if cached_project.name == project:
+                logger.debug(f"Using cached project from context: {project}")
+                return cached_project
 
     # Validate project exists by calling API
     logger.debug(f"Validating project: {project}")
@@ -230,7 +233,7 @@ async def get_active_project(
 
     # Cache in context if available
     if context:
-        context.set_state("active_project", active_project)
+        await context.set_state("active_project", active_project.model_dump())
         logger.debug(f"Cached project in context: {project}")
 
     logger.debug(f"Validated project: {active_project.name}")
@@ -307,7 +310,7 @@ async def resolve_project_and_path(
                 is_default=resolved.is_default,
             )
             if context:
-                context.set_state("active_project", active_project)
+                await context.set_state("active_project", active_project.model_dump())
 
             resolved_path = f"{resolved.permalink}/{remainder}" if include_project else remainder
             return active_project, resolved_path, True

src/basic_memory/mcp/prompts/recent_activity.py

@@ -46,7 +46,7 @@ async def recent_activity_prompt(
     logger.info(f"Getting recent activity, timeframe: {timeframe}, project: {project}")
 
     # Call the tool function - it returns a well-formatted string
-    activity_summary = await recent_activity.fn(project=project, timeframe=timeframe)
+    activity_summary = await recent_activity(project=project, timeframe=timeframe)
 
     # Build the prompt response
     # The tool already returns formatted markdown, so we use it directly

src/basic_memory/mcp/tools/build_context.py

@@ -192,6 +192,7 @@ def _format_context_markdown(graph: GraphContext, project: str) -> str:
     - "json" (default): Slimmed JSON with redundant fields removed
     - "text": Compact markdown text for LLM consumption
     """,
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def build_context(
     url: MemoryUrl,

src/basic_memory/mcp/tools/canvas.py

@@ -16,6 +16,7 @@ from basic_memory.mcp.tools.utils import call_put, call_post, resolve_entity_id
 
 @mcp.tool(
     description="Create an Obsidian canvas file to visualize concepts and connections.",
+    annotations={"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
 )
 async def canvas(
     nodes: List[Dict[str, Any]],

src/basic_memory/mcp/tools/chatgpt_tools.py

@@ -92,7 +92,10 @@ def _format_document_for_chatgpt(
     }
 
 
-@mcp.tool(description="Search for content across the knowledge base")
+@mcp.tool(
+    description="Search for content across the knowledge base",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 async def search(
     query: str,
     context: Context | None = None,
@@ -115,7 +118,7 @@ async def search(
         default_project = config.default_project
 
         # Call underlying search_notes with sensible defaults for ChatGPT
-        results = await search_notes.fn(
+        results = await search_notes(
             query=query,
             project=default_project,  # Use default project for ChatGPT
             page=1,
@@ -156,7 +159,10 @@ async def search(
         return [{"type": "text", "text": json.dumps(error_results, ensure_ascii=False)}]
 
 
-@mcp.tool(description="Fetch the full contents of a search result document")
+@mcp.tool(
+    description="Fetch the full contents of a search result document",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 async def fetch(
     id: str,
     context: Context | None = None,
@@ -178,13 +184,15 @@ async def fetch(
         config = ConfigManager().config
         default_project = config.default_project
 
-        # Call underlying read_note function
-        content = await read_note.fn(
-            identifier=id,
-            project=default_project,  # Use default project for ChatGPT
-            page=1,
-            page_size=10,  # Default pagination
-            context=context,
+        # Call underlying read_note function (default output_format="text" returns str)
+        content = str(
+            await read_note(
+                identifier=id,
+                project=default_project,  # Use default project for ChatGPT
+                page=1,
+                page_size=10,  # Default pagination
+                context=context,
+            )
         )
 
         # Format the document for ChatGPT

src/basic_memory/mcp/tools/cloud_info.py

@@ -5,7 +5,10 @@ from pathlib import Path
 from basic_memory.mcp.server import mcp
 
 
-@mcp.tool("cloud_info")
+@mcp.tool(
+    "cloud_info",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 def cloud_info() -> str:
     """Return optional Basic Memory Cloud information and setup guidance."""
     content_path = Path(__file__).parent.parent / "resources" / "cloud_info.md"

src/basic_memory/mcp/tools/delete_note.py

@@ -146,7 +146,10 @@ delete_note("{project}", "correct-identifier-from-search")
 If the note should be deleted but the operation keeps failing, send a message to support@basicmemory.com."""
 
 
-@mcp.tool(description="Delete a note or directory by title, permalink, or path")
+@mcp.tool(
+    description="Delete a note or directory by title, permalink, or path",
+    annotations={"destructiveHint": True, "openWorldHint": False},
+)
 async def delete_note(
     identifier: str,
     is_directory: bool = False,

src/basic_memory/mcp/tools/edit_note.py

@@ -125,6 +125,7 @@ Error editing note '{identifier}': {error_message}
 
 @mcp.tool(
     description="Edit an existing markdown note using various operations like append, prepend, find_replace, or replace_section.",
+    annotations={"destructiveHint": False, "openWorldHint": False},
 )
 async def edit_note(
     identifier: str,

src/basic_memory/mcp/tools/list_directory.py

@@ -11,6 +11,7 @@ from basic_memory.mcp.server import mcp
 
 @mcp.tool(
     description="List directory contents with filtering and depth control.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def list_directory(
     dir_name: str = "/",

src/basic_memory/mcp/tools/move_note.py

@@ -343,6 +343,7 @@ delete_note("{identifier}")
 
 @mcp.tool(
     description="Move a note or directory to a new location, updating database and maintaining links.",
+    annotations={"destructiveHint": False, "openWorldHint": False},
 )
 async def move_note(
     identifier: str,

src/basic_memory/mcp/tools/project_management.py

@@ -14,7 +14,10 @@ from basic_memory.schemas.project_info import ProjectInfoRequest
 from basic_memory.utils import generate_permalink
 
 
-@mcp.tool("list_memory_projects")
+@mcp.tool(
+    "list_memory_projects",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 async def list_memory_projects(
     output_format: Literal["text", "json"] = "text",
     context: Context | None = None,
@@ -77,7 +80,10 @@ async def list_memory_projects(
         return result
 
 
-@mcp.tool("create_memory_project")
+@mcp.tool(
+    "create_memory_project",
+    annotations={"destructiveHint": False, "openWorldHint": False},
+)
 async def create_memory_project(
     project_name: str,
     project_path: str,
@@ -193,7 +199,9 @@ async def create_memory_project(
         return result
 
 
-@mcp.tool()
+@mcp.tool(
+    annotations={"destructiveHint": True, "openWorldHint": False},
+)
 async def delete_project(project_name: str, context: Context | None = None) -> str:
     """Delete a Basic Memory project.
 

src/basic_memory/mcp/tools/read_content.py

@@ -148,7 +148,10 @@ def optimize_image(img, content_length, max_output_bytes=350000):
             return buf.getvalue()
 
 
-@mcp.tool(description="Read a file's raw content by path or permalink")
+@mcp.tool(
+    description="Read a file's raw content by path or permalink",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 async def read_content(
     path: str,
     project: Optional[str] = None,

src/basic_memory/mcp/tools/read_note.py

@@ -59,6 +59,7 @@ def _parse_opening_frontmatter(content: str) -> tuple[str, dict | None]:
     description="Read a markdown note by title or permalink.",
     # TODO: re-enable once MCP client rendering is working
     # meta={"ui/resourceUri": "ui://basic-memory/note-preview"},
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def read_note(
     identifier: str,
@@ -228,7 +229,7 @@ async def read_note(
 
         # Fallback 1: Try title search via API
         logger.info(f"Search title for: {identifier}")
-        title_results = await search_notes.fn(
+        title_results = await search_notes(
             query=identifier,
             search_type="title",
             project=active_project.name,
@@ -280,7 +281,7 @@ async def read_note(
 
         # Fallback 2: Text search as a last resort
         logger.info(f"Title search failed, trying text search for: {identifier}")
-        text_results = await search_notes.fn(
+        text_results = await search_notes(
             query=identifier,
             search_type="text",
             project=active_project.name,

src/basic_memory/mcp/tools/recent_activity.py

@@ -35,6 +35,7 @@ from basic_memory.schemas.search import SearchItemType
     - "3 weeks ago"
     Or standard formats like "7d"
     """,
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def recent_activity(
     type: Union[str, List[str]] = "",

src/basic_memory/mcp/tools/release_notes.py

@@ -5,7 +5,10 @@ from pathlib import Path
 from basic_memory.mcp.server import mcp
 
 
-@mcp.tool("release_notes")
+@mcp.tool(
+    "release_notes",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 def release_notes() -> str:
     """Return the latest product release notes for optional user review."""
     content_path = Path(__file__).parent.parent / "resources" / "release_notes.md"

src/basic_memory/mcp/tools/schema.py

@@ -71,6 +71,7 @@ def _no_schema_guidance(note_type: str, tool_name: str) -> str:
 
 @mcp.tool(
     description="Validate notes against their Picoschema definitions.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def schema_validate(
     note_type: Optional[str] = None,
@@ -166,6 +167,7 @@ async def schema_validate(
 
 @mcp.tool(
     description="Analyze existing notes and suggest a Picoschema definition.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def schema_infer(
     note_type: str,
@@ -267,6 +269,7 @@ async def schema_infer(
 
 @mcp.tool(
     description="Detect drift between a schema definition and actual note usage.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def schema_diff(
     note_type: str,

src/basic_memory/mcp/tools/search.py

@@ -250,6 +250,7 @@ Error searching for '{query}': {error_message}
     description="Search across all content in the knowledge base with advanced syntax support.",
     # TODO: re-enable once MCP client rendering is working
     # meta={"ui/resourceUri": "ui://basic-memory/search-results"},
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def search_notes(
     query: str,
@@ -514,6 +515,7 @@ async def search_notes(
 
 @mcp.tool(
     description="Search entities by structured frontmatter metadata.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def search_by_metadata(
     filters: Dict[str, Any],

src/basic_memory/mcp/tools/ui_sdk.py

@@ -20,6 +20,7 @@ def _text_block(message: str) -> List[ContentBlock]:
 @mcp.tool(
     description="Search notes and return an embedded MCP-UI resource (raw HTML).",
     output_schema=None,
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def search_notes_ui(
     query: str,
@@ -36,7 +37,7 @@ async def search_notes_ui(
     context: Context | None = None,
 ) -> List[ContentBlock]:
     """Return a search results UI as an embedded MCP-UI resource."""
-    result = await search_notes.fn(
+    result = await search_notes(
         query=query,
         project=project,
         page=page,
@@ -82,6 +83,7 @@ async def search_notes_ui(
 @mcp.tool(
     description="Read a note and return an embedded MCP-UI resource (raw HTML).",
     output_schema=None,
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def read_note_ui(
     identifier: str,
@@ -91,7 +93,7 @@ async def read_note_ui(
     context: Context | None = None,
 ) -> List[ContentBlock]:
     """Return a note preview UI as an embedded MCP-UI resource."""
-    content = await read_note.fn(
+    content = await read_note(
         identifier=identifier,
         project=project,
         page=page,

src/basic_memory/mcp/tools/view_note.py

@@ -12,6 +12,7 @@ from basic_memory.mcp.tools.read_note import read_note
 
 @mcp.tool(
     description="View a note as a formatted artifact for better readability.",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
 )
 async def view_note(
     identifier: str,
@@ -57,14 +58,16 @@ async def view_note(
     """
     logger.info(f"Viewing note: {identifier} in project: {project}")
 
-    # Call the existing read_note logic
-    content = await read_note.fn(
-        identifier=identifier,
-        project=project,
-        workspace=workspace,
-        page=page,
-        page_size=page_size,
-        context=context,
+    # Call the existing read_note logic (default output_format="text" returns str)
+    content = str(
+        await read_note(
+            identifier=identifier,
+            project=project,
+            workspace=workspace,
+            page=page,
+            page_size=page_size,
+            context=context,
+        )
     )
 
     # Check if this is an error message (note not found)

src/basic_memory/mcp/tools/workspaces.py

@@ -6,7 +6,10 @@ from basic_memory.mcp.project_context import get_available_workspaces
 from basic_memory.mcp.server import mcp
 
 
-@mcp.tool(description="List available cloud workspaces (tenant_id, type, role, and name).")
+@mcp.tool(
+    description="List available cloud workspaces (tenant_id, type, role, and name).",
+    annotations={"readOnlyHint": True, "openWorldHint": False},
+)
 async def list_workspaces(context: Context | None = None) -> str:
     """List workspaces available to the current cloud user."""
     workspaces = await get_available_workspaces(context=context)

src/basic_memory/mcp/tools/write_note.py

@@ -16,6 +16,7 @@ TagType = Union[List[str], str, None]
 
 @mcp.tool(
     description="Create or update a markdown note. Returns a markdown formatted summary of the semantic content.",
+    annotations={"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
 )
 async def write_note(
     title: str,