fix(mcp): route workspace-qualified memory urls

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

Author: phernandez

src/basic_memory/api/app.py

@@ -4,6 +4,7 @@
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.exception_handlers import http_exception_handler
+from fastapi.responses import JSONResponse
 from fastapi.routing import APIRouter
 from loguru import logger
 
@@ -29,6 +30,11 @@
 from basic_memory.config import init_api_logging
 from basic_memory.services.exceptions import EntityAlreadyExistsError
 from basic_memory.services.initialization import initialize_app
+from basic_memory.workspace_context import (
+    WORKSPACE_SLUG_HEADER,
+    WORKSPACE_TYPE_HEADER,
+    workspace_permalink_context,
+)
 
 
 @asynccontextmanager
@@ -87,6 +93,33 @@ async def lifespan(app: FastAPI):  # pragma: no cover
     lifespan=lifespan,
 )
 
+
+@app.middleware("http")
+async def workspace_permalink_context_middleware(request: Request, call_next):
+    """Populate workspace permalink context from request headers."""
+    workspace_slug = request.headers.get(WORKSPACE_SLUG_HEADER)
+    workspace_type = request.headers.get(WORKSPACE_TYPE_HEADER)
+
+    if bool(workspace_slug) != bool(workspace_type):
+        return JSONResponse(
+            status_code=400,
+            content={
+                "detail": (
+                    f"{WORKSPACE_SLUG_HEADER} and {WORKSPACE_TYPE_HEADER} must be provided together"
+                )
+            },
+        )
+
+    if not workspace_slug:
+        return await call_next(request)
+
+    with workspace_permalink_context(
+        workspace_slug=workspace_slug,
+        workspace_type=workspace_type,
+    ):
+        return await call_next(request)
+
+
 # Include v2 routers FIRST (more specific paths must match before /{project} catch-all)
 app.include_router(v2_knowledge, prefix="/v2/projects/{project_id}")
 app.include_router(v2_memory, prefix="/v2/projects/{project_id}")

src/basic_memory/mcp/async_client.py

@@ -94,9 +94,12 @@ async def _cloud_client(
     workspace: Optional[str] = None,
 ) -> AsyncIterator[AsyncClient]:
     """Create a cloud proxy client with resolved credentials."""
+    from basic_memory.workspace_context import workspace_permalink_headers
+
     token = await _resolve_cloud_token(config)
     proxy_base_url = f"{config.cloud_host}/proxy"
     headers = {"Authorization": f"Bearer {token}"}
+    headers.update(workspace_permalink_headers())
     if workspace:
         headers["X-Workspace-ID"] = workspace
     logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")

src/basic_memory/mcp/project_context.py

