fix(mcp): centralize workspace permalink routing (#808)

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/mcp/async_client.py

@@ -37,8 +37,15 @@ def _build_timeout() -> Timeout:
 
 def _asgi_client(timeout: Timeout) -> AsyncClient:
     """Create a local ASGI client."""
+    from basic_memory.workspace_context import workspace_permalink_headers
+
     return AsyncClient(
-        transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout
+        transport=ASGITransport(app=fastapi_app),
+        base_url="http://test",
+        timeout=timeout,
+        # Local ASGI calls still cross the HTTP boundary, so request handlers need
+        # the same workspace permalink metadata that cloud proxy calls receive.
+        headers=workspace_permalink_headers(),
     )
 
 

src/basic_memory/mcp/project_context.py

@@ -29,7 +29,11 @@
 from basic_memory.schemas.project_info import ProjectItem, ProjectList
 from basic_memory.schemas.v2 import ProjectResolveResponse
 from basic_memory.schemas.memory import memory_url_path
-from basic_memory.utils import generate_permalink, normalize_project_reference
+from basic_memory.utils import (
+    build_qualified_permalink_reference,
+    generate_permalink,
+    normalize_project_reference,
+)
 from basic_memory.workspace_context import (
     current_workspace_permalink_context,
     workspace_permalink_context,
@@ -417,14 +421,20 @@ def _unqualified_project_identifier(identifier: str) -> str:
     return project_identifier
 
 
-def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str] | None:
-    """Split ``memory://<workspace>/<project>/<path>`` into route segments."""
-    if not identifier.strip().startswith("memory://"):
-        return None
+def _identifier_path(identifier: str) -> str:
+    """Return the routable path portion of a raw identifier or memory URL."""
+    stripped = identifier.strip()
+    return memory_url_path(stripped) if stripped.startswith("memory://") else stripped
 
-    normalized = normalize_project_reference(memory_url_path(identifier))
+
+def _split_workspace_identifier_segments(identifier: str) -> tuple[str, str, str] | None:
+    """Split ``<workspace>/<project>/<path>`` identifiers into route segments."""
+    normalized = normalize_project_reference(_identifier_path(identifier)).strip("/")
     parts = normalized.split("/", 2)
     if len(parts) != 3:
+        # Trigger: two-segment identifiers such as `workspace/project` or `project/path`.
+        # Why: without a remainder, the shape is ambiguous with existing project-prefix routing.
+        # Outcome: fall through so the normal project-prefix/default-project resolver decides.
         return None
 
     workspace_slug, project_identifier, remainder = parts
@@ -433,6 +443,14 @@ def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str
     return workspace_slug, project_identifier, remainder
 
 
+def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str] | None:
+    """Split ``memory://<workspace>/<project>/<path>`` into route segments."""
+    if not identifier.strip().startswith("memory://"):
+        return None
+
+    return _split_workspace_identifier_segments(identifier)
+
+
 def _canonical_memory_path_for_workspace(
     *,
     workspace_slug: str,
@@ -448,10 +466,14 @@ def _canonical_memory_path_for_workspace(
     # Trigger: a caller supplied a workspace-qualified memory URL.
     # Why: the first two path segments are the global route, even for Personal.
     # Outcome: lookups preserve the complete workspace/project canonical permalink.
-    prefix = f"{generate_permalink(workspace_slug)}/{project_permalink}"
     if not normalized_remainder:
-        return prefix
-    return f"{prefix}/{normalized_remainder}"
+        normalized_remainder = project_permalink
+    return build_qualified_permalink_reference(
+        project_permalink,
+        normalized_remainder,
+        include_project=True,
+        workspace_permalink=workspace_slug,
+    )
 
 
 def _canonical_memory_path_for_active_route(
@@ -528,6 +550,27 @@ async def resolve_workspace_qualified_memory_url(
     if segments is None:
         return None
 
+    return await _resolve_workspace_segments(identifier, segments, context=context)
+
+
+async def resolve_workspace_qualified_identifier(
+    identifier: str,
+    context: Optional[Context] = None,
+) -> WorkspaceMemoryUrlResolution | None:
+    """Resolve a workspace-qualified permalink or memory URL against accessible workspaces."""
+    segments = _split_workspace_identifier_segments(identifier)
+    if segments is None:
+        return None
+
+    return await _resolve_workspace_segments(identifier, segments, context=context)
+
+
+async def _resolve_workspace_segments(
+    identifier: str,
+    segments: tuple[str, str, str],
+    context: Optional[Context] = None,
+) -> WorkspaceMemoryUrlResolution | None:
+    """Resolve parsed workspace/project/path segments against accessible workspaces."""
     workspace_slug, project_identifier, remainder = segments
     index = await _ensure_workspace_project_index(context=context)
     workspace = next(
@@ -1278,17 +1321,51 @@ async def detect_project_from_memory_url_prefix(
     if not identifier.strip().startswith("memory://"):
         return None
 
+    return await detect_project_from_identifier_prefix(identifier, config, context=context)
+
+
+async def detect_project_from_identifier_prefix(
+    identifier: str,
+    config: BasicMemoryConfig,
+    context: Optional[Context] = None,
+) -> Optional[str]:
+    """Resolve a project from a plain permalink, memory URL, or workspace route prefix."""
     local_project = detect_project_from_url_prefix(identifier, config)
     if local_project is not None:
         return local_project
 
+    normalized_identifier = normalize_project_reference(_identifier_path(identifier)).strip("/")
+    if "/" not in normalized_identifier:
+        # Trigger: plain text search query or single-segment title/permalink.
+        # Why: cloud project discovery can build a workspace index; only path-shaped
+        #   identifiers carry enough structure to justify that cost.
+        # Outcome: keep unqualified search/title input on the active/default project route.
+        return None
+
     if _cloud_workspace_discovery_available(config):
-        resolution = await resolve_workspace_qualified_memory_url(
+        workspace_resolution = await resolve_workspace_qualified_identifier(
             identifier,
             context=context,
         )
-        if resolution is not None:
-            return resolution.project_identifier
+        if workspace_resolution is not None:
+            return workspace_resolution.project_identifier
+
+        project_prefix, _ = _split_project_prefix(normalized_identifier)
+        if project_prefix is None:
+            return None
+
+        try:
+            project_resolution = await resolve_workspace_project_identifier(
+                project_prefix,
+                context=context,
+            )
+        except ValueError as exc:
+            message = str(exc).lower()
+            if "not found" in message or "no accessible workspaces" in message:
+                return None
+            raise
+
+        return project_resolution.qualified_name
 
     return None
 

src/basic_memory/mcp/tools/read_note.py

@@ -11,15 +11,14 @@
 
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import (
-    detect_project_from_memory_url_prefix,
+    detect_project_from_identifier_prefix,
     get_project_client,
     resolve_project_and_path,
 )
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.search import search_notes
 from basic_memory.schemas.memory import memory_url_path
-from basic_memory.utils import generate_permalink, validate_project_path
-from basic_memory.workspace_context import current_workspace_permalink_context
+from basic_memory.utils import validate_project_path
 
 
 def _is_exact_title_match(identifier: str, title: str) -> bool:
@@ -131,10 +130,10 @@ async def read_note(
         If the exact note isn't found, this tool provides helpful suggestions
         including related notes, search commands, and note creation templates.
     """
-    # Detect project from memory URL prefix before routing.
+    # Detect project from a memory URL or permalink prefix before routing.
     # project_id routes by external UUID, so it bypasses URL discovery entirely.
     if project is None and project_id is None:
-        detected = await detect_project_from_memory_url_prefix(
+        detected = await detect_project_from_identifier_prefix(
             identifier,
             ConfigManager().config,
             context=context,
@@ -278,51 +277,25 @@ def _result_file_path(item: dict[str, object]) -> Optional[str]:
                 value = item.get("file_path")
                 return str(value) if value else None
 
-            def _legacy_workspace_unqualified_path(path: str) -> str | None:
-                workspace_context = current_workspace_permalink_context()
-                if workspace_context is None:
-                    return None
-
-                workspace_prefix = generate_permalink(workspace_context.workspace_slug)
-                project_prefix = active_project.permalink
-                qualified_prefix = f"{workspace_prefix}/{project_prefix}"
-                normalized_path = path.strip("/")
-                if normalized_path == qualified_prefix:
-                    return project_prefix
-                if normalized_path.startswith(f"{qualified_prefix}/"):
-                    return f"{project_prefix}/{normalized_path.removeprefix(f'{qualified_prefix}/')}"
-                return None
-
-            direct_lookup_paths = [entity_path]
-            legacy_path = _legacy_workspace_unqualified_path(entity_path)
-            if legacy_path and legacy_path not in direct_lookup_paths:
-                # Trigger: existing cloud rows may still use project-prefixed permalinks.
-                # Why: new workspace-qualified IDs should read old notes without a re-sync.
-                # Outcome: try the legacy path after the canonical workspace path misses.
-                direct_lookup_paths.append(legacy_path)
-
-            for direct_lookup_path in direct_lookup_paths:
-                try:
-                    # Try to resolve identifier to entity ID
-                    entity_id = await knowledge_client.resolve_entity(
-                        direct_lookup_path, strict=True
-                    )
+            try:
+                # Try to resolve identifier to entity ID
+                entity_id = await knowledge_client.resolve_entity(entity_path, strict=True)
 
-                    # Fetch content using entity ID
-                    response = await resource_client.read(entity_id)
+                # Fetch content using entity ID
+                response = await resource_client.read(entity_id)
 
-                    # If successful, return the content
-                    if response.status_code == 200:
-                        logger.info(
-                            "Returning read_note result from resource: {path}",
-                            path=direct_lookup_path,
-                        )
-                        if output_format == "json":
-                            return await _read_json_payload(entity_id)
-                        return response.text
-                except Exception as e:  # pragma: no cover
-                    logger.info(f"Direct lookup failed for '{direct_lookup_path}': {e}")
-                    # Continue to alternate direct lookup paths, then fallback methods
+                # If successful, return the content
+                if response.status_code == 200:
+                    logger.info(
+                        "Returning read_note result from resource: {path}",
+                        path=entity_path,
+                    )
+                    if output_format == "json":
+                        return await _read_json_payload(entity_id)
+                    return response.text
+            except Exception as e:  # pragma: no cover
+                logger.info(f"Direct lookup failed for '{entity_path}': {e}")
+                # Continue to fallback methods
 
             # Fallback 1: Try title search via API
             logger.info(f"Search title for: {identifier}")

src/basic_memory/mcp/tools/search.py

@@ -11,10 +11,10 @@
 from pydantic import AliasChoices, BeforeValidator, Field
 
 from basic_memory.config import ConfigManager
-from basic_memory.utils import coerce_dict, coerce_list
+from basic_memory.utils import build_canonical_permalink, coerce_dict, coerce_list
 from basic_memory.mcp.container import get_container
 from basic_memory.mcp.project_context import (
-    detect_project_from_memory_url_prefix,
+    detect_project_from_identifier_prefix,
     get_project_client,
     resolve_project_and_path,
 )
@@ -402,11 +402,12 @@ def _qualify_permalink_for_project(permalink: object, project: str | None) -> ob
         return normalized_permalink
 
     workspace_slug, project_permalink = qualified_project.split("/", 1)
-    if normalized_permalink == project_permalink or normalized_permalink.startswith(
-        f"{project_permalink}/"
-    ):
-        return f"{workspace_slug}/{normalized_permalink}"
-    return f"{qualified_project}/{normalized_permalink}"
+    return build_canonical_permalink(
+        project_permalink,
+        normalized_permalink,
+        include_project=True,
+        workspace_permalink=workspace_slug,
+    )
 
 
 def _qualify_results_for_project(
@@ -805,10 +806,10 @@ async def search_notes(
             remainder = re.sub(r"\b(AND|OR|NOT)\b", "", remainder).strip()
             query = remainder or None
 
-    # Detect project from memory URL prefix before routing.
+    # Detect project from a memory URL or permalink prefix before routing.
     # project_id routes by external UUID, so it bypasses URL discovery entirely.
     if project is None and project_id is None and query is not None:
-        detected = await detect_project_from_memory_url_prefix(
+        detected = await detect_project_from_identifier_prefix(
             query,
             ConfigManager().config,
             context=context,

src/basic_memory/mcp/tools/utils.py

@@ -8,7 +8,7 @@
 from typing import Any, Optional
 
 import logfire
-from httpx import Response, URL, AsyncClient, HTTPStatusError
+from httpx import Response, URL, AsyncClient, HTTPStatusError, Headers
 from httpx._client import UseClientDefault, USE_CLIENT_DEFAULT
 from httpx._types import (
     RequestContent,
@@ -58,6 +58,22 @@ def _transport_error_span_attrs(exc: Exception) -> dict[str, Any]:
     }
 
 
+def _request_headers(headers: HeaderTypes | None) -> HeaderTypes | None:
+    """Merge request-local workspace permalink headers into outbound API calls."""
+    from basic_memory.workspace_context import workspace_permalink_headers
+
+    workspace_headers = workspace_permalink_headers()
+    if not workspace_headers:
+        return headers
+
+    if headers is None:
+        return workspace_headers
+
+    merged_headers = Headers(headers)
+    merged_headers.update(workspace_headers)
+    return merged_headers
+
+
 def get_error_message(
     status_code: int, url: URL | str, method: str, msg: Optional[str] = None
 ) -> str:
@@ -224,7 +240,7 @@ async def call_get(
             response = await client.get(
                 url,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -328,7 +344,7 @@ async def call_put(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -432,7 +448,7 @@ async def call_patch(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -542,7 +558,7 @@ async def call_post(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -667,7 +683,7 @@ async def call_delete(
             response = await client.delete(
                 url=url,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,