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

Merge branch 'main' of github.com:basicmachines-co/basic-memory

Browse Source
This commit is contained in:
phernandez
2026-02-18 10:50:58 -06:00
parent 36e67e6eec f2683291e4
commit 0a36256f8a
29 changed files with 902 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/cli/commands/__init__.py
+11 -1
View File
@@ -1,7 +1,16 @@
"""CLI commands for basic-memory."""
from . import status, db, doctor, import_memory_json, mcp, import_claude_conversations
from . import import_claude_projects, import_chatgpt, tool, project, format, schema, watch
from . import (
import_claude_projects,
import_chatgpt,
tool,
project,
format,
schema,
watch,
workspace,
)
__all__ = [
"status",
@@ -17,4 +26,5 @@ __all__ = [
"format",
"schema",
"watch",
"workspace",
]
src/basic_memory/cli/commands/tool.py
+89 -43
View File
@@ -13,9 +13,8 @@ from basic_memory.cli.app import app
from basic_memory.cli.commands.command_utils import run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.clients import KnowledgeClient, ResourceClient
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.project_context import get_project_client
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.base import Entity, TimeFrame
from basic_memory.schemas.memory import GraphContext, MemoryUrl, memory_url_path
@@ -25,9 +24,6 @@ from basic_memory.schemas.search import SearchItemType
from basic_memory.mcp.prompts.continue_conversation import (
continue_conversation as mcp_continue_conversation,
)
from basic_memory.mcp.prompts.recent_activity import (
recent_activity_prompt as recent_activity_prompt,
)
from basic_memory.mcp.tools import build_context as mcp_build_context
from basic_memory.mcp.tools import edit_note as mcp_edit_note
from basic_memory.mcp.tools import read_note as mcp_read_note
@@ -99,15 +95,26 @@ def _parse_opening_frontmatter(content: str) -> tuple[str, dict[str, Any] | None
async def _write_note_json(
title: str, content: str, folder: str, project_name: Optional[str], tags: Optional[List[str]]
title: str,
content: str,
folder: str,
project_name: Optional[str],
workspace: Optional[str],
tags: Optional[List[str]],
) -> dict:
"""Write a note and return structured JSON metadata."""
# Use the MCP tool to create/update the entity (handles create-or-update logic)
await mcp_write_note.fn(title, content, folder, project_name, tags)
await mcp_write_note.fn(
title=title,
content=content,
directory=folder,
project=project_name,
workspace=workspace,
tags=tags,
)
# Resolve the entity to get metadata back
async with get_client(project_name=project_name) as client:
active_project = await get_active_project(client, project_name)
async with get_project_client(project_name, workspace) as (client, active_project):
knowledge_client = KnowledgeClient(client, active_project.external_id)
entity = Entity(title=title, directory=folder)
@@ -125,11 +132,14 @@ async def _write_note_json(
async def _read_note_json(
identifier: str, project_name: Optional[str], page: int, page_size: int
identifier: str,
project_name: Optional[str],
workspace: Optional[str],
page: int,
page_size: int,
) -> dict:
"""Read a note and return structured JSON with content and metadata."""
async with get_client(project_name=project_name) as client:
active_project = await get_active_project(client, project_name)
async with get_project_client(project_name, workspace) as (client, active_project):
knowledge_client = KnowledgeClient(client, active_project.external_id)
resource_client = ResourceClient(client, active_project.external_id)
@@ -146,7 +156,10 @@ async def _read_note_json(
from basic_memory.mcp.tools.search import search_notes as mcp_search_tool
title_results = await mcp_search_tool.fn(
query=identifier, search_type="title", project=project_name
query=identifier,
search_type="title",
project=project_name,
workspace=workspace,
)
if title_results and hasattr(title_results, "results") and title_results.results:
result = title_results.results[0]
@@ -172,13 +185,13 @@ async def _edit_note_json(
operation: str,
content: str,
project_name: Optional[str],
workspace: Optional[str],
section: Optional[str],
find_text: Optional[str],
expected_replacements: int,
) -> dict:
"""Edit a note and return structured JSON metadata."""
async with get_client(project_name=project_name) as client:
active_project = await get_active_project(client, project_name)
async with get_project_client(project_name, workspace) as (client, active_project):
knowledge_client = KnowledgeClient(client, active_project.external_id)
entity_id = await knowledge_client.resolve_entity(identifier)
@@ -227,11 +240,12 @@ async def _recent_activity_json(
depth: Optional[int],
timeframe: Optional[TimeFrame],
project_name: Optional[str] = None,
workspace: Optional[str] = None,
page: int = 1,
page_size: int = 50,
) -> list:
"""Get recent activity and return structured JSON list."""
async with get_client(project_name=project_name) as client:
async with get_project_client(project_name, workspace) as (client, active_project):
# Build query params matching the MCP tool's logic
params: dict = {"page": page, "page_size": page_size, "max_related": 10}
if depth:
@@ -241,7 +255,6 @@ async def _recent_activity_json(
if type:
params["type"] = [t.value for t in type]
active_project = await get_active_project(client, project_name)
response = await call_get(
client,
f"/v2/projects/{active_project.external_id}/memory/recent",
@@ -275,6 +288,10 @@ def write_note(
help="The project to write to. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
content: Annotated[
Optional[str],
typer.Option(
@@ -362,12 +379,19 @@ def write_note(
with force_routing(local=local, cloud=cloud):
if format == "json":
result = run_with_cleanup(
_write_note_json(title, content, folder, project_name, tags)
_write_note_json(title, content, folder, project_name, workspace, tags)
)
print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
else:
note = run_with_cleanup(
mcp_write_note.fn(title, content, folder, project_name, tags)
mcp_write_note.fn(
title=title,
content=content,
directory=folder,
project=project_name,
workspace=workspace,
tags=tags,
)
)
rprint(note)
except ValueError as e:
@@ -389,6 +413,10 @@ def read_note(
help="The project to use for the note. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
page: int = 1,
page_size: int = 10,
format: str = typer.Option("text", "--format", help="Output format: text or json"),
@@ -429,7 +457,7 @@ def read_note(
with force_routing(local=local, cloud=cloud):
if format == "json":
result = run_with_cleanup(
_read_note_json(identifier, project_name, page, page_size)
_read_note_json(identifier, project_name, workspace, page, page_size)
)
stripped_content, parsed_frontmatter = _parse_opening_frontmatter(result["content"])
result["frontmatter"] = parsed_frontmatter
@@ -437,7 +465,15 @@ def read_note(
result["content"] = stripped_content
print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
else:
note = run_with_cleanup(mcp_read_note.fn(identifier, project_name, page, page_size))
note = run_with_cleanup(
mcp_read_note.fn(
identifier=identifier,
project=project_name,
workspace=workspace,
page=page,
page_size=page_size,
)
)
if strip_frontmatter:
note, _ = _parse_opening_frontmatter(note)
rprint(note)
@@ -462,6 +498,10 @@ def edit_note(
help="The project to edit. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
find_text: Annotated[
Optional[str], typer.Option("--find-text", help="Text to find for find_replace operation")
] = None,
@@ -509,6 +549,7 @@ def edit_note(
operation=operation,
content=content,
project_name=project_name,
workspace=workspace,
section=section,
find_text=find_text,
expected_replacements=expected_replacements,
@@ -522,6 +563,7 @@ def edit_note(
operation=operation,
content=content,
project=project_name,
workspace=workspace,
section=section,
find_text=find_text,
expected_replacements=expected_replacements,
@@ -547,6 +589,10 @@ def build_context(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
page: int = 1,
@@ -582,6 +628,7 @@ def build_context(
result = run_with_cleanup(
mcp_build_context.fn(
project=project_name,
workspace=workspace,
url=url,
depth=depth,
timeframe=timeframe,
@@ -609,6 +656,10 @@ def recent_activity(
Optional[str],
typer.Option(help="The project to use. If not provided, the default project will be used."),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
page: int = typer.Option(1, "--page", help="Page number for pagination (JSON format)"),
@@ -642,7 +693,15 @@ def recent_activity(
with force_routing(local=local, cloud=cloud):
if format == "json":
result = run_with_cleanup(
_recent_activity_json(type, depth, timeframe, project_name, page, page_size)
_recent_activity_json(
type=type,
depth=depth,
timeframe=timeframe,
project_name=project_name,
workspace=workspace,
page=page,
page_size=page_size,
)
)
print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
else:
@@ -652,6 +711,7 @@ def recent_activity(
depth=depth,
timeframe=timeframe,
project=project_name,
workspace=workspace,
)
)
# The tool returns a formatted string directly
@@ -682,6 +742,10 @@ def search_notes(
help="The project to use for the note. If not provided, the default project will be used."
),
] = None,
workspace: Annotated[
Optional[str],
typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
] = None,
after_date: Annotated[
Optional[str],
typer.Option("--after_date", help="Search results after date, eg. '2d', '1 week'"),
@@ -793,8 +857,9 @@ def search_notes(
with force_routing(local=local, cloud=cloud):
results = run_with_cleanup(
mcp_search.fn(
query or "",
project_name,
query=query or "",
project=project_name,
workspace=workspace,
search_type=search_type,
page=page,
after_date=after_date,
@@ -857,22 +922,3 @@ def continue_conversation(
typer.echo(f"Error continuing conversation: {e}", err=True)
raise typer.Exit(1)
raise
# @tool_app.command(name="show-recent-activity")
# def show_recent_activity(
# timeframe: Annotated[
# str, typer.Option(help="How far back to look for activity")
# ] = "7d",
# ):
# """Prompt to show recent activity."""
# try:
# # Prompt functions return formatted strings directly
# session = asyncio.run(recent_activity_prompt(timeframe=timeframe))
# rprint(session)
# except Exception as e: # pragma: no cover
# if not isinstance(e, typer.Exit):
# logger.exception("Error continuing conversation", e)
# typer.echo(f"Error continuing conversation: {e}", err=True)
# raise typer.Exit(1)
# raise
src/basic_memory/cli/commands/workspace.py
+57
View File
@@ -0,0 +1,57 @@
"""Workspace commands for Basic Memory cloud workspaces."""
import typer
from rich.console import Console
from rich.table import Table
from basic_memory.cli.app import app
from basic_memory.cli.commands.command_utils import run_with_cleanup
from basic_memory.mcp.project_context import get_available_workspaces
console = Console()
workspace_app = typer.Typer(help="Manage cloud workspaces")
app.add_typer(workspace_app, name="workspace")
@workspace_app.command("list")
def list_workspaces() -> None:
"""List cloud workspaces available to the current OAuth session."""
async def _list():
return await get_available_workspaces()
try:
workspaces = run_with_cleanup(_list())
except RuntimeError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1)
except Exception as exc: # pragma: no cover
console.print(f"[red]Error listing workspaces: {exc}[/red]")
raise typer.Exit(1)
if not workspaces:
console.print("[yellow]No accessible workspaces found.[/yellow]")
return
table = Table(title="Available Workspaces")
table.add_column("Name", style="cyan")
table.add_column("Type", style="blue")
table.add_column("Role", style="green")
table.add_column("Tenant ID", style="yellow")
for workspace in workspaces:
table.add_row(
workspace.name,
workspace.workspace_type,
workspace.role,
workspace.tenant_id,
)
console.print(table)
@app.command("workspaces")
def workspaces_alias() -> None:
"""Alias for `bm workspace list`."""
list_workspaces()
src/basic_memory/cli/main.py
+1
View File
@@ -28,6 +28,7 @@ if not _version_only_invocation(sys.argv[1:]):
schema,
status,
tool,
workspace,
)
warnings.filterwarnings("ignore") # pragma: no cover
src/basic_memory/mcp/async_client.py
+27 -3
View File
@@ -60,13 +60,36 @@ async def _resolve_cloud_token(config) -> str:
)
async def _cloud_client(config, timeout: Timeout) -> AsyncIterator[AsyncClient]:
@asynccontextmanager
async def _cloud_client(
config,
timeout: Timeout,
workspace: Optional[str] = None,
) -> AsyncIterator[AsyncClient]:
"""Create a cloud proxy client with resolved credentials."""
token = await _resolve_cloud_token(config)
proxy_base_url = f"{config.cloud_host}/proxy"
headers = {"Authorization": f"Bearer {token}"}
if workspace:
headers["X-Workspace-ID"] = workspace
logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")
async with AsyncClient(
base_url=proxy_base_url,
headers=headers,
timeout=timeout,
) as client:
yield client
@asynccontextmanager
async def get_cloud_control_plane_client() -> AsyncIterator[AsyncClient]:
"""Create a control-plane cloud client for endpoints outside /proxy."""
config = ConfigManager().config
timeout = _build_timeout()
token = await _resolve_cloud_token(config)
logger.info(f"Creating HTTP client for cloud control plane at: {config.cloud_host}")
async with AsyncClient(
base_url=config.cloud_host,
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
) as client:
@@ -86,6 +109,7 @@ def set_client_factory(factory: Callable[[], AbstractAsyncContextManager[AsyncCl
@asynccontextmanager
async def get_client(
project_name: Optional[str] = None,
workspace: Optional[str] = None,
) -> AsyncIterator[AsyncClient]:
"""Get an AsyncClient as a context manager.
@@ -116,7 +140,7 @@ async def get_client(
if _force_cloud_mode():
logger.info("Explicit cloud routing enabled - using cloud proxy client")
async for client in _cloud_client(config, timeout):
async with _cloud_client(config, timeout, workspace=workspace) as client:
yield client
return
@@ -129,7 +153,7 @@ async def get_client(
if project_mode == ProjectMode.CLOUD:
logger.info(f"Project '{project_name}' is cloud mode - using cloud proxy client")
try:
async for client in _cloud_client(config, timeout):
async with _cloud_client(config, timeout, workspace=workspace) as client:
yield client
except RuntimeError as exc:
raise RuntimeError(
src/basic_memory/mcp/project_context.py
+120 -4
View File
@@ -19,8 +19,9 @@ from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.config import ConfigManager
from basic_memory.config import ConfigManager, ProjectMode
from basic_memory.project_resolver import ProjectResolver
from basic_memory.schemas.cloud import WorkspaceInfo, WorkspaceListResponse
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
@@ -56,8 +57,7 @@ async def resolve_project_parameter(
# Load config for any values not explicitly provided
if default_project is None:
config = ConfigManager().config
if default_project is None:
default_project = config.default_project
default_project = config.default_project
# Create resolver with configuration and resolve
resolver = ProjectResolver.from_env(
@@ -76,6 +76,100 @@ async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = N
return [project.name for project in project_list.projects]
def _workspace_matches_identifier(workspace: WorkspaceInfo, identifier: str) -> bool:
"""Return True when identifier matches workspace tenant_id or name."""
if workspace.tenant_id == identifier:
return True
return workspace.name.lower() == identifier.lower()
def _workspace_choices(workspaces: list[WorkspaceInfo]) -> str:
"""Format deterministic workspace choices for prompt-style errors."""
return "\n".join(
[
(
f"- {item.name} "
f"(type={item.workspace_type}, role={item.role}, tenant_id={item.tenant_id})"
)
for item in workspaces
]
)
async def get_available_workspaces(context: Optional[Context] = None) -> list[WorkspaceInfo]:
"""Load available cloud workspaces for the current authenticated user."""
if context:
cached_workspaces = context.get_state("available_workspaces")
if isinstance(cached_workspaces, list) and all(
isinstance(item, WorkspaceInfo) for item in cached_workspaces
):
return cached_workspaces
from basic_memory.mcp.async_client import get_cloud_control_plane_client
from basic_memory.mcp.tools.utils import call_get
async with get_cloud_control_plane_client() as client:
response = await call_get(client, "/workspaces/")
workspace_list = WorkspaceListResponse.model_validate(response.json())
if context:
context.set_state("available_workspaces", workspace_list.workspaces)
return workspace_list.workspaces
async def resolve_workspace_parameter(
workspace: Optional[str] = None,
context: Optional[Context] = None,
) -> WorkspaceInfo:
"""Resolve workspace using explicit input, session cache, and cloud discovery."""
if context:
cached_workspace = context.get_state("active_workspace")
if isinstance(cached_workspace, WorkspaceInfo) and (
workspace is None or _workspace_matches_identifier(cached_workspace, workspace)
):
logger.debug(f"Using cached workspace from context: {cached_workspace.tenant_id}")
return cached_workspace
workspaces = await get_available_workspaces(context=context)
if not workspaces:
raise ValueError(
"No accessible workspaces found for this account. "
"Ensure you have an active subscription and tenant access."
)
selected_workspace: WorkspaceInfo | None = None
if workspace:
matches = [item for item in workspaces if _workspace_matches_identifier(item, workspace)]
if not matches:
raise ValueError(
f"Workspace '{workspace}' was not found.\n"
f"Available workspaces:\n{_workspace_choices(workspaces)}"
)
if len(matches) > 1:
raise ValueError(
f"Workspace name '{workspace}' matches multiple workspaces. "
"Use tenant_id instead.\n"
f"Available workspaces:\n{_workspace_choices(workspaces)}"
)
selected_workspace = matches[0]
elif len(workspaces) == 1:
selected_workspace = workspaces[0]
else:
raise ValueError(
"Multiple workspaces are available. Ask the user which workspace to use, then retry "
"with the 'workspace' argument set to the tenant_id or unique name.\n"
f"Available workspaces:\n{_workspace_choices(workspaces)}"
)
if context:
context.set_state("active_workspace", selected_workspace)
logger.debug(f"Cached workspace in context: {selected_workspace.tenant_id}")
return selected_workspace
async def get_active_project(
client: AsyncClient,
project: Optional[str] = None,
@@ -252,6 +346,7 @@ def add_project_metadata(result: str, project_name: str) -> str:
@asynccontextmanager
async def get_project_client(
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Optional[Context] = None,
) -> AsyncIterator[Tuple[AsyncClient, ProjectItem]]:
"""Resolve project, create correctly-routed client, and validate project.
@@ -263,6 +358,7 @@ async def get_project_client(
Args:
project: Optional explicit project parameter
workspace: Optional cloud workspace selector (tenant_id or unique name)
context: Optional FastMCP context for caching
Yields:
@@ -287,8 +383,28 @@ async def get_project_client(
f"Available projects: {project_names}"
)
# Step 2: Resolve project mode and optional workspace selection
config = ConfigManager().config
project_mode = config.get_project_mode(resolved_project)
active_workspace: WorkspaceInfo | None = None
# Trigger: workspace provided for a local project
# Why: workspace selection is a cloud routing concern only
# Outcome: fail fast with a deterministic guidance message
if project_mode != ProjectMode.CLOUD and workspace is not None:
raise ValueError(
f"Workspace '{workspace}' cannot be used with local project '{resolved_project}'. "
"Workspace selection is only supported for cloud-mode projects."
)
if project_mode == ProjectMode.CLOUD:
active_workspace = await resolve_workspace_parameter(workspace=workspace, context=context)
# Step 2: Create client routed based on project's mode
async with get_client(project_name=resolved_project) as client:
async with get_client(
project_name=resolved_project,
workspace=active_workspace.tenant_id if active_workspace else None,
) 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/__init__.py
+2
View File
@@ -23,6 +23,7 @@ from basic_memory.mcp.tools.canvas import canvas
from basic_memory.mcp.tools.list_directory import list_directory
from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.tools.move_note import move_note
from basic_memory.mcp.tools.workspaces import list_workspaces
from basic_memory.mcp.tools.project_management import (
list_memory_projects,
create_memory_project,
@@ -46,6 +47,7 @@ __all__ = [
"fetch",
"list_directory",
"list_memory_projects",
"list_workspaces",
"move_note",
"read_content",
"read_note",
src/basic_memory/mcp/tools/build_context.py
+2 -1
View File
@@ -196,6 +196,7 @@ def _format_context_markdown(graph: GraphContext, project: str) -> str:
async def build_context(
url: MemoryUrl,
project: Optional[str] = None,
workspace: Optional[str] = None,
depth: str | int | None = 1,
timeframe: Optional[TimeFrame] = "7d",
page: int = 1,
@@ -257,7 +258,7 @@ async def build_context(
# URL is already validated and normalized by MemoryUrl type annotation
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, context) as (client, active_project):
# Resolve memory:// identifier with project-prefix awareness
_, resolved_path, _ = await resolve_project_and_path(client, url, project, context)
src/basic_memory/mcp/tools/canvas.py
+2 -1
View File
@@ -23,6 +23,7 @@ async def canvas(
title: str,
directory: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> str:
"""Create an Obsidian canvas file with the provided nodes and edges.
@@ -93,7 +94,7 @@ async def canvas(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, 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
+2 -1
View File
@@ -151,6 +151,7 @@ async def delete_note(
identifier: str,
is_directory: bool = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> bool | str:
"""Delete a note or directory from the knowledge base.
@@ -215,7 +216,7 @@ async def delete_note(
with suggestions for finding the correct identifier, including search
commands and alternative formats to try.
"""
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, context) as (client, active_project):
logger.debug(
f"Deleting {'directory' if is_directory else 'note'}: {identifier} in project: {active_project.name}"
)
src/basic_memory/mcp/tools/edit_note.py
+2 -1
View File
@@ -131,6 +131,7 @@ async def edit_note(
operation: str,
content: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
section: Optional[str] = None,
find_text: Optional[str] = None,
expected_replacements: int = 1,
@@ -211,7 +212,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_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, 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 -1
View File
@@ -17,6 +17,7 @@ async def list_directory(
depth: int = 1,
file_name_glob: Optional[str] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> str:
"""List directory contents from the knowledge base with optional filtering.
@@ -61,7 +62,7 @@ async def list_directory(
Raises:
ToolError: If project doesn't exist or directory path is invalid
"""
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, 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
+2 -1
View File
@@ -348,6 +348,7 @@ async def move_note(
destination_path: str,
is_directory: bool = False,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> str:
"""Move a note or directory to a new location within the same project.
@@ -411,7 +412,7 @@ async def move_note(
- Re-indexes the entity for search
- Maintains all observations and relations
"""
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, context) as (client, active_project):
logger.debug(
f"Moving {'directory' if is_directory else 'note'}: {identifier} to {destination_path} in project: {active_project.name}"
)
src/basic_memory/mcp/tools/read_content.py
+5 -2
View File
@@ -150,7 +150,10 @@ def optimize_image(img, content_length, max_output_bytes=350000):
@mcp.tool(description="Read a file's raw content by path or permalink")
async def read_content(
path: str, project: Optional[str] = None, context: Context | None = None
path: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> dict:
"""Read a file's raw content by path or permalink.
@@ -201,7 +204,7 @@ async def read_content(
"""
logger.info("Reading file", path=path, project=project)
async with get_project_client(project, context) as (client, active_project):
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)
src/basic_memory/mcp/tools/read_note.py
+4 -1
View File
@@ -27,6 +27,7 @@ def _is_exact_title_match(identifier: str, title: str) -> bool:
async def read_note(
identifier: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
page: int = 1,
page_size: int = 10,
output_format: Literal["default", "ascii", "ansi"] = "default",
@@ -87,7 +88,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_project_client(project, context) as (client, active_project):
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)
@@ -148,6 +149,7 @@ async def read_note(
query=identifier,
search_type="title",
project=active_project.name,
workspace=workspace,
context=context,
)
@@ -198,6 +200,7 @@ async def read_note(
query=identifier,
search_type="text",
project=active_project.name,
workspace=workspace,
context=context,
)
src/basic_memory/mcp/tools/recent_activity.py
+5 -1
View File
@@ -40,6 +40,7 @@ async def recent_activity(
depth: int = 1,
timeframe: TimeFrame = "7d",
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> str:
"""Get recent activity for a specific project or across all projects.
@@ -246,7 +247,10 @@ 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, context) as (client, active_project):
async with get_project_client(resolved_project, workspace, context) as (
client,
active_project,
):
response = await call_get(
client,
f"/v2/projects/{active_project.external_id}/memory/recent",
src/basic_memory/mcp/tools/schema.py
+7 -8
View File
@@ -9,8 +9,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.schema import ValidationReport, InferenceReport, DriftReport
@@ -77,6 +76,7 @@ async def schema_validate(
note_type: Optional[str] = None,
identifier: Optional[str] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> ValidationReport | str:
"""Validate notes against their resolved schema.
@@ -115,8 +115,7 @@ async def schema_validate(
# Validate in a specific project
schema_validate(note_type="person", project="my-research")
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, workspace, context) as (client, active_project):
logger.info(
f"MCP tool call tool=schema_validate project={active_project.name} "
f"note_type={note_type} identifier={identifier}"
@@ -172,6 +171,7 @@ async def schema_infer(
note_type: str,
threshold: float = 0.25,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> InferenceReport | str:
"""Analyze existing notes and suggest a schema definition.
@@ -209,8 +209,7 @@ async def schema_infer(
# Infer in a specific project
schema_infer("person", project="my-research")
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, workspace, context) as (client, active_project):
logger.info(
f"MCP tool call tool=schema_infer project={active_project.name} "
f"note_type={note_type} threshold={threshold}"
@@ -272,6 +271,7 @@ async def schema_infer(
async def schema_diff(
note_type: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
context: Context | None = None,
) -> DriftReport | str:
"""Detect drift between a schema definition and actual note usage.
@@ -304,8 +304,7 @@ async def schema_diff(
# Check drift in a specific project
schema_diff("person", project="my-research")
"""
async with get_client() as client:
active_project = await get_active_project(client, project, context)
async with get_project_client(project, workspace, context) as (client, active_project):
logger.info(
f"MCP tool call tool=schema_diff project={active_project.name} note_type={note_type}"
)
src/basic_memory/mcp/tools/search.py
+4 -2
View File
@@ -250,6 +250,7 @@ Error searching for '{query}': {error_message}
async def search_notes(
query: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
page: int = 1,
page_size: int = 10,
search_type: str = "text",
@@ -421,7 +422,7 @@ async def search_notes(
types = types or []
entity_types = entity_types or []
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, context) as (client, active_project):
# Handle memory:// URLs by resolving to permalink search
_, resolved_query, is_memory_url = await resolve_project_and_path(
client, query, project, context
@@ -517,6 +518,7 @@ async def search_notes(
async def search_by_metadata(
filters: Dict[str, Any],
project: Optional[str] = None,
workspace: Optional[str] = None,
limit: int = 20,
offset: int = 0,
context: Context | None = None,
@@ -546,7 +548,7 @@ async def search_by_metadata(
page = (offset // limit) + 1
offset_within_page = offset % limit
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, 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/view_note.py
+9 -1
View File
@@ -16,6 +16,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,
page: int = 1,
page_size: int = 10,
context: Context | None = None,
@@ -57,7 +58,14 @@ async def view_note(
logger.info(f"Viewing note: {identifier} in project: {project}")
# Call the existing read_note logic
content = await read_note.fn(identifier, project, page, page_size, context)
content = await read_note.fn(
identifier=identifier,
project=project,
workspace=workspace,
page=page,
page_size=page_size,
context=context,
)
# Check if this is an error message (note not found)
if "# Note Not Found" in content:
src/basic_memory/mcp/tools/workspaces.py
+33
View File
@@ -0,0 +1,33 @@
"""Workspace discovery MCP tool."""
from fastmcp import Context
from basic_memory.mcp.project_context import get_available_workspaces
from basic_memory.mcp.server import mcp
@mcp.tool(description="List available cloud workspaces (tenant_id, type, role, and name).")
async def list_workspaces(context: Context | None = None) -> str:
"""List workspaces available to the current cloud user."""
workspaces = await get_available_workspaces(context=context)
if not workspaces:
return (
"# No Workspaces Available\n\n"
"No accessible workspaces were found for this account. "
"Ensure the account has an active subscription and tenant access."
)
lines = [
f"# Available Workspaces ({len(workspaces)})",
"",
"Use `workspace` as either the `tenant_id` or unique `name` in project-scoped tool calls.",
"",
]
for workspace in workspaces:
lines.append(
f"- {workspace.name} "
f"(type={workspace.workspace_type}, role={workspace.role}, tenant_id={workspace.tenant_id})"
)
return "\n".join(lines)
src/basic_memory/mcp/tools/write_note.py
+2 -1
View File
@@ -22,6 +22,7 @@ async def write_note(
content: str,
directory: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
tags: list[str] | str | None = None,
note_type: str = "note",
metadata: dict | None = None,
@@ -127,7 +128,7 @@ async def write_note(
HTTPError: If project doesn't exist or is inaccessible
SecurityError: If directory path attempts path traversal
"""
async with get_project_client(project, context) as (client, active_project):
async with get_project_client(project, workspace, context) as (client, active_project):
logger.info(
f"MCP tool call tool=write_note project={active_project.name} directory={directory}, title={title}, tags={tags}"
)
src/basic_memory/schemas/cloud.py
+25
View File
@@ -48,3 +48,28 @@ class CloudProjectCreateResponse(BaseModel):
new_project: dict | None = Field(
None, description="Information about the newly created project"
)
class WorkspaceInfo(BaseModel):
"""Workspace entry from /workspaces/ endpoint."""
tenant_id: str = Field(..., description="Workspace tenant identifier")
workspace_type: str = Field(..., description="Workspace type (personal or organization)")
name: str = Field(..., description="Workspace display name")
role: str = Field(..., description="Current user's role in the workspace")
organization_id: str | None = Field(None, description="Organization ID for org workspaces")
has_active_subscription: bool = Field(
default=False, description="Whether the workspace has an active subscription"
)
class WorkspaceListResponse(BaseModel):
"""Response from /workspaces/ endpoint."""
workspaces: list[WorkspaceInfo] = Field(
default_factory=list, description="Available workspaces"
)
count: int = Field(default=0, description="Number of available workspaces")
current_workspace_id: str | None = Field(
default=None, description="Current workspace tenant ID when available"
)
test-int/cli/test_project_commands_integration.py
+18 -11
View File
@@ -138,30 +138,37 @@ def test_remove_main_project(app, app_config, config_manager):
new_default_path = Path(new_default_dir)
# Ensure main exists
result = runner.invoke(cli_app, ["project", "list"], env=WIDE_TERMINAL_ENV)
if "main" not in result.stdout:
result = runner.invoke(cli_app, ["project", "add", "main", str(main_path)])
# Trigger: this test must work on Windows runners where output may contain "runneradmin".
# Why: substring checks against command output can mistake path text for project names.
# Outcome: use config state for setup decisions, then validate behavior via CLI invocation.
if "main" not in config_manager.config.projects:
result = runner.invoke(cli_app, ["project", "add", "main", str(main_path), "--local"])
print(result.stdout)
assert result.exit_code == 0
# Confirm main is present
result = runner.invoke(cli_app, ["project", "list"], env=WIDE_TERMINAL_ENV)
assert "main" in result.stdout
assert "main" in config_manager.config.projects
# Add a second project
result = runner.invoke(cli_app, ["project", "add", "new_default", str(new_default_path)])
result = runner.invoke(
cli_app, ["project", "add", "new_default", str(new_default_path), "--local"]
)
assert result.exit_code == 0
# Set new_default as default (if needed)
result = runner.invoke(cli_app, ["project", "default", "new_default"])
result = runner.invoke(cli_app, ["project", "default", "new_default", "--local"])
assert result.exit_code == 0
# Remove main
result = runner.invoke(cli_app, ["project", "remove", "main"])
result = runner.invoke(cli_app, ["project", "remove", "main", "--local"])
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
# Confirm only new_default exists and main does not
result = runner.invoke(cli_app, ["project", "list"], env=WIDE_TERMINAL_ENV)
result = runner.invoke(cli_app, ["project", "list", "--local"], env=WIDE_TERMINAL_ENV)
assert result.exit_code == 0
assert "main" not in result.stdout
assert "new_default" in result.stdout
config_after_list = config_manager.load_config()
assert "main" not in config_after_list.projects
assert "new_default" in config_after_list.projects
tests/cli/test_cli_tool_json_output.py
+20 -4
View File
@@ -175,6 +175,23 @@ def test_read_note_text_output(mock_mcp_read, mock_config_cls):
mock_mcp_read.fn.assert_called_once()
@patch("basic_memory.cli.commands.tool.ConfigManager")
@patch("basic_memory.cli.commands.tool.mcp_read_note")
def test_read_note_workspace_passthrough(mock_mcp_read, mock_config_cls):
"""read-note --workspace passes workspace through to the MCP tool call."""
mock_config_cls.return_value = _mock_config_manager()
mock_mcp_read.fn = AsyncMock(return_value="# Test Note")
result = runner.invoke(
cli_app,
["tool", "read-note", "test-note", "--workspace", "tenant-123"],
)
assert result.exit_code == 0, f"CLI failed: {result.output}"
mock_mcp_read.fn.assert_called_once()
assert mock_mcp_read.fn.call_args.kwargs["workspace"] == "tenant-123"
@patch("basic_memory.cli.commands.tool.ConfigManager")
@patch(
"basic_memory.cli.commands.tool._read_note_json",
@@ -387,10 +404,9 @@ def test_recent_activity_json_pagination(mock_recent_json):
assert isinstance(data, list)
# Verify pagination params were passed through
mock_recent_json.assert_called_once()
call_kwargs = mock_recent_json.call_args
# positional args: type, depth, timeframe, project_name, page, page_size
assert call_kwargs[0][4] == 2 # page
assert call_kwargs[0][5] == 10 # page_size
call_kwargs = mock_recent_json.call_args.kwargs
assert call_kwargs["page"] == 2
assert call_kwargs["page_size"] == 10
# --- build-context --format json ---
tests/cli/test_workspace_commands.py
+77
View File
@@ -0,0 +1,77 @@
"""Tests for workspace CLI commands."""
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.schemas.cloud import WorkspaceInfo
# Importing registers workspace commands on the shared app instance.
import basic_memory.cli.commands.workspace as workspace_cmd # noqa: F401
@pytest.fixture
def runner():
return CliRunner()
def test_workspace_list_prints_available_workspaces(runner, monkeypatch):
async def fake_get_available_workspaces(context=None):
return [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
),
WorkspaceInfo(
tenant_id="22222222-2222-2222-2222-222222222222",
workspace_type="organization",
name="Team",
role="editor",
),
]
monkeypatch.setattr(workspace_cmd, "get_available_workspaces", fake_get_available_workspaces)
result = runner.invoke(app, ["workspace", "list"])
assert result.exit_code == 0
assert "Available Workspaces" in result.stdout
assert "Personal" in result.stdout
assert "Team" in result.stdout
assert "11111111-1111-1111-1111-111111111111" in result.stdout
def test_workspaces_alias_matches_workspace_list_output(runner, monkeypatch):
async def fake_get_available_workspaces(context=None):
return [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
)
]
monkeypatch.setattr(workspace_cmd, "get_available_workspaces", fake_get_available_workspaces)
list_result = runner.invoke(app, ["workspace", "list"])
alias_result = runner.invoke(app, ["workspaces"])
assert list_result.exit_code == 0
assert alias_result.exit_code == 0
assert list_result.stdout == alias_result.stdout
def test_workspace_list_requires_oauth_login_message(runner, monkeypatch):
async def fail_get_available_workspaces(context=None): # pragma: no cover
raise RuntimeError("Workspace discovery requires OAuth login. Run 'bm cloud login' first.")
monkeypatch.setattr(workspace_cmd, "get_available_workspaces", fail_get_available_workspaces)
result = runner.invoke(app, ["workspace", "list"])
assert result.exit_code == 1
assert "Workspace discovery requires OAuth login" in result.stdout
assert "bm cloud login" in result.stdout
tests/mcp/test_async_client_modes.py
+67 -1
View File
@@ -6,7 +6,11 @@ import pytest
from basic_memory.cli.auth import CLIAuth
from basic_memory.config import ProjectMode
from basic_memory.mcp import async_client as async_client_module
from basic_memory.mcp.async_client import get_client, set_client_factory
from basic_memory.mcp.async_client import (
get_client,
get_cloud_control_plane_client,
set_client_factory,
)
@pytest.fixture(autouse=True)
@@ -62,6 +66,19 @@ async def test_get_client_explicit_cloud_uses_api_key(config_manager, monkeypatc
assert client.headers.get("Authorization") == "Bearer bmc_test_key_123"
@pytest.mark.asyncio
async def test_get_client_cloud_adds_workspace_header(config_manager):
cfg = config_manager.load_config()
cfg.cloud_host = "https://cloud.example.test"
cfg.cloud_api_key = "bmc_test_key_123"
cfg.set_project_mode("research", ProjectMode.CLOUD)
config_manager.save_config(cfg)
async with get_client(project_name="research", workspace="tenant-123") as client:
assert str(client.base_url).rstrip("/") == "https://cloud.example.test/proxy"
assert client.headers.get("X-Workspace-ID") == "tenant-123"
@pytest.mark.asyncio
async def test_get_client_explicit_cloud_raises_without_credentials(config_manager, monkeypatch):
cfg = config_manager.load_config()
@@ -219,3 +236,52 @@ async def test_get_client_explicit_cloud_overrides_local_project(config_manager,
async with get_client(project_name="main") as client:
assert str(client.base_url).rstrip("/") == "https://cloud.example.test/proxy"
assert client.headers.get("Authorization") == "Bearer bmc_test_key_123"
@pytest.mark.asyncio
async def test_get_cloud_control_plane_client_uses_api_key_when_available(config_manager):
cfg = config_manager.load_config()
cfg.cloud_host = "https://cloud.example.test"
cfg.cloud_api_key = "bmc_test_key_123"
cfg.cloud_client_id = "cid"
cfg.cloud_domain = "https://auth.example.test"
config_manager.save_config(cfg)
async with get_cloud_control_plane_client() as client:
assert str(client.base_url).rstrip("/") == "https://cloud.example.test"
assert client.headers.get("Authorization") == "Bearer bmc_test_key_123"
@pytest.mark.asyncio
async def test_get_cloud_control_plane_client_uses_oauth_token(config_manager):
cfg = config_manager.load_config()
cfg.cloud_host = "https://cloud.example.test"
cfg.cloud_api_key = None
cfg.cloud_client_id = "cid"
cfg.cloud_domain = "https://auth.example.test"
config_manager.save_config(cfg)
auth = CLIAuth(client_id=cfg.cloud_client_id, authkit_domain=cfg.cloud_domain)
auth.token_file.parent.mkdir(parents=True, exist_ok=True)
auth.token_file.write_text(
'{"access_token":"oauth-control-123","refresh_token":null,"expires_at":9999999999,"token_type":"Bearer"}',
encoding="utf-8",
)
async with get_cloud_control_plane_client() as client:
assert str(client.base_url).rstrip("/") == "https://cloud.example.test"
assert client.headers.get("Authorization") == "Bearer oauth-control-123"
@pytest.mark.asyncio
async def test_get_cloud_control_plane_client_raises_without_credentials(config_manager):
cfg = config_manager.load_config()
cfg.cloud_host = "https://cloud.example.test"
cfg.cloud_api_key = None
cfg.cloud_client_id = "cid"
cfg.cloud_domain = "https://auth.example.test"
config_manager.save_config(cfg)
with pytest.raises(RuntimeError, match="Cloud routing requested but no credentials found"):
async with get_cloud_control_plane_client():
pass
tests/mcp/test_project_context.py
+170
View File
@@ -9,6 +9,19 @@ from __future__ import annotations
import pytest
class _ContextState:
"""Minimal FastMCP context-state stub for unit tests."""
def __init__(self):
self._state: dict[str, object] = {}
def get_state(self, key: str):
return self._state.get(key)
def set_state(self, key: str, value: object) -> None:
self._state[key] = value
@pytest.mark.asyncio
async def test_returns_none_when_no_default_and_no_project(config_manager, monkeypatch):
from basic_memory.mcp.project_context import resolve_project_parameter
@@ -104,3 +117,160 @@ async def test_env_constraint_overrides_default(config_manager, config_home, mon
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "env-project")
assert await resolve_project_parameter(project=None) == "env-project"
@pytest.mark.asyncio
async def test_workspace_auto_selects_single_and_caches(monkeypatch):
from basic_memory.mcp.project_context import resolve_workspace_parameter
from basic_memory.schemas.cloud import WorkspaceInfo
context = _ContextState()
only_workspace = WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
)
async def fake_get_available_workspaces(context=None):
return [only_workspace]
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fake_get_available_workspaces,
)
resolved = await resolve_workspace_parameter(context=context)
assert resolved.tenant_id == only_workspace.tenant_id
assert context.get_state("active_workspace") == only_workspace
@pytest.mark.asyncio
async def test_workspace_requires_user_choice_when_multiple(monkeypatch):
from basic_memory.mcp.project_context import resolve_workspace_parameter
from basic_memory.schemas.cloud import WorkspaceInfo
workspaces = [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
),
WorkspaceInfo(
tenant_id="22222222-2222-2222-2222-222222222222",
workspace_type="organization",
name="Team",
role="editor",
),
]
async def fake_get_available_workspaces(context=None):
return workspaces
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fake_get_available_workspaces,
)
with pytest.raises(ValueError, match="Multiple workspaces are available"):
await resolve_workspace_parameter(context=_ContextState())
@pytest.mark.asyncio
async def test_workspace_explicit_selection_by_tenant_id_or_name(monkeypatch):
from basic_memory.mcp.project_context import resolve_workspace_parameter
from basic_memory.schemas.cloud import WorkspaceInfo
team_workspace = WorkspaceInfo(
tenant_id="22222222-2222-2222-2222-222222222222",
workspace_type="organization",
name="Team",
role="editor",
)
workspaces = [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
),
team_workspace,
]
async def fake_get_available_workspaces(context=None):
return workspaces
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fake_get_available_workspaces,
)
resolved_by_id = await resolve_workspace_parameter(workspace=team_workspace.tenant_id)
assert resolved_by_id.tenant_id == team_workspace.tenant_id
resolved_by_name = await resolve_workspace_parameter(workspace="team")
assert resolved_by_name.tenant_id == team_workspace.tenant_id
@pytest.mark.asyncio
async def test_workspace_invalid_selection_lists_choices(monkeypatch):
from basic_memory.mcp.project_context import resolve_workspace_parameter
from basic_memory.schemas.cloud import WorkspaceInfo
workspaces = [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
)
]
async def fake_get_available_workspaces(context=None):
return workspaces
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fake_get_available_workspaces,
)
with pytest.raises(ValueError, match="Workspace 'missing-workspace' was not found"):
await resolve_workspace_parameter(workspace="missing-workspace")
@pytest.mark.asyncio
async def test_workspace_uses_cached_workspace_without_fetch(monkeypatch):
from basic_memory.mcp.project_context import resolve_workspace_parameter
from basic_memory.schemas.cloud import WorkspaceInfo
cached_workspace = WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
)
context = _ContextState()
context.set_state("active_workspace", cached_workspace)
async def fail_if_called(context=None): # pragma: no cover
raise AssertionError("Workspace fetch should not run when cache is available")
monkeypatch.setattr(
"basic_memory.mcp.project_context.get_available_workspaces",
fail_if_called,
)
resolved = await resolve_workspace_parameter(context=context)
assert resolved.tenant_id == cached_workspace.tenant_id
@pytest.mark.asyncio
async def test_get_project_client_rejects_workspace_for_local_project():
from basic_memory.mcp.project_context import get_project_client
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
tests/mcp/test_tool_contracts.py
+30 -10
View File
@@ -11,6 +11,7 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"build_context": [
"url",
"project",
"workspace",
"depth",
"timeframe",
"page",
@@ -18,33 +19,39 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"max_related",
"format",
],
"canvas": ["nodes", "edges", "title", "directory", "project"],
"canvas": ["nodes", "edges", "title", "directory", "project", "workspace"],
"cloud_info": [],
"create_memory_project": ["project_name", "project_path", "set_default"],
"delete_note": ["identifier", "is_directory", "project"],
"delete_note": ["identifier", "is_directory", "project", "workspace"],
"delete_project": ["project_name"],
"edit_note": [
"identifier",
"operation",
"content",
"project",
"workspace",
"section",
"find_text",
"expected_replacements",
],
"fetch": ["id"],
"list_directory": ["dir_name", "depth", "file_name_glob", "project"],
"list_directory": ["dir_name", "depth", "file_name_glob", "project", "workspace"],
"list_memory_projects": [],
"move_note": ["identifier", "destination_path", "is_directory", "project"],
"read_content": ["path", "project"],
"read_note": ["identifier", "project", "page", "page_size", "output_format"],
"list_workspaces": [],
"move_note": ["identifier", "destination_path", "is_directory", "project", "workspace"],
"read_content": ["path", "project", "workspace"],
"read_note": ["identifier", "project", "workspace", "page", "page_size", "output_format"],
"release_notes": [],
"recent_activity": ["type", "depth", "timeframe", "project"],
"recent_activity": ["type", "depth", "timeframe", "project", "workspace"],
"schema_diff": ["note_type", "project", "workspace"],
"schema_infer": ["note_type", "threshold", "project", "workspace"],
"schema_validate": ["note_type", "identifier", "project", "workspace"],
"search": ["query"],
"search_by_metadata": ["filters", "project", "limit", "offset"],
"search_by_metadata": ["filters", "project", "workspace", "limit", "offset"],
"search_notes": [
"query",
"project",
"workspace",
"page",
"page_size",
"search_type",
@@ -57,8 +64,17 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"status",
"min_similarity",
],
"view_note": ["identifier", "project", "page", "page_size"],
"write_note": ["title", "content", "directory", "project", "tags", "note_type", "metadata"],
"view_note": ["identifier", "project", "workspace", "page", "page_size"],
"write_note": [
"title",
"content",
"directory",
"project",
"workspace",
"tags",
"note_type",
"metadata",
],
}
@@ -73,11 +89,15 @@ TOOL_FUNCTIONS: dict[str, object] = {
"fetch": tools.fetch,
"list_directory": tools.list_directory,
"list_memory_projects": tools.list_memory_projects,
"list_workspaces": tools.list_workspaces,
"move_note": tools.move_note,
"read_content": tools.read_content,
"read_note": tools.read_note,
"release_notes": tools.release_notes,
"recent_activity": tools.recent_activity,
"schema_diff": tools.schema_diff,
"schema_infer": tools.schema_infer,
"schema_validate": tools.schema_validate,
"search": tools.search,
"search_by_metadata": tools.search_by_metadata,
"search_notes": tools.search_notes,
tests/mcp/test_tool_workspace_management.py
+107
View File
@@ -0,0 +1,107 @@
"""Tests for workspace MCP tools."""
import pytest
from basic_memory.mcp.tools.workspaces import list_workspaces
from basic_memory.schemas.cloud import WorkspaceInfo
class _ContextState:
def __init__(self):
self._state: dict[str, object] = {}
def get_state(self, key: str):
return self._state.get(key)
def set_state(self, key: str, value: object) -> None:
self._state[key] = value
@pytest.mark.asyncio
async def test_list_workspaces_formats_workspace_rows(monkeypatch):
async def fake_get_available_workspaces(context=None):
return [
WorkspaceInfo(
tenant_id="11111111-1111-1111-1111-111111111111",
workspace_type="personal",
name="Personal",
role="owner",
),
WorkspaceInfo(
tenant_id="22222222-2222-2222-2222-222222222222",
workspace_type="organization",
name="Team",
role="editor",
),
]
monkeypatch.setattr(
"basic_memory.mcp.tools.workspaces.get_available_workspaces",
fake_get_available_workspaces,
)
result = await list_workspaces.fn()
assert "# Available Workspaces (2)" in result
assert "Personal (type=personal, role=owner" in result
assert "Team (type=organization, role=editor" in result
@pytest.mark.asyncio
async def test_list_workspaces_handles_empty_list(monkeypatch):
async def fake_get_available_workspaces(context=None):
return []
monkeypatch.setattr(
"basic_memory.mcp.tools.workspaces.get_available_workspaces",
fake_get_available_workspaces,
)
result = await list_workspaces.fn()
assert "# No Workspaces Available" in result
@pytest.mark.asyncio
async def test_list_workspaces_oauth_error_bubbles_up(monkeypatch):
async def fake_get_available_workspaces(context=None):
raise RuntimeError("Workspace discovery requires OAuth login. Run 'bm cloud login' first.")
monkeypatch.setattr(
"basic_memory.mcp.tools.workspaces.get_available_workspaces",
fake_get_available_workspaces,
)
with pytest.raises(RuntimeError, match="Workspace discovery requires OAuth login"):
await list_workspaces.fn()
@pytest.mark.asyncio
async def test_list_workspaces_uses_context_cache_path(monkeypatch):
context = _ContextState()
call_count = {"fetches": 0}
workspace = WorkspaceInfo(
tenant_id="33333333-3333-3333-3333-333333333333",
workspace_type="personal",
name="Cached",
role="owner",
)
async def fake_get_available_workspaces(context=None):
assert context is not None
cached = context.get_state("available_workspaces")
if cached:
return cached
call_count["fetches"] += 1
context.set_state("available_workspaces", [workspace])
return [workspace]
monkeypatch.setattr(
"basic_memory.mcp.tools.workspaces.get_available_workspaces",
fake_get_available_workspaces,
)
first = await list_workspaces.fn(context=context)
second = await list_workspaces.fn(context=context)
assert "# Available Workspaces (1)" in first
assert "# Available Workspaces (1)" in second
assert call_count["fetches"] == 1