basicmachines-co-basic-memory/src/basic_memory/mcp/project_context.py

"""Project context utilities for Basic Memory MCP server.

Provides project lookup utilities for MCP tools.
Handles project validation and context management in one place.

Note: This module uses ProjectResolver for unified project resolution.
The resolve_project_parameter function is a thin wrapper for backwards
compatibility with existing MCP tools.
"""

from contextlib import asynccontextmanager
from typing import AsyncIterator, Optional, List, Tuple

from httpx import AsyncClient
from httpx._types import (
    HeaderTypes,
)
from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError

from basic_memory.config import ConfigManager
from basic_memory.project_resolver import ProjectResolver
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


async def resolve_project_parameter(
    project: Optional[str] = None,
    allow_discovery: bool = False,
    cloud_mode: Optional[bool] = None,
    default_project_mode: Optional[bool] = None,
    default_project: Optional[str] = None,
) -> Optional[str]:
    """Resolve project parameter using unified linear priority chain.

    This is a thin wrapper around ProjectResolver for backwards compatibility.
    New code should consider using ProjectResolver directly for more detailed
    resolution information.

    Resolution order (same for local and cloud modes):
    1. ENV_CONSTRAINT: BASIC_MEMORY_MCP_PROJECT env var (highest priority)
    2. EXPLICIT: project parameter passed directly
    3. DEFAULT: default project when default_project_mode=true
    4. Fallback: cloud → CLOUD_DISCOVERY or ValueError; local → NONE

    Args:
        project: Optional explicit project parameter
        allow_discovery: If True, allows returning None in cloud mode for discovery mode
            (used by tools like recent_activity that can operate across all projects)
        cloud_mode: Optional explicit cloud mode. If not provided, reads from ConfigManager.
        default_project_mode: Optional explicit default project mode. If not provided, reads from ConfigManager.
        default_project: Optional explicit default project. If not provided, reads from ConfigManager.

    Returns:
        Resolved project name or None if no resolution possible
    """
    # Load config for any values not explicitly provided
    if cloud_mode is None or default_project_mode is None or default_project is None:
        config = ConfigManager().config
        if cloud_mode is None:
            cloud_mode = config.cloud_mode
        if default_project_mode is None:
            default_project_mode = config.default_project_mode
        if default_project is None:
            default_project = config.default_project

    # Create resolver with configuration and resolve
    resolver = ProjectResolver.from_env(
        cloud_mode=cloud_mode,
        default_project_mode=default_project_mode,
        default_project=default_project,
    )
    result = resolver.resolve(project=project, allow_discovery=allow_discovery)
    return result.project


async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = None) -> List[str]:
    # Deferred import to avoid circular dependency with tools
    from basic_memory.mcp.tools.utils import call_get

    response = await call_get(client, "/v2/projects/", headers=headers)
    project_list = ProjectList.model_validate(response.json())
    return [project.name for project in project_list.projects]


async def get_active_project(
    client: AsyncClient,
    project: Optional[str] = None,
    context: Optional[Context] = None,
    headers: HeaderTypes | None = None,
) -> ProjectItem:
    """Get and validate project, setting it in context if available.

    Args:
        client: HTTP client for API calls
        project: Optional project name (resolved using hierarchy)
        context: Optional FastMCP context to cache the result

    Returns:
        The validated project item

    Raises:
        ValueError: If no project can be resolved
        HTTPError: If project doesn't exist or is inaccessible
    """
    # Deferred import to avoid circular dependency with tools
    from basic_memory.mcp.tools.utils import call_post

    resolved_project = await resolve_project_parameter(project)
    if not resolved_project:
        project_names = await get_project_names(client, headers)
        raise ValueError(
            "No project specified. "
            "Either set 'default_project_mode=true' in config, or use 'project' argument.\n"
            f"Available projects: {project_names}"
        )

    project = resolved_project

    # Check if already cached in context
    if context:
        cached_project = context.get_state("active_project")
        if cached_project and cached_project.name == project:
            logger.debug(f"Using cached project from context: {project}")
            return cached_project

    # Validate project exists by calling API
    logger.debug(f"Validating project: {project}")
    response = await call_post(
        client,
        "/v2/projects/resolve",
        json={"identifier": project},
        headers=headers,
    )
    resolved = ProjectResolveResponse.model_validate(response.json())
    active_project = ProjectItem(
        id=resolved.project_id,
        external_id=resolved.external_id,
        name=resolved.name,
        path=resolved.path,
        is_default=resolved.is_default,
    )

    # Cache in context if available
    if context:
        context.set_state("active_project", active_project)
        logger.debug(f"Cached project in context: {project}")

    logger.debug(f"Validated project: {active_project.name}")
    return active_project