@@ -9,7 +9,7 @@
 """
 
 import asyncio
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, nullcontext
 from dataclasses import dataclass, field
 from typing import AsyncIterator, Awaitable, Callable, Optional, List, Tuple, cast
 from uuid import UUID
@@ -30,6 +30,10 @@
 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.workspace_context import (
+    current_workspace_permalink_context,
+    workspace_permalink_context,
+)
 
 # --- Workspace provider injection ---
 # Mirrors the set_client_factory() pattern in async_client.py.
@@ -62,6 +66,18 @@ class WorkspaceProjectIndex:
     failed_workspaces: tuple[WorkspaceInfo, ...] = ()
 
 
+@dataclass(frozen=True)
+class WorkspaceMemoryUrlResolution:
+    """Resolved workspace/project route for a workspace-qualified memory URL."""
+
+    entry: WorkspaceProjectEntry
+    canonical_path: str
+
+    @property
+    def project_identifier(self) -> str:
+        return self.entry.qualified_name
+
+
 def set_workspace_provider(provider: Callable[[], Awaitable[list[WorkspaceInfo]]]) -> None:
     """Override workspace discovery (for cloud app, testing, etc)."""
     global _workspace_provider
@@ -390,6 +406,89 @@ 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
+
+    normalized = normalize_project_reference(memory_url_path(identifier))
+    parts = normalized.split("/", 2)
+    if len(parts) != 3:
+        return None
+
+    workspace_slug, project_identifier, remainder = parts
+    if not workspace_slug or not project_identifier or not remainder:
+        return None
+    return workspace_slug, project_identifier, remainder
+
+
+def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool:
+    """Return True when workspace discovery can be used without forcing local routing."""
+    from basic_memory.mcp.async_client import (
+        _explicit_routing,
+        _force_local_mode,
+        is_factory_mode,
+    )
+
+    if _explicit_routing() and _force_local_mode():
+        return False
+
+    return (
+        is_factory_mode()
+        or (_explicit_routing() and not _force_local_mode())
+        or has_cloud_credentials(config)
+    )
+
+
+async def resolve_workspace_qualified_memory_url(
+    identifier: str,
+    context: Optional[Context] = None,
+) -> WorkspaceMemoryUrlResolution | None:
+    """Resolve a workspace-qualified memory URL against accessible workspaces."""
+    segments = _split_workspace_memory_url_segments(identifier)
+    if segments is None:
+        return None
+
+    workspace_slug, project_identifier, remainder = segments
+    index = await _ensure_workspace_project_index(context=context)
+    workspace = next(
+        (item for item in index.workspaces if item.slug.casefold() == workspace_slug.casefold()),
+        None,
+    )
+    if workspace is None:
+        return None
+
+    project_permalink = generate_permalink(project_identifier)
+    matches = [
+        entry
+        for entry in index.entries_by_permalink.get(project_permalink, ())
+        if entry.workspace.tenant_id == workspace.tenant_id
+    ]
+    if not matches:
+        if any(
+            failed_workspace.tenant_id == workspace.tenant_id
+            for failed_workspace in index.failed_workspaces
+        ):
+            raise ValueError(
+                f"Projects for workspace '{workspace.name}' ({workspace.slug}) "
+                "could not be loaded. Retry after workspace discovery recovers."
+            )
+
+        available = ", ".join(
+            entry.qualified_name
+            for entry in index.entries
+            if entry.workspace.tenant_id == workspace.tenant_id
+        )
+        raise ValueError(
+            f"Project '{project_identifier}' was not found in workspace "
+            f"'{workspace.name}' ({workspace.slug}). Available projects: {available}"
+        )
+
+    entry = matches[0]
+    canonical_path = f"{entry.workspace.slug}/{entry.project.permalink}/{remainder}"
+    return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path)
+
+
 def _format_qualified_choices(entries: tuple[WorkspaceProjectEntry, ...]) -> str:
     """Format qualified project choices for collision errors."""
     return " or ".join(entry.qualified_name for entry in entries)
@@ -856,13 +955,33 @@ async def resolve_project_and_path(
             return active_project, identifier, False
 
         normalized_path = normalize_project_reference(memory_url_path(identifier))
+        cached_project = await _get_cached_active_project(context)
+        cached_workspace = await _get_cached_active_workspace(context)
+        if cached_project and cached_workspace:
+            workspace_prefix = generate_permalink(cached_workspace.slug)
+            qualified_prefix = f"{workspace_prefix}/{cached_project.permalink}"
+            if normalized_path == qualified_prefix or normalized_path.startswith(
+                f"{qualified_prefix}/"
+            ):
+                return cached_project, normalized_path, True
+
+        workspace_context = current_workspace_permalink_context()
+        if workspace_context and project:
+            workspace_prefix = generate_permalink(workspace_context.workspace_slug)
+            project_permalink = generate_permalink(_unqualified_project_identifier(project))
+            qualified_prefix = f"{workspace_prefix}/{project_permalink}"
+            if normalized_path == qualified_prefix or normalized_path.startswith(
+                f"{qualified_prefix}/"
+            ):
+                active_project = await get_active_project(client, project, context, headers)
+                return active_project, normalized_path, True
+
         project_prefix, remainder = _split_project_prefix(normalized_path)
         include_project = config.permalinks_include_project
         # Trigger: memory URL begins with a potential project segment
         # Why: allow project-scoped memory URLs without requiring a separate project parameter
         # Outcome: attempt to resolve the prefix as a project and route to it
         if project_prefix:
-            cached_project = await _get_cached_active_project(context)
             if cached_project and _project_matches_identifier(cached_project, project_prefix):
                 resolved_project = await resolve_project_parameter(project_prefix, context=context)
                 if resolved_project and generate_permalink(resolved_project) != generate_permalink(
@@ -975,6 +1094,26 @@ def detect_project_from_url_prefix(identifier: str, config: BasicMemoryConfig) -
     return None
 
 
+async def detect_project_from_memory_url_prefix(
+    identifier: str,
+    config: BasicMemoryConfig,
+    context: Optional[Context] = None,
+) -> Optional[str]:
+    """Resolve a project from a memory URL prefix, including workspace-qualified URLs."""
+    if not identifier.strip().startswith("memory://"):
+        return None
+
+    if _cloud_workspace_discovery_available(config):
+        resolution = await resolve_workspace_qualified_memory_url(
+            identifier,
+            context=context,
+        )
+        if resolution is not None:
+            return resolution.project_identifier
+
+    return detect_project_from_url_prefix(identifier, config)
+
+
 @asynccontextmanager
 async def get_project_client(
     project: Optional[str] = None,
@@ -1113,12 +1252,18 @@ async def get_project_client(
             workspace_id=workspace_id,
         ):
             logger.debug("Using resolved workspace for cloud project routing")
-            async with get_client(
-                project_name=project_for_api,
-                workspace=workspace_id,
-            ) as client:
-                active_project = await get_active_project(client, project_for_api, context)
-                yield client, active_project
+            permalink_context = (
+                workspace_permalink_context(active_ws.slug, active_ws.workspace_type)
+                if active_ws is not None
+                else nullcontext()
+            )
+            with permalink_context:
+                async with get_client(
+                    project_name=project_for_api,
+                    workspace=workspace_id,
+                ) as client:
+                    active_project = await get_active_project(client, project_for_api, context)
+                    yield client, active_project
         return
 
     # Step 4: Local routing (default)

src/basic_memory/mcp/tools/build_context.py

@@ -9,7 +9,7 @@
 
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import (
-    detect_project_from_url_prefix,
+    detect_project_from_memory_url_prefix,
     get_project_client,
     resolve_project_and_path,
 )
@@ -211,7 +211,11 @@ async def build_context(
     """
     # Detect project from memory URL prefix before routing
     if project is None:
