fix(mcp): resolve projects by external_id, remove workspace from MCP tools (#777)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Paul Hernandez
2026-05-01 05:15:44 -05:00
committed by GitHub
parent 2fccc74a20
commit 0b335476d6
31 changed files with 688 additions and 447 deletions
+41 -1
View File
@@ -91,10 +91,11 @@ SQLite Database (Index)
# List all projects
projects = await list_memory_projects()
# Response structure:
# Response structure (each entry includes external_id you can pass as project_id):
# [
# {
# "name": "main",
# "external_id": "550e8400-e29b-41d4-a716-446655440000",
# "path": "/Users/name/notes",
# "is_default": True,
# "note_count": 156,
@@ -102,6 +103,7 @@ projects = await list_memory_projects()
# },
# {
# "name": "work",
# "external_id": "9f86d081-884c-42a3-b5e3-1c0c5b4c8e52",
# "path": "/Users/name/work-notes",
# "is_default": False,
# "note_count": 89,
@@ -164,6 +166,44 @@ active_project = "main"
results = await search_notes(query="topic", project=active_project)
```
### `project` vs `project_id`
Every project has two identifiers:
- **`project`** — human-readable name (e.g., `"main"`). Easy to use, but can collide across cloud workspaces.
- **`project_id`** — stable `external_id` UUID. Always unambiguous; takes precedence over `project` when both are passed.
**When to prefer `project_id`:**
1. **Cloud multi-workspace setups.** If the user belongs to more than one workspace (personal + organization, or several organizations) and the same project name might exist in more than one of them, pass `project_id` to route to the exact project. Without it, name resolution falls back to the default workspace, which may not be the one the user means.
2. **After `list_memory_projects()`.** Once you have the `external_id`, prefer using it — it's the same number of characters in JSON and saves a name-resolution round-trip.
3. **When persisting a project choice across a long session.** UUIDs are stable; names can be renamed.
**When `project` (name) is fine:**
- Local single-workspace setups (no collision risk).
- One-off operations where the name is clearly visible to the user (e.g., quick `search_notes(project="main", ...)`).
- The user explicitly references a project by name in their message.
**Example — cloud multi-workspace pattern:**
```python
# Discover and pick the right project for this user
projects = await list_memory_projects()
target = next(p for p in projects if p["name"] == "research" and p["workspace"]["slug"] == "acme")
# Use the UUID for all subsequent operations — no ambiguity
await write_note(
title="Meeting Notes",
content="...",
folder="meetings",
project_id=target["external_id"],
)
results = await search_notes(query="kickoff", project_id=target["external_id"])
```
**Precedence rule:** When both are passed, `project_id` wins. This lets you safely supply `project="main"` for backward compatibility while still routing precisely with `project_id`.
### Cross-Project Operations
**Some tools work across all projects when project parameter omitted:**
+2 -6
View File
@@ -99,9 +99,7 @@ def _abort_if_mcp_processes_alive() -> None:
if not zombies:
return
console.print(
"[red]Refusing to reset:[/red] basic-memory MCP processes are still running."
)
console.print("[red]Refusing to reset:[/red] basic-memory MCP processes are still running.")
console.print(
"[yellow]On macOS/Linux these would keep reading the deleted memory.db inode "
"and return phantom search results (see #765).[/yellow]"
@@ -117,9 +115,7 @@ def _abort_if_mcp_processes_alive() -> None:
"Where-Object {$_.CommandLine -like '*basic-memory*mcp*'}[/green]"
)
else:
console.print(
" 2. Verify nothing remains: [green]pgrep -fa 'basic-memory mcp'[/green]"
)
console.print(" 2. Verify nothing remains: [green]pgrep -fa 'basic-memory mcp'[/green]")
console.print(" 3. Re-run [green]bm reset[/green].")
raise typer.Exit(1)
+54 -27
View File
@@ -62,9 +62,12 @@ def write_note(
help="The project to write to. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -106,7 +109,7 @@ def write_note(
content=content,
directory=folder,
project=project,
workspace=workspace,
project_id=project_id,
tags=tags,
output_format="json",
)
@@ -132,9 +135,12 @@ def read_note(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -156,7 +162,7 @@ def read_note(
mcp_read_note(
identifier=identifier,
project=project,
workspace=workspace,
project_id=project_id,
include_frontmatter=include_frontmatter,
output_format="json",
)
@@ -195,9 +201,12 @@ def edit_note(
help="The project to edit. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -222,7 +231,7 @@ def edit_note(
operation=operation,
content=content,
project=project,
workspace=workspace,
project_id=project_id,
section=section,
find_text=find_text,
expected_replacements=expected_replacements,
@@ -260,9 +269,12 @@ def build_context(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -284,7 +296,7 @@ def build_context(
mcp_build_context(
url=url,
project=project,
workspace=workspace,
project_id=project_id,
depth=depth,
timeframe=timeframe,
page=page,
@@ -317,9 +329,12 @@ def recent_activity(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -346,7 +361,7 @@ def recent_activity(
page=page,
page_size=page_size,
project=project,
workspace=workspace,
project_id=project_id,
output_format="json",
)
)
@@ -408,9 +423,12 @@ def search_notes(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -482,7 +500,7 @@ def search_notes(
mcp_search(
query=query or None,
project=project,
workspace=workspace,
project_id=project_id,
search_type=search_type,
output_format="json",
page=page,
@@ -592,9 +610,12 @@ def schema_validate(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -629,7 +650,7 @@ def schema_validate(
note_type=note_type,
identifier=identifier,
project=project,
workspace=workspace,
project_id=project_id,
output_format="json",
)
)
@@ -660,9 +681,12 @@ def schema_infer(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -686,7 +710,7 @@ def schema_infer(
note_type=note_type,
threshold=threshold,
project=project,
workspace=workspace,
project_id=project_id,
output_format="json",
)
)
@@ -714,9 +738,12 @@ def schema_diff(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
project_id: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
typer.Option(
"--project-id",
help="Project external_id (UUID). Takes precedence over --project; use to disambiguate same-named projects across cloud workspaces.",
),
] = None,
local: bool = typer.Option(
False, "--local", help="Force local API routing (ignore cloud mode)"
@@ -738,7 +765,7 @@ def schema_diff(
mcp_schema_diff(
note_type=note_type,
project=project,
workspace=workspace,
project_id=project_id,
output_format="json",
)
)
+54 -39
View File
@@ -10,8 +10,9 @@ compatibility with existing MCP tools.
import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import AsyncIterator, Awaitable, Callable, Optional, List, Tuple, cast
from uuid import UUID
from httpx import AsyncClient
from httpx._types import (
@@ -52,11 +53,12 @@ class WorkspaceProjectEntry:
@dataclass(frozen=True)
class WorkspaceProjectIndex:
"""Session-local cloud project lookup index keyed by project permalink."""
"""Session-local cloud project lookup index keyed by project permalink and external_id."""
workspaces: tuple[WorkspaceInfo, ...]
entries: tuple[WorkspaceProjectEntry, ...]
entries_by_permalink: dict[str, tuple[WorkspaceProjectEntry, ...]]
entries_by_external_id: dict[str, WorkspaceProjectEntry] = field(default_factory=dict)
failed_workspaces: tuple[WorkspaceInfo, ...] = ()
@@ -351,10 +353,12 @@ def _build_workspace_project_index(
*,
failed_workspaces: tuple[WorkspaceInfo, ...] = (),
) -> WorkspaceProjectIndex:
"""Build the permalink lookup table for workspace-project entries."""
"""Build the permalink and external_id lookup tables for workspace-project entries."""
grouped: dict[str, list[WorkspaceProjectEntry]] = {}
by_external_id: dict[str, WorkspaceProjectEntry] = {}
for entry in entries:
grouped.setdefault(entry.project.permalink, []).append(entry)
by_external_id[entry.project.external_id] = entry
return WorkspaceProjectIndex(
workspaces=workspaces,
@@ -363,6 +367,7 @@ def _build_workspace_project_index(
permalink: tuple(items)
for permalink, items in sorted(grouped.items(), key=lambda item: item[0])
},
entries_by_external_id=by_external_id,
failed_workspaces=failed_workspaces,
)
@@ -549,8 +554,20 @@ async def resolve_workspace_project_identifier(
project: str,
context: Optional[Context] = None,
) -> WorkspaceProjectEntry:
"""Resolve an unqualified or ``<workspace>/<project>`` cloud project identifier."""
"""Resolve a project by external_id (UUID), qualified name, or unqualified name."""
index = await _ensure_workspace_project_index(context=context)
# Fast path: direct lookup by external_id when the identifier is a UUID
# Canonicalize via str(UUID(...)) so uppercase, brace-wrapped, or urn:uuid forms
# all hash to the same lowercase-hyphenated key as the stored external_ids.
try:
canonical_external_id = str(UUID(project))
entry = index.entries_by_external_id.get(canonical_external_id)
if entry:
return entry
except ValueError:
pass
workspace_slug, project_identifier = _split_qualified_project_identifier(project)
project_permalink = generate_permalink(project_identifier)
@@ -617,6 +634,11 @@ async def resolve_workspace_project_identifier(
return cached_matches[0]
if len(matches) > 1:
# Prefer the project in the default workspace when name is ambiguous
default_match = next((entry for entry in matches if entry.workspace.is_default), None)
if default_match:
return default_match
choices = _format_qualified_choices(matches)
details = "\n".join(
f"- {entry.workspace.name} ({entry.workspace.slug}): {entry.qualified_name}"
@@ -956,8 +978,8 @@ def detect_project_from_url_prefix(identifier: str, config: BasicMemoryConfig) -
@asynccontextmanager
async def get_project_client(
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Optional[Context] = None,
project_id: Optional[str] = None,
) -> AsyncIterator[Tuple[AsyncClient, ProjectItem]]:
"""Resolve project, create correctly-routed client, and validate project.
@@ -972,16 +994,13 @@ async def get_project_client(
3. Cloud project mode → resolve project through workspace/project index
4. Otherwise → local ASGI client
Workspace resolution priority (when cloud routing):
1. Explicit ``workspace`` parameter
2. Per-project ``workspace_id`` from config
3. Qualified project identifier (``<workspace-slug>/<project>``)
4. Workspace/project index lookup with collision detection
Args:
project: Optional explicit project parameter
workspace: Optional cloud workspace selector (tenant_id or unique name)
project: Optional explicit project parameter (name or permalink)
context: Optional FastMCP context for caching
project_id: Optional project external_id (UUID). When provided, takes
precedence over ``project`` and disambiguates the project across
workspaces. Use this when the same project name exists in multiple
cloud workspaces.
Yields:
Tuple of (client, active_project)
@@ -998,8 +1017,12 @@ async def get_project_client(
is_factory_mode,
)
# When project_id (UUID) is provided, prefer it as the resolution identifier.
# external_id is unambiguous across workspaces; project name can collide.
project_identifier = project_id if project_id else project
# Step 1: Resolve project name from config (no network call)
resolved_project = await resolve_project_parameter(project, context=context)
resolved_project = await resolve_project_parameter(project_identifier, context=context)
config = ConfigManager().config
factory_mode = is_factory_mode()
explicit_cloud_routing = _explicit_routing() and not _force_local_mode()
@@ -1050,19 +1073,14 @@ async def get_project_client(
project_entry = config.projects.get(resolved_project)
project_mode = config.get_project_mode(resolved_project)
# Trigger: workspace provided for a local project (without explicit --cloud)
# Why: workspace selection is a cloud routing concern only
# Outcome: fail fast with a deterministic guidance message
if (
not factory_mode
and project_mode != ProjectMode.CLOUD
and workspace is not None
and not _explicit_routing()
):
raise ValueError(
f"Workspace '{workspace}' cannot be used with local project '{resolved_project}'. "
"Workspace selection is only supported for cloud-mode projects."
)
# Trigger: identifier is a UUID (project_id) but local config keys by name only
# Why: get_project_mode defaults to CLOUD for unknown identifiers; a UUID is
# never registered in local config, so it would always falsely route cloud
# Outcome: in pure local mode, treat UUID identifiers as local routing; cloud
# discovery still happens when factory/explicit/credentials are present
cloud_available = factory_mode or explicit_cloud_routing or has_cloud_credentials(config)
if project_id and not cloud_available:
project_mode = ProjectMode.LOCAL
if factory_mode or project_mode == ProjectMode.CLOUD or explicit_cloud_routing:
route_mode = "factory" if factory_mode else "cloud_proxy"
@@ -1070,17 +1088,8 @@ async def get_project_client(
workspace_id: str
project_for_api = _unqualified_project_identifier(resolved_project)
# Trigger: a script or config entry pins the tenant explicitly
# Why: explicit tenant configuration remains the escape hatch during migration
# Outcome: route to that workspace, but validate the project name inside it
if workspace is not None:
active_ws = await resolve_workspace_parameter(workspace=workspace, context=context)
workspace_id = active_ws.tenant_id
elif project_entry and project_entry.workspace_id:
# Trigger: the local project config already stores the cloud tenant id.
# Why: routing can send that id directly; requiring workspace discovery here
# would turn a control-plane listing outage into a project routing failure.
# Outcome: preserve project-scoped routing even when discovery is unavailable.
if project_entry and project_entry.workspace_id:
# Per-project config stores the cloud tenant id directly
workspace_id = project_entry.workspace_id
else:
resolved_entry = cloud_default_entry
@@ -1120,6 +1129,12 @@ async def get_project_client(
route_mode=route_mode,
):
logger.debug("Using default local ASGI routing for project client")
async with get_client(project_name=resolved_project) as client:
# Trigger: UUID identifiers won't match name-keyed local config entries.
# Why: get_client(project_name=<uuid>) would consult get_project_mode and
# default to CLOUD for unknown identifiers, breaking pure-local routing.
# Outcome: skip per-project routing for UUIDs — local mode routes every
# project through the same ASGI client; the API resolves the UUID below.
client_kwargs = {} if project_id else {"project_name": resolved_project}
async with get_client(**client_kwargs) as client:
active_project = await get_active_project(client, resolved_project, context)
yield client, active_project
@@ -18,13 +18,15 @@ Basic Memory creates a semantic knowledge graph from markdown files. Focus on bu
**Resolution priority:**
1. CLI constraint: `BASIC_MEMORY_MCP_PROJECT` env var (highest priority)
2. Explicit parameter: `project="name"` in tool calls
2. Explicit parameter: `project_id="<uuid>"` (preferred when known) or `project="name"` in tool calls
3. Default project: `default_project` in config (fallback)
**`project` vs `project_id`:** Every project has a stable `external_id` (UUID) returned by `list_memory_projects()`. Pass it as `project_id=...` to address a project unambiguously — required when the same project name exists in multiple cloud workspaces. For local single-project setups, the `project` name is fine.
### Quick Setup Check
```python
# Discover projects
# Discover projects (each entry includes external_id you can pass as project_id)
projects = await list_memory_projects()
```
@@ -169,6 +171,14 @@ await write_note(
**Multi-project users:**
- Always specify project explicitly in tool calls
**Cloud multi-workspace users:** project names can collide across workspaces. After calling `list_memory_projects()`, prefer the project's `external_id` via `project_id=...` for any subsequent tool calls — it routes to the exact project regardless of name collisions. The `project` name parameter falls back to the default workspace on ambiguity, which may not be what you want.
```python
# Cloud / multi-workspace: prefer project_id (UUID) once you've discovered it
projects = await list_memory_projects()
results = await search_notes(query="auth", project_id=projects[0]["external_id"])
```
**Discovery:**
```python
# Start with discovery
+9 -3
View File
@@ -139,7 +139,7 @@ async def build_context(
Field(validation_alias=AliasChoices("url", "uri", "memory_url")),
],
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
depth: str | int | None = 1,
timeframe: Annotated[
Optional[TimeFrame],
@@ -179,6 +179,9 @@ async def build_context(
Args:
project: Project name to build context from. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
url: memory:// URI pointing to discussion content (e.g. memory://specs/search)
depth: How many relation hops to traverse (1-3 recommended for performance)
timeframe: How far back to look. Supports natural language like "2 days ago", "last week"
@@ -228,7 +231,7 @@ async def build_context(
entrypoint="mcp",
tool_name="build_context",
requested_project=project,
workspace_id=workspace,
requested_project_id=project_id,
depth=depth or 1,
timeframe=timeframe,
page=page,
@@ -237,7 +240,10 @@ async def build_context(
output_format=output_format,
is_memory_url=str(url).startswith("memory://"),
):
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=build_context project={active_project.name} "
f"url={url} depth={depth} timeframe={timeframe} output_format={output_format}"
+8 -2
View File
@@ -29,7 +29,7 @@ async def canvas(
Field(validation_alias=AliasChoices("directory", "folder", "dir", "path")),
],
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
context: Context | None = None,
) -> str:
"""Create an Obsidian canvas file with the provided nodes and edges.
@@ -46,6 +46,9 @@ async def canvas(
Args:
project: Project name to create canvas in. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
nodes: List of node objects following JSON Canvas 1.0 spec
edges: List of edge objects following JSON Canvas 1.0 spec
title: The title of the canvas (will be saved as title.canvas)
@@ -100,7 +103,10 @@ async def canvas(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
# Ensure path has .canvas extension
file_title = title if title.endswith(".canvas") else f"{title}.canvas"
file_path = f"{directory}/{file_title}"
+8 -2
View File
@@ -159,7 +159,7 @@ async def delete_note(
Field(default=False, validation_alias=AliasChoices("is_directory", "is_dir")),
] = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> bool | str | dict:
@@ -183,6 +183,9 @@ 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.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
output_format: "text" preserves existing behavior (bool/string). "json"
returns machine-readable deletion metadata.
context: Optional FastMCP context for performance caching.
@@ -237,7 +240,10 @@ async def delete_note(
if detected:
project = detected
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.debug(
f"Deleting {'directory' if is_directory else 'note'}: {identifier} in project: {active_project.name}"
)
+11 -9
View File
@@ -176,13 +176,11 @@ async def edit_note(
content: Annotated[
str,
Field(
validation_alias=AliasChoices(
"content", "new_content", "replacement", "replace_with"
)
validation_alias=AliasChoices("content", "new_content", "replacement", "replace_with")
),
],
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
# Section/heading naming varies across tools; accept the descriptive forms.
section: Annotated[
Optional[str],
@@ -197,9 +195,7 @@ async def edit_note(
Optional[str],
Field(
default=None,
validation_alias=AliasChoices(
"find_text", "find", "old_text", "old_content", "search"
),
validation_alias=AliasChoices("find_text", "find", "old_text", "old_content", "search"),
),
] = None,
expected_replacements: Optional[int] = None,
@@ -228,6 +224,9 @@ async def edit_note(
content: The content to add or use for replacement
project: Project name to edit in. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
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)
@@ -303,14 +302,17 @@ async def edit_note(
entrypoint="mcp",
tool_name="edit_note",
requested_project=project,
workspace_id=workspace,
requested_project_id=project_id,
edit_operation=operation,
output_format=output_format,
has_section=bool(section),
has_find_text=bool(find_text),
expected_replacements=effective_replacements,
):
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=edit_note project={active_project.name} "
f"identifier={identifier} operation={operation} output_format={output_format}"
+8 -2
View File
@@ -32,7 +32,7 @@ async def list_directory(
),
] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
context: Context | None = None,
) -> str:
"""List directory contents from the knowledge base with optional filtering.
@@ -50,6 +50,9 @@ async def list_directory(
Examples: "*.md", "*meeting*", "project_*"
project: Project name to list directory from. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -77,7 +80,10 @@ async def list_directory(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.debug(
f"Listing directory '{dir_name}' in project {project} with depth={depth}, glob='{file_name_glob}'"
)
+8 -2
View File
@@ -371,7 +371,7 @@ async def move_note(
Field(default=False, validation_alias=AliasChoices("is_directory", "is_dir")),
] = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | dict:
@@ -396,6 +396,9 @@ 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.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
output_format: "text" returns existing markdown guidance/success text. "json"
returns machine-readable move metadata.
context: Optional FastMCP context for performance caching.
@@ -495,7 +498,10 @@ async def move_note(
"error": "DESTINATION_FOLDER_NOT_FOR_DIRECTORIES",
}
return f"# Move Failed - Invalid Parameters\n\n{error_msg}"
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
destination_target = destination_folder or destination_path
logger.info(
f"MCP tool call tool=move_note project={active_project.name} "
@@ -11,11 +11,10 @@ from fastmcp import Context
from loguru import logger
from basic_memory.config import ConfigManager, has_cloud_credentials
from basic_memory.mcp.async_client import get_client, get_cloud_proxy_client, is_factory_mode
from basic_memory.mcp.async_client import get_client, is_factory_mode
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
ensure_workspace_project_index,
resolve_workspace_parameter,
)
from basic_memory.mcp.server import mcp
from basic_memory.schemas.project_info import ProjectInfoRequest, ProjectItem, ProjectList
@@ -25,36 +24,6 @@ from basic_memory.utils import generate_permalink
# --- Helpers for dual-fetch + merge ---
async def _fetch_cloud_projects(
workspace: str | None = None,
context: Context | None = None,
) -> ProjectList | None:
"""Fetch projects from the cloud API, returning None on failure.
Logs warnings on failure so list_memory_projects can fall back to local-only
results. Project-scoped routing does not use this listing fallback.
"""
try:
from basic_memory.mcp.clients import ProjectClient
async with get_cloud_proxy_client(workspace=workspace) as cloud_client:
cloud_project_client = ProjectClient(cloud_client)
cloud_list = await cloud_project_client.list_projects()
if context: # pragma: no cover
await context.info(f"Discovered {len(cloud_list.projects)} cloud projects")
return cloud_list
except Exception as exc:
logger.warning(
f"Cloud project discovery failed while listing projects; "
f"showing local-only project list: {exc}"
)
if context: # pragma: no cover
await context.info(
"Cloud project discovery failed while listing projects; showing local projects only"
)
return None
def _merge_projects(
local_list: ProjectList | None,
cloud_list: ProjectList | None,
@@ -126,9 +95,13 @@ def _merge_projects(
ws_type = cloud_workspace_type if cloud_proj else None
ws_tenant_id = cloud_workspace_tenant_id if cloud_proj else None
proj = cloud_proj or local_proj
external_id = proj.external_id if proj else None
merged.append(
{
"name": name,
"external_id": external_id,
"path": path,
"local_path": local_path,
"cloud_path": cloud_path,
@@ -184,6 +157,7 @@ def _merge_workspace_projects(
merged.append(
{
"name": cloud_proj.name,
"external_id": cloud_proj.external_id,
"path": local_path or cloud_path,
"local_path": local_path,
"cloud_path": cloud_path,
@@ -207,6 +181,7 @@ def _merge_workspace_projects(
merged.append(
{
"name": project.name,
"external_id": project.external_id,
"path": project.path,
"local_path": project.path,
"cloud_path": None,
@@ -248,9 +223,9 @@ def _format_project_list_text(merged: list[dict]) -> str:
name = project["name"]
label = f"{display_name} ({name})" if display_name else name
source = project["source"]
qualified_name = project.get("qualified_name")
qualified_suffix = f" [{qualified_name}]" if qualified_name else ""
result += f"- {label} ({source}){qualified_suffix}\n"
external_id = project.get("external_id", "")
id_suffix = f" [{external_id}]" if external_id else ""
result += f"- {label} ({source}){id_suffix}\n"
result += "\n" + "" * 40 + "\n"
result += "Next: Ask which project to use for this session.\n"
@@ -282,7 +257,6 @@ def _format_project_list_json(
)
async def list_memory_projects(
output_format: Literal["text", "json"] = "text",
workspace: str | None = None,
context: Context | None = None,
) -> str | dict:
"""List all available projects with their status.
@@ -290,11 +264,14 @@ async def list_memory_projects(
Shows projects from both local and cloud sources when cloud credentials
are available, merging by permalink to give a unified view.
Each project entry includes an `external_id` (UUID). Pass that value as the
`project_id` parameter on other tools to address a specific project
unambiguously across cloud workspaces — useful when the same project name
exists in more than one workspace.
Args:
output_format: "text" returns the existing human-readable project list.
"json" returns structured project metadata.
workspace: Cloud workspace name or tenant_id. Falls back to
config.default_workspace when not specified.
context: Optional FastMCP context for progress/status logging.
"""
if context: # pragma: no cover
@@ -309,7 +286,7 @@ async def list_memory_projects(
# Why: there is no local ASGI server; the factory IS the cloud source
# Outcome: single fetch, projects reported as source="cloud" with workspace metadata
if is_factory_mode():
async with get_client(workspace=workspace) as client:
async with get_client() as client:
project_client = ProjectClient(client)
project_list = await project_client.list_projects()
@@ -324,13 +301,7 @@ async def list_memory_projects(
workspaces = await get_available_workspaces(context)
if workspaces:
# In factory mode the user is authenticated to a single workspace;
# use the explicit workspace param or fall back to the first available.
matched = None
if workspace:
matched = next((ws for ws in workspaces if ws.tenant_id == workspace), None)
if matched is None:
matched = workspaces[0]
matched = workspaces[0]
cloud_ws_name = matched.name
cloud_ws_type = matched.workspace_type
cloud_ws_tenant_id = matched.tenant_id
@@ -372,42 +343,19 @@ async def list_memory_projects(
cloud_ws_is_default = False
config = ConfigManager().config
if has_cloud_credentials(config):
if workspace:
try:
active_workspace = await resolve_workspace_parameter(workspace, context)
except Exception as exc:
logger.warning(
f"Cloud workspace discovery failed while listing projects for "
f"workspace '{workspace}'; trying direct workspace routing before "
f"falling back to local-only project list: {exc}"
try:
workspace_index = await ensure_workspace_project_index(context=context)
cloud_entries = workspace_index.entries
except Exception as exc:
logger.warning(
f"Cloud workspace project index discovery failed while listing projects; "
f"showing local-only project list: {exc}"
)
if context: # pragma: no cover
await context.info(
"Cloud workspace project discovery failed while listing projects; "
"showing local projects only"
)
if context: # pragma: no cover
await context.info(
"Cloud workspace discovery failed while listing projects; "
"trying direct workspace routing"
)
cloud_list = await _fetch_cloud_projects(workspace, context)
else:
cloud_list = await _fetch_cloud_projects(active_workspace.tenant_id, context)
cloud_ws_name = active_workspace.name
cloud_ws_type = active_workspace.workspace_type
cloud_ws_tenant_id = active_workspace.tenant_id
cloud_ws_slug = active_workspace.slug
cloud_ws_is_default = active_workspace.is_default
else:
try:
workspace_index = await ensure_workspace_project_index(context=context)
cloud_entries = workspace_index.entries
except Exception as exc:
logger.warning(
f"Cloud workspace project index discovery failed while listing projects; "
f"showing local-only project list: {exc}"
)
if context: # pragma: no cover
await context.info(
"Cloud workspace project discovery failed while listing projects; "
"showing local projects only"
)
if cloud_entries:
merged = _merge_workspace_projects(local_list, cloud_entries)
@@ -515,6 +463,7 @@ async def create_memory_project(
if output_format == "json":
return {
"name": existing_match.name,
"external_id": existing_match.external_id,
"path": existing_match.path,
"is_default": is_default,
"created": False,
@@ -524,6 +473,7 @@ async def create_memory_project(
f"✓ Project already exists: {existing_match.name}\n\n"
f"Project Details:\n"
f"• Name: {existing_match.name}\n"
f"• External ID: {existing_match.external_id}\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"
@@ -538,6 +488,7 @@ async def create_memory_project(
new_project = status_response.new_project
return {
"name": new_project.name if new_project else project_name,
"external_id": new_project.external_id if new_project else None,
"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
@@ -551,6 +502,7 @@ async def create_memory_project(
if status_response.new_project:
result += "Project Details:\n"
result += f"• Name: {status_response.new_project.name}\n"
result += f"• External ID: {status_response.new_project.external_id}\n"
result += f"• Path: {status_response.new_project.path}\n"
if set_default:
+12 -4
View File
@@ -164,7 +164,7 @@ async def read_content(
Field(validation_alias=AliasChoices("path", "file_path", "filepath", "file")),
],
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
context: Context | None = None,
) -> dict:
"""Read a file's raw content by path or permalink.
@@ -185,6 +185,9 @@ async def read_content(
- A permalink (docs/example)
project: Project name to read from. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -222,9 +225,14 @@ async def read_content(
logger.info(f"MCP tool call tool=read_content project={project} path={path}")
async with get_project_client(project, workspace, context) as (client, active_project):
# Resolve path with project-prefix awareness for memory:// URLs
_, url, _ = await resolve_project_and_path(client, path, project, context)
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
# Resolve path with project-prefix awareness for memory:// URLs.
# Use active_project.name so resolution stays consistent when project_id
# was used or `project` was wrong/ambiguous (matches the cached resolution).
_, url, _ = await resolve_project_and_path(client, path, active_project.name, context)
# Validate path to prevent path traversal attacks
# For memory:// URLs, validate the extracted path (not the raw URL which
+21 -6
View File
@@ -70,7 +70,7 @@ def _parse_opening_frontmatter(content: str) -> tuple[str, dict | None]:
async def read_note(
identifier: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
include_frontmatter: bool = False,
context: Context | None = None,
@@ -94,6 +94,9 @@ async def read_note(
project: Project name to read from. Optional - server will resolve using the
hierarchy above. If unknown, use list_memory_projects() to discover
available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
identifier: The title or permalink of the note to read
Can be a full memory:// URL, a permalink, a title, or search text
output_format: "text" returns markdown content or guidance text.
@@ -138,13 +141,21 @@ async def read_note(
entrypoint="mcp",
tool_name="read_note",
requested_project=project,
workspace_id=workspace,
requested_project_id=project_id,
output_format=output_format,
include_frontmatter=include_frontmatter,
):
async with get_project_client(project, workspace, context) as (client, active_project):
# Resolve identifier with project-prefix awareness for memory:// URLs
_, entity_path, _ = await resolve_project_and_path(client, identifier, project, context)
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
# Resolve identifier with project-prefix awareness for memory:// URLs.
# Pass active_project.name (the canonical resolved name) rather than the
# original `project` arg so the inner get_active_project cache hits even
# when project_id was used or `project` was wrong/ambiguous.
_, entity_path, _ = await resolve_project_and_path(
client, identifier, active_project.name, context
)
# Validate identifier to prevent path traversal attacks
# For memory:// URLs, validate the extracted path (not the raw URL which
@@ -235,10 +246,14 @@ async def read_note(
# Why: search_notes applies the same memory:// normalization and tool-level
# query handling as the rest of MCP routing, which raw client calls skip.
# Outcome: unresolved memory URLs still fall back through normalized search.
# Pass project_id (external_id UUID) so the workspace selection from the
# outer get_project_client() is preserved across the inner re-resolution.
# Without this, project names that collide across workspaces could re-resolve
# to a different tenant via the default-workspace fallback (CLI/context=None).
search_type = "title" if title_only else "text"
response = await search_notes(
project=active_project.name,
workspace=workspace,
project_id=active_project.external_id,
query=identifier_text,
search_type=search_type,
output_format="json",
+15 -5
View File
@@ -62,7 +62,7 @@ async def recent_activity(
Field(default=10, validation_alias=AliasChoices("page_size", "limit", "per_page")),
] = 10,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | list[dict]:
@@ -104,6 +104,9 @@ 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.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
output_format: "text" returns human-readable summary text. "json" returns
a flat list of recent items.
context: Optional FastMCP context for performance caching.
@@ -179,9 +182,14 @@ async def recent_activity(
if "type" not in params:
params["type"] = [SearchItemType.ENTITY.value]
# Resolve project parameter using the three-tier hierarchy
# allow_discovery=True enables Discovery Mode, so a project is not required
resolved_project = await resolve_project_parameter(project, allow_discovery=True)
# Resolve project parameter using the three-tier hierarchy.
# allow_discovery=True enables Discovery Mode, so a project is not required.
# project_id (UUID) takes precedence over project name — without this fallback,
# callers passing only project_id would fall into Discovery Mode.
effective_identifier = project_id if project_id else project
resolved_project = await resolve_project_parameter(
effective_identifier, allow_discovery=True
)
if resolved_project is None:
# Discovery Mode: Get activity across all projects
@@ -296,7 +304,9 @@ async def recent_activity(
f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}"
)
async with get_project_client(resolved_project, workspace, context) as (
async with get_project_client(
resolved_project, context=context, project_id=project_id
) as (
client,
active_project,
):
+24 -6
View File
@@ -211,7 +211,7 @@ async def schema_validate(
note_type: Optional[str] = None,
identifier: Optional[str] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> ValidationReport | str | dict:
@@ -236,6 +236,9 @@ async def schema_validate(
identifier: Specific note to validate (permalink, title, or path).
If provided, validates only this note.
project: Project name. Optional -- server will resolve.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -251,7 +254,10 @@ async def schema_validate(
# Validate in a specific project
schema_validate(note_type="person", project="my-research")
"""
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=schema_validate project={active_project.name} "
f"note_type={note_type} identifier={identifier}"
@@ -318,7 +324,7 @@ async def schema_infer(
note_type: str,
threshold: float = 0.25,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | dict:
@@ -342,6 +348,9 @@ async def schema_infer(
threshold: Minimum frequency (0-1) for a field to be suggested as optional.
Default 0.25 (25%). Fields above 95% become required.
project: Project name. Optional -- server will resolve.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -357,7 +366,10 @@ async def schema_infer(
# Infer in a specific project
schema_infer("person", project="my-research")
"""
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=schema_infer project={active_project.name} "
f"note_type={note_type} threshold={threshold}"
@@ -432,7 +444,7 @@ async def schema_infer(
async def schema_diff(
note_type: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> str | dict:
@@ -453,6 +465,9 @@ async def schema_diff(
Args:
note_type: The note type to check for drift (e.g., "person").
project: Project name. Optional -- server will resolve.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -466,7 +481,10 @@ async def schema_diff(
# Check drift in a specific project
schema_diff("person", project="my-research")
"""
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=schema_diff project={active_project.name} note_type={note_type}"
)
+14 -8
View File
@@ -308,7 +308,7 @@ async def search_notes(
Field(default=None, validation_alias=AliasChoices("query", "q", "search", "text")),
] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
# `offset` is intentionally NOT aliased to `page`: offset is item-indexed
# (skip N items) while page is 1-indexed page-number. Direct aliasing would
# silently return the wrong slice.
@@ -359,9 +359,7 @@ async def search_notes(
Optional[float],
Field(
default=None,
validation_alias=AliasChoices(
"min_similarity", "threshold", "similarity_threshold"
),
validation_alias=AliasChoices("min_similarity", "threshold", "similarity_threshold"),
),
] = None,
context: Context | None = None,
@@ -447,6 +445,9 @@ async def search_notes(
Omit or pass None for filter-only searches using metadata_filters, tags, or status.
project: Project name to search in. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
page: The page number of results to return (default 1)
page_size: The number of results to return per page (default 10)
search_type: Type of search to perform, one of:
@@ -561,7 +562,7 @@ async def search_notes(
entrypoint="mcp",
tool_name="search_notes",
requested_project=project,
workspace_id=workspace,
requested_project_id=project_id,
search_type=search_type or "default",
output_format=output_format,
page=page,
@@ -575,12 +576,17 @@ async def search_notes(
has_tags_filter=bool(tags),
has_status_filter=bool(status),
):
async with get_project_client(project, workspace, context) as (client, active_project):
# Handle memory:// URLs by resolving to permalink search
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
# Handle memory:// URLs by resolving to permalink search.
# Use active_project.name so resolution hits the cached active project
# when project_id was used or `project` was wrong/ambiguous.
is_memory_url = False
if query is not None:
_, resolved_query, is_memory_url = await resolve_project_and_path(
client, query, project, context
client, query, active_project.name, context
)
if is_memory_url:
query = resolved_query
+4
View File
@@ -25,6 +25,7 @@ def _text_block(message: str) -> List[ContentBlock]:
async def search_notes_ui(
query: str,
project: Optional[str] = None,
project_id: Optional[str] = None,
page: int = 1,
page_size: int = 10,
search_type: Optional[str] = None,
@@ -49,6 +50,7 @@ async def search_notes_ui(
result = await search_notes(
query=query,
project=project,
project_id=project_id,
page=page,
page_size=page_size,
search_type=search_type,
@@ -97,12 +99,14 @@ async def search_notes_ui(
async def read_note_ui(
identifier: str,
project: Optional[str] = None,
project_id: Optional[str] = None,
context: Context | None = None,
) -> List[ContentBlock]:
"""Return a note preview UI as an embedded MCP-UI resource."""
content = await read_note(
identifier=identifier,
project=project,
project_id=project_id,
output_format="text",
context=context,
)
+5 -2
View File
@@ -17,7 +17,7 @@ from basic_memory.mcp.tools.read_note import read_note
async def view_note(
identifier: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
context: Context | None = None,
) -> str:
"""View a markdown note as a formatted artifact.
@@ -30,6 +30,9 @@ async def view_note(
identifier: The title or permalink of the note to view
project: Project name to read from. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
context: Optional FastMCP context for performance caching.
Returns:
@@ -56,7 +59,7 @@ async def view_note(
await read_note(
identifier=identifier,
project=project,
workspace=workspace,
project_id=project_id,
context=context,
)
)
+9 -3
View File
@@ -31,7 +31,7 @@ async def write_note(
Field(validation_alias=AliasChoices("directory", "folder", "dir", "path")),
],
project: Optional[str] = None,
workspace: Optional[str] = None,
project_id: Optional[str] = None,
tags: list[str] | str | None = None,
note_type: str = "note",
metadata: Annotated[dict | None, BeforeValidator(coerce_dict)] = None,
@@ -82,6 +82,9 @@ async def write_note(
project: Project name to write to. Optional - server will resolve using the
hierarchy above. If unknown, use list_memory_projects() to discover
available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known —
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
tags: Tags to categorize the note. Can be a list of strings, a comma-separated string, or None.
Note: If passing from external MCP clients, use a string format (e.g. "tag1,tag2,tag3")
note_type: Type of note to create (stored in frontmatter). Defaults to "note".
@@ -162,12 +165,15 @@ async def write_note(
entrypoint="mcp",
tool_name="write_note",
requested_project=project,
workspace_id=workspace,
requested_project_id=project_id,
note_type=note_type,
overwrite=effective_overwrite,
output_format=output_format,
):
async with get_project_client(project, workspace, context) as (client, active_project):
async with get_project_client(project, context=context, project_id=project_id) as (
client,
active_project,
):
logger.info(
f"MCP tool call tool=write_note project={active_project.name} directory={directory}, title={title}, tags={tags}"
)
@@ -107,8 +107,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
await self._ensure_vector_tables()
except SemanticDependenciesMissingError as exc:
logger.warning(
f"Semantic search disabled: {exc}. "
"Falling back to keyword-only search."
f"Semantic search disabled: {exc}. Falling back to keyword-only search."
)
self._semantic_enabled = False
+11 -1
View File
@@ -489,7 +489,17 @@ async def test_aliases_not_advertised_in_schema(mcp_server, app):
),
"search_notes": (
["query", "page", "page_size", "note_types", "after_date", "min_similarity"],
["q", "search", "offset", "limit", "note_type", "types", "since", "after", "threshold"],
[
"q",
"search",
"offset",
"limit",
"note_type",
"types",
"since",
"after",
"threshold",
],
),
"recent_activity": (
["type", "timeframe", "page", "page_size"],
@@ -100,3 +100,71 @@ async def test_read_note_underscored_folder_by_permalink(mcp_server, app, test_p
assert "# Example Note" in result_text
assert "This is a test note in an underscored folder." in result_text
assert f"{test_project.name}/archive/articles/example-note" in result_text # permalink
@pytest.mark.asyncio
async def test_read_note_by_project_id(mcp_server, app, test_project):
"""Read a note by passing project_id (UUID) instead of project name.
Verifies the project_id parameter routes through get_project_client correctly
in pure local mode (no cloud creds), where get_project_mode() would otherwise
default unknown identifiers to CLOUD and break routing.
"""
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "By ID Note",
"directory": "test",
"content": "# By ID Note\n\nLooked up by external_id.",
},
)
# Read by external_id (UUID) instead of project name
read_result = await client.call_tool(
"read_note",
{
"project_id": test_project.external_id,
"identifier": "By ID Note",
},
)
assert len(read_result.content) == 1
assert read_result.content[0].type == "text"
result_text = read_result.content[0].text
assert "# By ID Note" in result_text
assert "Looked up by external_id." in result_text
@pytest.mark.asyncio
async def test_read_note_project_id_takes_precedence_over_name(mcp_server, app, test_project):
"""When project_id is passed alongside a wrong project name, project_id wins."""
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Precedence Note",
"directory": "test",
"content": "# Precedence Note\n\nproject_id wins.",
},
)
# Pass an obviously-wrong project name alongside the correct project_id.
# If project_id takes precedence (as documented), the read still succeeds.
read_result = await client.call_tool(
"read_note",
{
"project": "this-project-does-not-exist",
"project_id": test_project.external_id,
"identifier": "Precedence Note",
},
)
assert len(read_result.content) == 1
result_text = read_result.content[0].text
assert "# Precedence Note" in result_text
assert "project_id wins." in result_text
+32 -16
View File
@@ -113,6 +113,38 @@ def test_write_note_json_output(mock_mcp_write):
assert mock_mcp_write.call_args.kwargs["output_format"] == "json"
@patch(
"basic_memory.cli.commands.tool.mcp_write_note",
new_callable=AsyncMock,
return_value=WRITE_NOTE_RESULT,
)
def test_write_note_project_id_passthrough(mock_mcp_write):
"""--project-id forwards to the MCP tool's project_id parameter.
Regression: removing --workspace without exposing --project-id left CLI
callers unable to disambiguate same-named projects across cloud workspaces.
"""
uuid = "11111111-1111-1111-1111-111111111111"
result = runner.invoke(
cli_app,
[
"tool",
"write-note",
"--title",
"Test Note",
"--folder",
"notes",
"--content",
"hello",
"--project-id",
uuid,
],
)
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert mock_mcp_write.call_args.kwargs["project_id"] == uuid
@patch(
"basic_memory.cli.commands.tool.mcp_write_note",
new_callable=AsyncMock,
@@ -167,22 +199,6 @@ def test_read_note_json_output(mock_mcp_read):
assert mock_mcp_read.call_args.kwargs["output_format"] == "json"
@patch(
"basic_memory.cli.commands.tool.mcp_read_note",
new_callable=AsyncMock,
return_value=READ_NOTE_RESULT,
)
def test_read_note_workspace_passthrough(mock_mcp_read):
"""read-note --workspace passes workspace through to the MCP tool call."""
result = runner.invoke(
cli_app,
["tool", "read-note", "test-note", "--workspace", "tenant-123"],
)
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert mock_mcp_read.call_args.kwargs["workspace"] == "tenant-123"
@patch(
"basic_memory.cli.commands.tool.mcp_read_note",
new_callable=AsyncMock,
+1 -2
View File
@@ -189,8 +189,7 @@ def test_observation_skips_obsidian_callouts():
t.meta.get("observation")
for t in tokens
if t.type == "inline" and t.meta and t.meta.get("observation")
if t.meta["observation"]["category"]
and t.meta["observation"]["category"].startswith("!")
if t.meta["observation"]["category"] and t.meta["observation"]["category"].startswith("!")
]
assert callout_observations == [], (
f"Obsidian callouts should not produce observations, got: {callout_observations}"
+178 -20
View File
@@ -606,8 +606,10 @@ async def test_resolve_workspace_project_identifier_handles_qualified_and_collis
assert resolved.workspace.slug == "acme"
assert resolved.project.external_id == "acme-project-id"
with pytest.raises(ValueError, match="Use: personal/meeting-notes or acme/meeting-notes"):
await resolve_workspace_project_identifier("meeting-notes")
# Ambiguous name resolves to the default workspace (personal)
resolved = await resolve_workspace_project_identifier("meeting-notes")
assert resolved.workspace.slug == "personal"
assert resolved.project.external_id == "personal-project-id"
@pytest.mark.asyncio
@@ -663,6 +665,180 @@ async def test_resolve_workspace_project_identifier_uses_active_workspace_for_du
assert resolved.project.external_id == "acme-project-id"
@pytest.mark.asyncio
async def test_resolve_workspace_project_identifier_resolves_by_external_id(monkeypatch):
"""Direct lookup by external_id (UUID) bypasses name/permalink resolution."""
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_project_identifier,
)
personal = _workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
acme = _workspace(
tenant_id="acme-tenant",
workspace_type="organization",
slug="acme",
name="Acme",
role="editor",
)
# Same project name in two workspaces — UUID lookup must pick the right one
# without falling back to default-workspace ambiguity handling.
entries = (
WorkspaceProjectEntry(
workspace=personal,
project=_project(
"Meeting Notes", id=1, external_id="11111111-1111-1111-1111-111111111111"
),
),
WorkspaceProjectEntry(
workspace=acme,
project=_project(
"Meeting Notes", id=2, external_id="22222222-2222-2222-2222-222222222222"
),
),
)
index = _build_workspace_project_index((personal, acme), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
resolved = await resolve_workspace_project_identifier(
"22222222-2222-2222-2222-222222222222"
)
assert resolved.workspace.slug == "acme"
assert resolved.project.external_id == "22222222-2222-2222-2222-222222222222"
@pytest.mark.asyncio
async def test_resolve_workspace_project_identifier_normalizes_uuid_forms(monkeypatch):
"""Uppercase, brace-wrapped, and urn:uuid forms canonicalize before lookup."""
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
resolve_workspace_project_identifier,
)
workspace = _workspace(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
canonical_uuid = "33333333-3333-3333-3333-333333333333"
entries = (
WorkspaceProjectEntry(
workspace=workspace,
project=_project("Meeting Notes", id=1, external_id=canonical_uuid),
),
)
index = _build_workspace_project_index((workspace,), entries)
async def fake_index(context=None):
return index
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
# Each variant should canonicalize to the same lowercase-hyphenated form
for variant in (
canonical_uuid.upper(),
f"{{{canonical_uuid}}}",
f"urn:uuid:{canonical_uuid}",
canonical_uuid.replace("-", ""),
):
resolved = await resolve_workspace_project_identifier(variant)
assert resolved.project.external_id == canonical_uuid, (
f"variant {variant!r} did not resolve to canonical UUID"
)
@pytest.mark.asyncio
async def test_get_project_client_with_project_id_routes_locally_without_cloud(
config_manager, monkeypatch
):
"""A UUID project_id in pure local mode must not trigger cloud routing.
Regression: get_project_mode() defaults unknown identifiers to CLOUD because the
local config keys by name. UUIDs are never registered locally, so without the
fix below, passing project_id would falsely route through the cloud client and
error out for users with no cloud credentials.
"""
import basic_memory.mcp.project_context as project_context
# Pure local mode: no cloud creds, no factory mode, no explicit cloud routing
config = config_manager.load_config()
config.cloud_api_key = None
config_manager.save_config(config)
monkeypatch.setattr(project_context, "has_cloud_credentials", lambda _config: False)
captured: dict[str, object] = {}
@asynccontextmanager
async def fake_get_client(**kwargs) -> AsyncIterator[object]:
captured["get_client_kwargs"] = kwargs
yield object()
async def fake_get_active_project(client, project, context, headers=None):
captured["validated_project"] = project
return _project("Local Project", id=99, external_id=project)
monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
monkeypatch.setattr(project_context, "get_active_project", fake_get_active_project)
canonical_uuid = "55555555-5555-5555-5555-555555555555"
async with project_context.get_project_client(project_id=canonical_uuid) as (_, active):
assert active.external_id == canonical_uuid
# When project_id is set, get_client must be called WITHOUT project_name so the
# ASGI fallback is selected (not the per-project cloud-by-default path).
assert captured["get_client_kwargs"] == {}
assert captured["validated_project"] == canonical_uuid
@pytest.mark.asyncio
async def test_get_project_client_prefers_project_id_over_project_name(monkeypatch):
"""When both project and project_id are passed, the UUID takes precedence."""
import basic_memory.mcp.project_context as project_context
# Capture which identifier flows into resolution, then short-circuit before
# the rest of the routing chain runs (avoids real network calls).
captured: dict[str, str | None] = {}
sentinel = RuntimeError("stop after resolution")
async def fake_resolve(project=None, *, allow_discovery=True, context=None):
captured["project"] = project
raise sentinel
monkeypatch.setattr(project_context, "resolve_project_parameter", fake_resolve)
canonical_uuid = "44444444-4444-4444-4444-444444444444"
with pytest.raises(RuntimeError, match="stop after resolution"):
async with project_context.get_project_client(
project="ambiguous-name",
project_id=canonical_uuid,
):
pass
assert captured["project"] == canonical_uuid
@pytest.mark.asyncio
async def test_resolve_project_parameter_uses_cached_active_project_before_api_default_lookup(
config_manager, monkeypatch
@@ -828,24 +1004,6 @@ async def test_resolve_project_and_path_uses_cached_project_for_memory_url_prefi
assert is_memory_url is True
@pytest.mark.asyncio
async def test_get_project_client_rejects_workspace_for_local_project(config_manager):
from basic_memory.mcp.project_context import get_project_client
from basic_memory.config import ProjectEntry
# Register "main" as a LOCAL project so get_project_mode returns LOCAL
config = config_manager.load_config()
(config_manager.config_dir.parent / "main").mkdir(parents=True, exist_ok=True)
config.projects["main"] = ProjectEntry(path=str(config_manager.config_dir.parent / "main"))
config_manager.save_config(config)
with pytest.raises(
ValueError, match="Workspace 'tenant-123' cannot be used with local project"
):
async with get_project_client(project="main", workspace="tenant-123"):
pass
class TestDetectProjectFromUrlPrefix:
"""Test detect_project_from_url_prefix for URL-based project detection."""
+16 -16
View File
@@ -13,7 +13,7 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"build_context": [
"url",
"project",
"workspace",
"project_id",
"depth",
"timeframe",
"page",
@@ -21,25 +21,25 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"max_related",
"output_format",
],
"canvas": ["nodes", "edges", "title", "directory", "project", "workspace"],
"canvas": ["nodes", "edges", "title", "directory", "project", "project_id"],
"cloud_info": [],
"create_memory_project": ["project_name", "project_path", "set_default", "output_format"],
"delete_note": ["identifier", "is_directory", "project", "workspace", "output_format"],
"delete_note": ["identifier", "is_directory", "project", "project_id", "output_format"],
"delete_project": ["project_name"],
"edit_note": [
"identifier",
"operation",
"content",
"project",
"workspace",
"project_id",
"section",
"find_text",
"expected_replacements",
"output_format",
],
"fetch": ["id"],
"list_directory": ["dir_name", "depth", "file_name_glob", "project", "workspace"],
"list_memory_projects": ["output_format", "workspace"],
"list_directory": ["dir_name", "depth", "file_name_glob", "project", "project_id"],
"list_memory_projects": ["output_format"],
"list_workspaces": ["output_format"],
"move_note": [
"identifier",
@@ -47,14 +47,14 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"destination_folder",
"is_directory",
"project",
"workspace",
"project_id",
"output_format",
],
"read_content": ["path", "project", "workspace"],
"read_content": ["path", "project", "project_id"],
"read_note": [
"identifier",
"project",
"workspace",
"project_id",
"output_format",
"include_frontmatter",
],
@@ -66,17 +66,17 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"page",
"page_size",
"project",
"workspace",
"project_id",
"output_format",
],
"schema_diff": ["note_type", "project", "workspace", "output_format"],
"schema_infer": ["note_type", "threshold", "project", "workspace", "output_format"],
"schema_validate": ["note_type", "identifier", "project", "workspace", "output_format"],
"schema_diff": ["note_type", "project", "project_id", "output_format"],
"schema_infer": ["note_type", "threshold", "project", "project_id", "output_format"],
"schema_validate": ["note_type", "identifier", "project", "project_id", "output_format"],
"search": ["query"],
"search_notes": [
"query",
"project",
"workspace",
"project_id",
"page",
"page_size",
"search_type",
@@ -89,13 +89,13 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"status",
"min_similarity",
],
"view_note": ["identifier", "project", "workspace"],
"view_note": ["identifier", "project", "project_id"],
"write_note": [
"title",
"content",
"directory",
"project",
"workspace",
"project_id",
"tags",
"note_type",
"metadata",
+21
View File
@@ -215,6 +215,9 @@ async def test_list_and_create_project_text_and_json_modes(app, test_project, tm
)
assert isinstance(create_text, str)
assert "mode-create-project" in create_text
# external_id should appear in the human-readable output too, so users see
# the UUID without re-listing projects.
assert "External ID:" in create_text
create_json_again = await create_memory_project(
project_name=project_name,
@@ -227,6 +230,24 @@ async def test_list_and_create_project_text_and_json_modes(app, test_project, tm
assert Path(create_json_again["path"]) == Path(project_path)
assert create_json_again["created"] is False
assert create_json_again["already_exists"] is True
# external_id (UUID) must be present so callers can immediately use it as
# project_id in subsequent tool calls without a list_memory_projects() round-trip.
assert isinstance(create_json_again["external_id"], str)
assert len(create_json_again["external_id"]) > 0
# Verify create-new JSON path also returns external_id (not just already-exists).
new_project_name = "mode-create-fresh"
new_project_path = str(tmp_path.parent / (tmp_path.name + "-projects") / "mode-create-fresh")
create_json_new = await create_memory_project(
project_name=new_project_name,
project_path=new_project_path,
output_format="json",
)
assert isinstance(create_json_new, dict)
assert create_json_new["created"] is True
assert create_json_new["already_exists"] is False
assert isinstance(create_json_new["external_id"], str)
assert len(create_json_new["external_id"]) > 0
default_project_name = "mode-default-project"
default_project_path = str(
+1 -170
View File
@@ -218,35 +218,6 @@ async def test_list_memory_projects_cloud_failure_graceful(app, test_project):
assert f"- {test_project.name} (local)" in result
@pytest.mark.asyncio
async def test_list_memory_projects_explicit_workspace_discovery_failure_graceful(
app,
test_project,
):
"""Explicit workspace discovery failure still returns the local project list."""
with (
patch(
"basic_memory.mcp.tools.project_management.has_cloud_credentials",
return_value=True,
),
patch(
"basic_memory.mcp.tools.project_management.resolve_workspace_parameter",
new_callable=AsyncMock,
side_effect=RuntimeError("workspace discovery unavailable"),
),
patch(
"basic_memory.mcp.tools.project_management._fetch_cloud_projects",
new_callable=AsyncMock,
return_value=None,
) as mock_fetch,
):
result = await list_memory_projects(workspace="tenant-abc")
mock_fetch.assert_awaited_once_with("tenant-abc", None)
assert "Available projects:" in result
assert f"- {test_project.name} (local)" in result
@pytest.mark.asyncio
async def test_list_memory_projects_factory_mode(app, test_project):
"""In factory mode (cloud app), projects are reported as cloud-sourced with workspace metadata."""
@@ -345,40 +316,6 @@ async def test_list_memory_projects_factory_mode_workspace_lookup_failure(app, t
assert "- cloud-proj (cloud)" in result
@pytest.mark.asyncio
async def test_list_memory_projects_factory_mode_explicit_workspace(app, test_project):
"""In factory mode, explicit workspace param selects the matching workspace."""
factory_project = _make_project("cloud-proj", "/cloud-proj", is_default=True)
factory_list = _make_list([factory_project], default="cloud-proj")
personal_ws = _make_workspace("tenant-personal", "Personal", "personal")
org_ws = _make_workspace("tenant-org", "Acme Corp", "organization")
with (
patch(
"basic_memory.mcp.tools.project_management.is_factory_mode",
return_value=True,
),
patch(
"basic_memory.mcp.clients.project.ProjectClient.list_projects",
new_callable=AsyncMock,
return_value=factory_list,
),
patch(
"basic_memory.mcp.project_context.get_available_workspaces",
new_callable=AsyncMock,
return_value=[personal_ws, org_ws],
),
):
result = await list_memory_projects(output_format="json", workspace="tenant-org")
assert isinstance(result, dict)
proj = result["projects"][0]
assert proj["workspace_name"] == "Acme Corp"
assert proj["workspace_type"] == "organization"
assert proj["workspace_tenant_id"] == "tenant-org"
@pytest.mark.asyncio
async def test_list_memory_projects_json_with_cloud(app, test_project):
"""JSON output includes local_path, cloud_path, and source fields."""
@@ -553,32 +490,6 @@ def _make_workspace_index(workspace_projects):
return _build_workspace_project_index(workspaces, entries)
@pytest.mark.asyncio
async def test_list_memory_projects_passes_explicit_workspace(app, test_project):
"""Explicit workspace param is forwarded to _fetch_cloud_projects."""
cloud_list = _make_list([_make_project("cloud-proj", "/cloud-proj")])
with (
patch(
"basic_memory.mcp.tools.project_management.has_cloud_credentials",
return_value=True,
),
patch(
"basic_memory.mcp.tools.project_management._fetch_cloud_projects",
new_callable=AsyncMock,
return_value=cloud_list,
) as mock_fetch,
patch(
"basic_memory.mcp.project_context.get_available_workspaces",
new_callable=AsyncMock,
return_value=[_make_workspace("my-org-tenant-id", "My Org", "organization")],
),
):
await list_memory_projects(workspace="my-org-tenant-id")
mock_fetch.assert_awaited_once_with("my-org-tenant-id", None)
@pytest.mark.asyncio
async def test_list_memory_projects_aggregates_without_config_workspace(app, test_project):
"""When no explicit workspace is given, cloud discovery fans out across workspaces."""
@@ -608,84 +519,4 @@ async def test_list_memory_projects_aggregates_without_config_workspace(app, tes
result = await list_memory_projects()
mock_index.assert_awaited_once()
assert "- cloud-proj (cloud) [default/cloud-proj]" in result
@pytest.mark.asyncio
async def test_list_memory_projects_explicit_workspace_overrides_config(app, test_project):
"""Explicit workspace takes precedence over config.default_workspace."""
cloud_list = _make_list([_make_project("cloud-proj", "/cloud-proj")])
with (
patch("basic_memory.mcp.tools.project_management.ConfigManager") as mock_cm_cls,
patch(
"basic_memory.mcp.tools.project_management.has_cloud_credentials",
return_value=True,
),
patch(
"basic_memory.mcp.tools.project_management._fetch_cloud_projects",
new_callable=AsyncMock,
return_value=cloud_list,
) as mock_fetch,
patch(
"basic_memory.mcp.project_context.get_available_workspaces",
new_callable=AsyncMock,
return_value=[_make_workspace("explicit-ws", "Explicit WS", "organization")],
),
):
mock_config = mock_cm_cls.return_value.config
mock_config.default_workspace = "config-default-ws"
await list_memory_projects(workspace="explicit-ws")
# Explicit workspace wins over config default
mock_fetch.assert_awaited_once_with("explicit-ws", None)
@pytest.mark.asyncio
async def test_list_memory_projects_json_includes_workspace_info(app, test_project):
"""JSON output includes workspace_name, workspace_type, workspace_tenant_id for cloud projects."""
local_proj = _make_project("local-only", "/local/path", is_default=True)
local_list = _make_list([local_proj], default="local-only")
cloud_proj = _make_project("cloud-proj", "/cloud/path", id=10, external_id="cloud-uuid")
cloud_list = _make_list([cloud_proj])
ws = _make_workspace("org-tenant-abc", "Acme Corp", "organization")
with (
patch(
"basic_memory.mcp.clients.project.ProjectClient.list_projects",
new_callable=AsyncMock,
return_value=local_list,
),
patch(
"basic_memory.mcp.tools.project_management.has_cloud_credentials",
return_value=True,
),
patch(
"basic_memory.mcp.tools.project_management._fetch_cloud_projects",
new_callable=AsyncMock,
return_value=cloud_list,
),
patch(
"basic_memory.mcp.project_context.get_available_workspaces",
new_callable=AsyncMock,
return_value=[ws],
),
):
result = await list_memory_projects(output_format="json", workspace="org-tenant-abc")
assert isinstance(result, dict)
by_name = {p["name"]: p for p in result["projects"]}
# Cloud project carries workspace info
cloud = by_name["cloud-proj"]
assert cloud["workspace_name"] == "Acme Corp"
assert cloud["workspace_type"] == "organization"
assert cloud["workspace_tenant_id"] == "org-tenant-abc"
# Local-only project has no workspace info
local = by_name["local-only"]
assert local["workspace_name"] is None
assert local["workspace_type"] is None
assert local["workspace_tenant_id"] is None
assert "- cloud-proj (cloud) [00000000-0000-0000-0000-000000000001]" in result
+5 -5
View File
@@ -63,7 +63,7 @@ async def test_write_note_emits_root_operation_and_project_context(
"entrypoint": "mcp",
"tool_name": "write_note",
"requested_project": test_project.name,
"workspace_id": None,
"requested_project_id": None,
"note_type": "note",
"overwrite": False,
"output_format": "json",
@@ -108,7 +108,7 @@ async def test_read_note_emits_root_operation_and_project_context(
"entrypoint": "mcp",
"tool_name": "read_note",
"requested_project": test_project.name,
"workspace_id": None,
"requested_project_id": None,
"output_format": "json",
"include_frontmatter": True,
},
@@ -156,7 +156,7 @@ async def test_search_notes_emits_root_operation_and_project_context(
"entrypoint": "mcp",
"tool_name": "search_notes",
"requested_project": test_project.name,
"workspace_id": None,
"requested_project_id": None,
"search_type": "text",
"output_format": "json",
"page": 1,
@@ -209,7 +209,7 @@ async def test_edit_note_emits_root_operation_and_project_context(
"entrypoint": "mcp",
"tool_name": "edit_note",
"requested_project": test_project.name,
"workspace_id": None,
"requested_project_id": None,
"edit_operation": "append",
"output_format": "json",
"has_section": False,
@@ -261,7 +261,7 @@ async def test_build_context_emits_root_operation_and_project_context(
"entrypoint": "mcp",
"tool_name": "build_context",
"requested_project": test_project.name,
"workspace_id": None,
"requested_project_id": None,
"depth": 2,
"timeframe": "7d",
"page": 1,
+2 -5
View File
@@ -259,9 +259,7 @@ async def test_delete_entity_without_permalink(search_service, sample_entity):
@pytest.mark.asyncio
async def test_handle_delete_clears_entity_vectors(
search_service, sample_entity, monkeypatch
):
async def test_handle_delete_clears_entity_vectors(search_service, sample_entity, monkeypatch):
"""Regression guard for #764: handle_delete must drive vector-row cleanup
so deleting an entity doesn't leave orphaned rows in `search_vector_chunks`
or `search_vector_embeddings`.
@@ -288,8 +286,7 @@ async def test_handle_delete_clears_entity_vectors(
await search_service.handle_delete(sample_entity)
assert calls == [sample_entity.id], (
f"handle_delete must call delete_entity_vector_rows({sample_entity.id}); "
f"got calls={calls}"
f"handle_delete must call delete_entity_vector_rows({sample_entity.id}); got calls={calls}"
)