admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add per-project local/cloud routing with API key auth (#555)

Browse Source 
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-02-13 09:52:50 -06:00
committed by GitHub
parent 1428d18de1
commit d84708ca7f
30 changed files with 1319 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/cli/commands/cloud/core_commands.py
+76
View File
@@ -210,3 +210,79 @@ def promo(enabled: bool = typer.Option(True, "--on/--off", help="Enable or disab
console.print("[green]Cloud promo messages enabled[/green]")
else:
console.print("[yellow]Cloud promo messages disabled[/yellow]")
@cloud_app.command("set-key")
def set_key(
api_key: str = typer.Argument(..., help="API key (bmc_ prefixed) for cloud access"),
) -> None:
"""Save a cloud API key for per-project cloud routing.
The API key is account-level and used by projects set to cloud mode.
Create a key in the web app or use 'bm cloud create-key'.
Example:
bm cloud set-key bmc_abc123...
"""
if not api_key.startswith("bmc_"):
console.print("[red]Error: API key must start with 'bmc_'[/red]")
raise typer.Exit(1)
config_manager = ConfigManager()
config = config_manager.load_config()
config.cloud_api_key = api_key
config_manager.save_config(config)
console.print("[green]API key saved[/green]")
console.print("[dim]Projects set to cloud mode will use this key for authentication[/dim]")
console.print("[dim]Set a project to cloud mode: bm project set-cloud <name>[/dim]")
@cloud_app.command("create-key")
def create_key(
name: str = typer.Argument(..., help="Human-readable name for the API key"),
) -> None:
"""Create a new cloud API key and save it locally.
Requires active OAuth session (run 'bm cloud login' first).
The key is created via the cloud API and saved to local config.
Example:
bm cloud create-key "my-laptop"
"""
async def _create_key():
_, _, host_url = get_cloud_config()
host_url = host_url.rstrip("/")
console.print(f"[dim]Creating API key '{name}'...[/dim]")
response = await make_api_request(
method="POST",
url=f"{host_url}/api/keys",
json_data={"name": name},
)
key_data = response.json()
api_key = key_data.get("key")
if not api_key:
console.print("[red]Error: No key returned from API[/red]")
raise typer.Exit(1)
# Save to config
config_manager = ConfigManager()
config = config_manager.load_config()
config.cloud_api_key = api_key
config_manager.save_config(config)
console.print(f"[green]API key '{name}' created and saved[/green]")
console.print("[dim]Projects set to cloud mode will use this key for authentication[/dim]")
console.print("[dim]Set a project to cloud mode: bm project set-cloud <name>[/dim]")
try:
run_with_cleanup(_create_key())
except CloudAPIError as e:
console.print(f"[red]Error creating API key: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Unexpected error: {e}[/red]")
raise typer.Exit(1)
src/basic_memory/cli/commands/project.py
+72 -2
View File
@@ -13,7 +13,7 @@ from rich.table import Table
from basic_memory.cli.app import app
from basic_memory.cli.commands.command_utils import get_project_info, run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, ProjectMode
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.tools.utils import call_delete, call_get, call_patch, call_post, call_put
from basic_memory.schemas.project_info import ProjectList, ProjectStatusResponse
@@ -77,6 +77,7 @@ def list_projects(
table = Table(title="Basic Memory Projects")
table.add_column("Name", style="cyan")
table.add_column("Path", style="green")
table.add_column("Mode", style="blue")
# Add Local Path column if in cloud mode and not forcing local
if config.cloud_mode_enabled and not local:
@@ -90,9 +91,10 @@ def list_projects(
for project in result.projects:
is_default = "[X]" if project.is_default else ""
normalized_path = normalize_project_path(project.path)
project_mode = config.get_project_mode(project.name).value
# Build row based on mode
row = [project.name, format_path(normalized_path)]
row = [project.name, format_path(normalized_path), project_mode]
# Add local path if in cloud mode and not forcing local
if config.cloud_mode_enabled and not local:
@@ -511,6 +513,74 @@ def move_project(
raise typer.Exit(1)
@project_app.command("set-cloud")
def set_cloud(
name: str = typer.Argument(..., help="Name of the project to route through cloud"),
) -> None:
"""Set a project to cloud mode (route through cloud API).
Requires either an API key or an active OAuth session.
Examples:
bm cloud set-key bmc_abc123... # save API key, then:
bm project set-cloud research # route "research" through cloud
bm cloud login # OAuth login, then:
bm project set-cloud research # route "research" through cloud
"""
from basic_memory.cli.auth import CLIAuth
config_manager = ConfigManager()
config = config_manager.config
# Validate project exists in config
if name not in config.projects:
console.print(f"[red]Error: Project '{name}' not found in config[/red]")
raise typer.Exit(1)
# Validate credentials: API key or OAuth session
has_api_key = bool(config.cloud_api_key)
has_oauth = False
if not has_api_key:
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
has_oauth = auth.load_tokens() is not None
if not has_api_key and not has_oauth:
console.print("[red]Error: No cloud credentials found[/red]")
console.print("[dim]Run 'bm cloud set-key <key>' or 'bm cloud login' first[/dim]")
raise typer.Exit(1)
config.set_project_mode(name, ProjectMode.CLOUD)
config_manager.save_config(config)
console.print(f"[green]Project '{name}' set to cloud mode[/green]")
console.print("[dim]MCP tools and CLI commands for this project will route through cloud[/dim]")
@project_app.command("set-local")
def set_local(
name: str = typer.Argument(..., help="Name of the project to revert to local mode"),
) -> None:
"""Revert a project to local mode (use in-process ASGI transport).
Example:
bm project set-local research
"""
config_manager = ConfigManager()
config = config_manager.config
# Validate project exists in config
if name not in config.projects:
console.print(f"[red]Error: Project '{name}' not found in config[/red]")
raise typer.Exit(1)
config.set_project_mode(name, ProjectMode.LOCAL)
config_manager.save_config(config)
console.print(f"[green]Project '{name}' set to local mode[/green]")
console.print("[dim]MCP tools and CLI commands for this project will use local transport[/dim]")
@project_app.command("sync")
def sync_project_command(
name: str = typer.Option(..., "--name", help="Project name to sync"),
src/basic_memory/config.py
+37 -1
View File
@@ -24,6 +24,13 @@ WATCH_STATUS_JSON = "watch-status.json"
Environment = Literal["test", "dev", "user"]
class ProjectMode(str, Enum):
"""Per-project routing mode."""
LOCAL = "local"
CLOUD = "cloud"
class DatabaseBackend(str, Enum):
"""Supported database backends."""
@@ -37,6 +44,7 @@ class ProjectConfig:
name: str
home: Path
mode: ProjectMode = ProjectMode.LOCAL
@property
def project(self):
@@ -264,6 +272,16 @@ class BasicMemoryConfig(BaseSettings):
description="Most recent cloud promo version shown in CLI.",
)
cloud_api_key: Optional[str] = Field(
default=None,
description="API key for cloud access (bmc_ prefixed). Account-level, not per-project.",
)
project_modes: Dict[str, ProjectMode] = Field(
default_factory=dict,
description="Per-project routing mode. Projects not listed default to LOCAL.",
)
@property
def is_test_env(self) -> bool:
"""Check if running in a test environment.
@@ -297,6 +315,21 @@ class BasicMemoryConfig(BaseSettings):
# Fall back to config file value
return self.cloud_mode
def get_project_mode(self, project_name: str) -> ProjectMode:
"""Get the routing mode for a project.
Returns the per-project mode if set, otherwise LOCAL.
"""
return self.project_modes.get(project_name, ProjectMode.LOCAL)
def set_project_mode(self, project_name: str, mode: ProjectMode) -> None:
"""Set the routing mode for a project."""
if mode == ProjectMode.LOCAL:
# Remove from dict to keep config clean — LOCAL is the default
self.project_modes.pop(project_name, None)
else:
self.project_modes[project_name] = mode
@classmethod
def for_cloud_tenant(
cls,
@@ -387,7 +420,10 @@ class BasicMemoryConfig(BaseSettings):
@property
def project_list(self) -> List[ProjectConfig]: # pragma: no cover
"""Get all configured projects as ProjectConfig objects."""
return [ProjectConfig(name=name, home=Path(path)) for name, path in self.projects.items()]
return [
ProjectConfig(name=name, home=Path(path), mode=self.get_project_mode(name))
for name, path in self.projects.items()
]
@model_validator(mode="after")
def ensure_project_paths_exists(self) -> "BasicMemoryConfig": # pragma: no cover
src/basic_memory/mcp/async_client.py
+76 -10
View File
@@ -6,7 +6,7 @@ from httpx import ASGITransport, AsyncClient, Timeout
from loguru import logger
from basic_memory.api.app import app as fastapi_app
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, ProjectMode
def _force_local_mode() -> bool:
@@ -45,31 +45,54 @@ def set_client_factory(factory: Callable[[], AbstractAsyncContextManager[AsyncCl
@asynccontextmanager
async def get_client() -> AsyncIterator[AsyncClient]:
async def get_client(
project_name: Optional[str] = None,
) -> AsyncIterator[AsyncClient]:
"""Get an AsyncClient as a context manager.
This function provides proper resource management for HTTP clients,
ensuring connections are closed after use. It supports three modes:
ensuring connections are closed after use. Routing priority:
1. **Factory injection** (cloud app, tests):
If a custom factory is set via set_client_factory(), use that.
2. **CLI cloud mode**:
When cloud_mode_enabled is True, create HTTP client with auth
token from CLIAuth for requests to cloud proxy endpoint.
2. **Per-project cloud mode** (project_name provided):
If the project's mode is CLOUD, routes to cloud using API key or
OAuth token. Honored even when FORCE_LOCAL is set, because the user
explicitly declared this project as cloud.
3. **Local mode** (default):
3. **Per-project local mode** (project_name provided):
If the project's mode is LOCAL (or unspecified, default LOCAL), route
to local ASGI transport. This allows mixed local/cloud routing even when
global cloud mode is enabled.
4. **Force-local** (BASIC_MEMORY_FORCE_LOCAL env var):
Routes to local ASGI transport, ignoring global cloud settings.
5. **Global cloud mode** (deprecated fallback):
When cloud_mode_enabled is True, uses OAuth JWT token.
6. **Local mode** (default):
Use ASGI transport for in-process requests to local FastAPI app.
Args:
project_name: Optional project name for per-project routing.
If provided and the project's mode is CLOUD, routes to cloud
using the API key or OAuth token.
Usage:
async with get_client() as client:
response = await client.get("/path")
# Per-project routing
async with get_client(project_name="research") as client:
response = await client.get("/path")
Yields:
AsyncClient: Configured HTTP client for the current mode
Raises:
RuntimeError: If cloud mode is enabled but user is not authenticated
RuntimeError: If cloud routing needed but no API key / not authenticated
"""
if _client_factory:
# Use injected factory (cloud app, tests)
@@ -85,18 +108,61 @@ async def get_client() -> AsyncIterator[AsyncClient]:
pool=30.0, # 30 seconds for connection pool
)
# Trigger: project has per-project cloud mode set
# Why: per-project CLOUD is an explicit user declaration that should be
# honored even from the MCP server (which sets FORCE_LOCAL)
# Outcome: HTTP client with API key or OAuth auth to cloud proxy
if project_name and config.get_project_mode(project_name) == ProjectMode.CLOUD:
# Try API key first (explicit, no network)
token = config.cloud_api_key
if not token:
# Fall back to OAuth session (may refresh token)
from basic_memory.cli.auth import CLIAuth
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
token = await auth.get_valid_token()
if not token:
raise RuntimeError(
f"Project '{project_name}' is set to cloud mode but no credentials found. "
"Run 'bm cloud set-key <key>' or 'bm cloud login' first."
)
proxy_base_url = f"{config.cloud_host}/proxy"
logger.info(
f"Creating HTTP client for cloud project '{project_name}' at: {proxy_base_url}"
)
async with AsyncClient(
base_url=proxy_base_url,
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
) as client:
yield client
# Trigger: project is not explicitly cloud (LOCAL is the default)
# Why: project-scoped routing should honor local mode even when global
# cloud mode is enabled for backward compatibility
# Outcome: uses ASGI transport for in-process local API calls
elif project_name and config.get_project_mode(project_name) == ProjectMode.LOCAL:
logger.info(f"Project '{project_name}' is set to local mode - using ASGI transport")
async with AsyncClient(
transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout
) as client:
yield client
# Trigger: BASIC_MEMORY_FORCE_LOCAL env var is set
# Why: allows local MCP server and CLI commands to route locally
# even when cloud_mode_enabled is True
# Outcome: uses ASGI transport for in-process local API calls
if _force_local_mode():
elif _force_local_mode():
logger.info("Force local mode enabled - using ASGI client for local Basic Memory API")
async with AsyncClient(
transport=ASGITransport(app=fastapi_app), base_url="http://test", timeout=timeout
) as client:
yield client
elif config.cloud_mode_enabled:
# CLI cloud mode: inject auth when creating client
# Global cloud mode (deprecated fallback): inject OAuth auth when creating client
from basic_memory.cli.auth import CLIAuth
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
src/basic_memory/mcp/project_context.py
+48 -1
View File
@@ -8,7 +8,9 @@ The resolve_project_parameter function is a thin wrapper for backwards
compatibility with existing MCP tools.
"""
from typing import Optional, List
from contextlib import asynccontextmanager
from typing import AsyncIterator, Optional, List, Tuple
from httpx import AsyncClient
from httpx._types import (
HeaderTypes,
@@ -162,3 +164,48 @@ def add_project_metadata(result: str, project_name: str) -> str:
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
src/basic_memory/mcp/tools/build_context.py
+2 -6
View File
@@ -5,8 +5,7 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import (
@@ -100,10 +99,7 @@ async def build_context(
# URL is already validated and normalized by MemoryUrl type annotation
async with get_client() as client:
# Get the active project using the new stateless approach
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
# Import here to avoid circular import
from basic_memory.mcp.clients import MemoryClient
src/basic_memory/mcp/tools/canvas.py
+2 -5
View File
@@ -9,8 +9,7 @@ from typing import Dict, List, Any, Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_put, call_post, resolve_entity_id
@@ -94,9 +93,7 @@ async def canvas(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) 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}"
src/basic_memory/mcp/tools/delete_note.py
+3 -6
View File
@@ -5,9 +5,8 @@ from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.async_client import get_client
def _format_delete_error_response(project: str, error_message: str, identifier: str) -> str:
@@ -216,13 +215,11 @@ async def delete_note(
with suggestions for finding the correct identifier, including search
commands and alternative formats to try.
"""
async with get_client() as client:
async with get_project_client(project, context) as (client, active_project):
logger.debug(
f"Deleting {'directory' if is_directory else 'note'}: {identifier} in project: {project}"
f"Deleting {'directory' if is_directory else 'note'}: {identifier} in project: {active_project.name}"
)
active_project = await get_active_project(client, project, context)
# Import here to avoid circular import
from basic_memory.mcp.clients import KnowledgeClient
src/basic_memory/mcp/tools/edit_note.py
+2 -5
View File
@@ -5,8 +5,7 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project, add_project_metadata
from basic_memory.mcp.project_context import get_project_client, add_project_metadata
from basic_memory.mcp.server import mcp
@@ -212,9 +211,7 @@ async def edit_note(
search_notes() first to find the correct identifier. The tool provides detailed
error messages with suggestions if operations fail.
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
logger.info("MCP tool call", tool="edit_note", identifier=identifier, operation=operation)
# Validate operation
src/basic_memory/mcp/tools/list_directory.py
+2 -5
View File
@@ -5,8 +5,7 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
@@ -62,9 +61,7 @@ async def list_directory(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
logger.debug(
f"Listing directory '{dir_name}' in project {project} with depth={depth}, glob='{file_name_glob}'"
)
src/basic_memory/mcp/tools/move_note.py
+3 -6
View File
@@ -6,9 +6,8 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.utils import validate_project_path
@@ -412,13 +411,11 @@ async def move_note(
- Re-indexes the entity for search
- Maintains all observations and relations
"""
async with get_client() as client:
async with get_project_client(project, context) as (client, active_project):
logger.debug(
f"Moving {'directory' if is_directory else 'note'}: {identifier} to {destination_path} in project: {project}"
f"Moving {'directory' if is_directory else 'note'}: {identifier} to {destination_path} in project: {active_project.name}"
)
active_project = await get_active_project(client, project, context)
# Validate destination path to prevent path traversal attacks
project_path = active_project.home
if not validate_project_path(destination_path, project_path):
src/basic_memory/mcp/tools/read_content.py
+2 -5
View File
@@ -15,9 +15,8 @@ from PIL import Image as PILImage
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.tools.utils import call_get, resolve_entity_id
from basic_memory.schemas.memory import memory_url_path
from basic_memory.utils import validate_project_path
@@ -202,9 +201,7 @@ async def read_content(
"""
logger.info("Reading file", path=path, project=project)
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
url = memory_url_path(path)
# Validate path to prevent path traversal attacks
src/basic_memory/mcp/tools/read_note.py
+2 -6
View File
@@ -6,8 +6,7 @@ from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
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
@@ -77,10 +76,7 @@ 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.
"""
async with get_client() as client:
# Get and validate the project
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
# Validate identifier to prevent path traversal attacks
# We need to check both the raw identifier and the processed path
processed_path = memory_url_path(identifier)
src/basic_memory/mcp/tools/recent_activity.py
+95 -99
View File
@@ -7,7 +7,10 @@ from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project, resolve_project_parameter
from basic_memory.mcp.project_context import (
get_project_client,
resolve_project_parameter,
)
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.base import TimeFrame
@@ -99,50 +102,51 @@ async def recent_activity(
- For focused queries, consider using build_context with a specific URI
- Max timeframe is 1 year in the past
"""
async with get_client() as client:
# Build common parameters for API calls
params = {
"page": 1,
"page_size": 10,
"max_related": 10,
}
if depth:
params["depth"] = depth
if timeframe:
params["timeframe"] = timeframe # pyright: ignore
# Build common parameters for API calls
params: dict = {
"page": 1,
"page_size": 10,
"max_related": 10,
}
if depth:
params["depth"] = depth
if timeframe:
params["timeframe"] = timeframe # pyright: ignore
# Validate and convert type parameter
if type:
# Convert single string to list
if isinstance(type, str):
type_list = [type]
else:
type_list = type
# Validate and convert type parameter
if type:
# Convert single string to list
if isinstance(type, str):
type_list = [type]
else:
type_list = type
# Validate each type against SearchItemType enum
validated_types = []
for t in type_list:
try:
# Try to convert string to enum
if isinstance(t, str):
validated_types.append(SearchItemType(t.lower()))
except ValueError:
valid_types = [t.value for t in SearchItemType]
raise ValueError(f"Invalid type: {t}. Valid types are: {valid_types}")
# Validate each type against SearchItemType enum
validated_types = []
for t in type_list:
try:
# Try to convert string to enum
if isinstance(t, str):
validated_types.append(SearchItemType(t.lower()))
except ValueError:
valid_types = [t.value for t in SearchItemType]
raise ValueError(f"Invalid type: {t}. Valid types are: {valid_types}")
# Add validated types to params
params["type"] = [t.value for t in validated_types] # pyright: ignore
# Add validated types to params
params["type"] = [t.value for t in validated_types] # pyright: ignore
# 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
resolved_project = await resolve_project_parameter(project, allow_discovery=True)
if resolved_project is None:
# Discovery Mode: Get activity across all projects
logger.info(
f"Getting recent activity across all projects: type={type}, depth={depth}, timeframe={timeframe}"
)
if resolved_project is None:
# Discovery Mode: Get activity across all projects
# Uses plain get_client() since we iterate across all projects (no single project routing)
logger.info(
f"Getting recent activity across all projects: type={type}, depth={depth}, timeframe={timeframe}"
)
async with get_client() as client:
# Get list of all projects
response = await call_get(client, "/v2/projects/")
project_list = ProjectList.model_validate(response.json())
@@ -181,76 +185,68 @@ async def recent_activity(
most_active_count = item_count
most_active_project = project_info.name
# Build summary stats
summary = ActivityStats(
total_projects=len(project_list.projects),
active_projects=active_projects,
most_active_project=most_active_project,
total_items=total_items,
total_entities=total_entities,
total_relations=total_relations,
total_observations=total_observations,
)
# Build summary stats
summary = ActivityStats(
total_projects=len(project_list.projects),
active_projects=active_projects,
most_active_project=most_active_project,
total_items=total_items,
total_entities=total_entities,
total_relations=total_relations,
total_observations=total_observations,
)
# Generate guidance for the assistant
guidance_lines = ["\n" + "─" * 40]
if active_projects == 0:
# No recent activity
guidance_lines.extend(
[
"No recent activity found in any project.",
"Consider: Ask which project to use or if they want to create a new one.",
]
)
else:
# At least one project has activity: suggest the most active project.
suggested_project = most_active_project or next(
(
name
for name, activity in projects_activity.items()
if activity.item_count > 0
),
None,
)
if suggested_project:
suffix = (
f"(most active with {most_active_count} items)"
if most_active_count > 0
else ""
)
guidance_lines.append(
f"Suggested project: '{suggested_project}' {suffix}".strip()
)
if active_projects == 1:
guidance_lines.append(
f"Ask user: 'Should I use {suggested_project} for this task?'"
)
else:
guidance_lines.append(
f"Ask user: 'Should I use {suggested_project} for this task, or would you prefer a different project?'"
)
# Generate guidance for the assistant
guidance_lines = ["\n" + "─" * 40]
if active_projects == 0:
# No recent activity
guidance_lines.extend(
[
"",
"Session reminder: Remember their project choice throughout this conversation.",
"No recent activity found in any project.",
"Consider: Ask which project to use or if they want to create a new one.",
]
)
guidance = "\n".join(guidance_lines)
# Format discovery mode output
return _format_discovery_output(projects_activity, summary, timeframe, guidance)
else:
# Project-Specific Mode: Get activity for specific project
logger.info(
f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}"
# At least one project has activity: suggest the most active project.
suggested_project = most_active_project or next(
(name for name, activity in projects_activity.items() if activity.item_count > 0),
None,
)
if suggested_project:
suffix = (
f"(most active with {most_active_count} items)" if most_active_count > 0 else ""
)
guidance_lines.append(f"Suggested project: '{suggested_project}' {suffix}".strip())
if active_projects == 1:
guidance_lines.append(
f"Ask user: 'Should I use {suggested_project} for this task?'"
)
else:
guidance_lines.append(
f"Ask user: 'Should I use {suggested_project} for this task, or would you prefer a different project?'"
)
active_project = await get_active_project(client, resolved_project, context)
guidance_lines.extend(
[
"",
"Session reminder: Remember their project choice throughout this conversation.",
]
)
guidance = "\n".join(guidance_lines)
# Format discovery mode output
return _format_discovery_output(projects_activity, summary, timeframe, guidance)
else:
# Project-Specific Mode: Get activity for specific project
# Uses get_project_client() for per-project routing (local vs cloud)
logger.info(
f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}"
)
async with get_project_client(resolved_project, context) as (client, active_project):
response = await call_get(
client,
f"/v2/projects/{active_project.external_id}/memory/recent",
src/basic_memory/mcp/tools/search.py
+3 -7
View File
@@ -6,8 +6,7 @@ from typing import List, Optional, Dict, Any
from loguru import logger
from fastmcp import Context
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.server import mcp
from basic_memory.schemas.search import (
SearchItemType,
@@ -430,9 +429,7 @@ async def search_notes(
if status:
search_query.status = status
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
logger.info(f"Searching for {search_query} in project {active_project.name}")
try:
@@ -498,8 +495,7 @@ async def search_by_metadata(
page = (offset // limit) + 1
offset_within_page = offset % limit
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, context) as (client, active_project):
logger.info(
f"Structured search in project {active_project.name} filters={filters} limit={limit} offset={offset}"
)
src/basic_memory/mcp/tools/write_note.py
+3 -7
View File
@@ -4,8 +4,7 @@ from typing import List, Union, Optional
from loguru import logger
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project, add_project_metadata
from basic_memory.mcp.project_context import get_project_client, add_project_metadata
from basic_memory.mcp.server import mcp
from fastmcp import Context
from basic_memory.schemas.base import Entity
@@ -110,14 +109,11 @@ async def write_note(
HTTPError: If project doesn't exist or is inaccessible
SecurityError: If directory path attempts path traversal
"""
async with get_client() as client:
async with get_project_client(project, context) as (client, active_project):
logger.info(
f"MCP tool call tool=write_note project={project} directory={directory}, title={title}, tags={tags}"
f"MCP tool call tool=write_note project={active_project.name} directory={directory}, title={title}, tags={tags}"
)
# Get and validate the project (supports optional project parameter)
active_project = await get_active_project(client, project, context)
# Normalize "/" to empty string for root directory (must happen before validation)
if directory == "/":
directory = ""
src/basic_memory/services/initialization.py
+11 -1
View File
@@ -13,7 +13,7 @@ from pathlib import Path
from loguru import logger
from basic_memory import db
from basic_memory.config import BasicMemoryConfig
from basic_memory.config import BasicMemoryConfig, ProjectMode
from basic_memory.models import Project
from basic_memory.repository import (
ProjectRepository,
@@ -115,6 +115,16 @@ async def initialize_file_sync(
active_projects = [p for p in active_projects if p.name == constrained_project]
logger.info(f"Background sync constrained to project: {constrained_project}")
# Skip cloud-mode projects — their files live on the cloud instance, not locally
cloud_projects = [
p.name for p in active_projects if app_config.get_project_mode(p.name) == ProjectMode.CLOUD
]
if cloud_projects:
active_projects = [
p for p in active_projects if app_config.get_project_mode(p.name) != ProjectMode.CLOUD
]
logger.info(f"Skipping cloud-mode projects for local sync: {cloud_projects}")
# Start sync for all projects as background tasks (non-blocking)
async def sync_project_background(project: Project):
"""Sync a single project in the background."""
src/basic_memory/sync/watch_service.py
+17 -1
View File
@@ -10,7 +10,7 @@ from typing import List, Optional, Set, Sequence, Callable, Awaitable, TYPE_CHEC
if TYPE_CHECKING:
from basic_memory.sync.sync_service import SyncService
from basic_memory.config import BasicMemoryConfig, WATCH_STATUS_JSON
from basic_memory.config import BasicMemoryConfig, ProjectMode, WATCH_STATUS_JSON
from basic_memory.ignore_utils import load_gitignore_patterns, should_ignore_path
from basic_memory.models import Project
from basic_memory.repository import ProjectRepository
@@ -177,6 +177,22 @@ class WatchService:
# Reload projects to catch any new/removed projects
projects = await self.project_repository.get_active_projects()
# Trigger: project is configured for cloud routing
# Why: cloud projects should not be watched/synced by local file watchers
# Outcome: watch cycle only observes local-mode projects
cloud_projects = [
p.name
for p in projects
if self.app_config.get_project_mode(p.name) == ProjectMode.CLOUD
]
if cloud_projects:
projects = [
p
for p in projects
if self.app_config.get_project_mode(p.name) != ProjectMode.CLOUD
]
logger.info(f"Skipping cloud-mode projects in watch cycle: {cloud_projects}")
project_paths = [project.path for project in projects]
logger.debug(f"Starting watch cycle for directories: {project_paths}")