-        detected = detect_project_from_url_prefix(url, ConfigManager().config)
+        detected = await detect_project_from_memory_url_prefix(
+            url,
+            ConfigManager().config,
+            context=context,
+        )
         if detected:
             project = detected
 

src/basic_memory/mcp/tools/delete_note.py

@@ -7,7 +7,11 @@
 from pydantic import AliasChoices, Field
 
 from basic_memory.config import ConfigManager
-from basic_memory.mcp.project_context import detect_project_from_url_prefix, get_project_client
+from basic_memory.mcp.project_context import (
+    detect_project_from_memory_url_prefix,
+    get_project_client,
+    resolve_project_and_path,
+)
 from basic_memory.mcp.server import mcp
 
 
@@ -236,7 +240,11 @@ async def delete_note(
     #      where "research" is a directory, not a project name
     # Outcome: project is set from the URL prefix, routing goes to the correct project
     if project is None and identifier.strip().startswith("memory://"):
-        detected = detect_project_from_url_prefix(identifier, ConfigManager().config)
+        detected = await detect_project_from_memory_url_prefix(
+            identifier,
+            ConfigManager().config,
+            context=context,
+        )
         if detected:
             project = detected
 
@@ -253,11 +261,17 @@ async def delete_note(
 
         # Use typed KnowledgeClient for API calls
         knowledge_client = KnowledgeClient(client, active_project.external_id)
+        _, target_identifier, _ = await resolve_project_and_path(
+            client,
+            identifier,
+            active_project.name,
+            context,
+        )
 
         # Handle directory deletes
         if is_directory:
             try:
-                result = await knowledge_client.delete_directory(identifier)
+                result = await knowledge_client.delete_directory(target_identifier)
                 if output_format == "json":
                     return {
                         "deleted": result.failed_deletes == 0,
@@ -339,7 +353,7 @@ async def delete_note(
         note_file_path = None
         try:
             # Resolve identifier to entity ID
-            entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
+            entity_id = await knowledge_client.resolve_entity(target_identifier, strict=True)
             if output_format == "json":
                 entity = await knowledge_client.get_entity(entity_id)
                 note_title = entity.title