def _split_project_prefix(path: str) -> tuple[Optional[str], str]:
    """Split a possible project prefix from a memory URL path."""
    if "/" not in path:
        return None, path

    project_prefix, remainder = path.split("/", 1)
    if not project_prefix or not remainder:
        return None, path

    if "*" in project_prefix:
        return None, path

    return project_prefix, remainder


async def resolve_project_and_path(
    client: AsyncClient,
    identifier: str,
    project: Optional[str] = None,
    context: Optional[Context] = None,
    headers: HeaderTypes | None = None,
) -> tuple[ProjectItem, str, bool]:
    """Resolve project and normalized path for memory:// identifiers.

    Returns:
        Tuple of (active_project, normalized_path, is_memory_url)
    """
    is_memory_url = identifier.strip().startswith("memory://")
    if not is_memory_url:
        active_project = await get_active_project(client, project, context, headers)
        return active_project, identifier, False

    normalized_path = normalize_project_reference(memory_url_path(identifier))
    project_prefix, remainder = _split_project_prefix(normalized_path)
    include_project = ConfigManager().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:
        try:
            from basic_memory.mcp.tools.utils import call_post

            response = await call_post(
                client,
                "/v2/projects/resolve",
                json={"identifier": project_prefix},
                headers=headers,
            )
            resolved = ProjectResolveResponse.model_validate(response.json())
        except ToolError as exc:
            if "project not found" not in str(exc).lower():
                raise
        else:
            resolved_project = await resolve_project_parameter(project_prefix)
            if resolved_project and generate_permalink(resolved_project) != generate_permalink(
                project_prefix
            ):
                raise ValueError(
                    f"Project is constrained to '{resolved_project}', cannot use '{project_prefix}'."
                )

            active_project = ProjectItem(
                id=resolved.project_id,
                external_id=resolved.external_id,
                name=resolved.name,
                path=resolved.path,
                is_default=resolved.is_default,
            )
            if context:
                context.set_state("active_project", active_project)

            resolved_path = (
                f"{resolved.permalink}/{remainder}" if include_project else remainder
            )
            return active_project, resolved_path, True

    # Trigger: no resolvable project prefix in the memory URL
    # Why: preserve existing memory URL behavior within the active project
    # Outcome: use the active project and normalize the path for lookup
    active_project = await get_active_project(client, project, context, headers)
    resolved_path = normalized_path
    if include_project:
        # Trigger: project-prefixed permalinks are enabled and the path lacks a prefix
        # Why: ensure memory URL lookups align with canonical permalinks
        # Outcome: prefix the path with the active project's permalink
        project_prefix = active_project.permalink
        if resolved_path != project_prefix and not resolved_path.startswith(f"{project_prefix}/"):
            resolved_path = f"{project_prefix}/{resolved_path}"
    return active_project, resolved_path, True


def add_project_metadata(result: str, project_name: str) -> str:
    """Add project context as metadata footer for assistant session tracking.

    Provides clear project context to help the assistant remember which
    project is being used throughout the conversation session.

    Args:
        result: The tool result string
        project_name: The project name that was used

    Returns:
        Result with project session tracking metadata
    """
    return f"{result}\n\n[Session: Using project '{project_name}']"


@asynccontextmanager
async def get_project_client(
    project: Optional[str] = None,
    context: Optional[Context] = None,
) -> AsyncIterator[Tuple[AsyncClient, ProjectItem]]:
    """Resolve project, create correctly-routed client, and validate project.

    Solves the bootstrap problem: we need to know the project name to choose
    the right client (local vs cloud), but we need the client to validate
    the project. This helper resolves the project from config first (no
    network), creates the correctly-routed client, then validates via API.

    Args:
        project: Optional explicit project parameter
        context: Optional FastMCP context for caching

    Yields:
        Tuple of (client, active_project)

    Raises:
        ValueError: If no project can be resolved
        RuntimeError: If cloud project but no API key configured
    """
    # Deferred import to avoid circular dependency
    from basic_memory.mcp.async_client import get_client

    # Step 1: Resolve project name from config (no network call)
    resolved_project = await resolve_project_parameter(project)
    if not resolved_project:
        # Fall back to local client to discover projects and raise helpful error
        async with get_client() as client:
            project_names = await get_project_names(client)
            raise ValueError(
                "No project specified. "
                "Either set 'default_project_mode=true' in config, or use 'project' argument.\n"
                f"Available projects: {project_names}"
            )

    # Step 2: Create client routed based on project's mode
    async with get_client(project_name=resolved_project) as client:
        # Step 3: Validate project exists via API
        active_project = await get_active_project(client, resolved_project, context)
        yield client, active_project