This document describes the architectural patterns and composition structure of Basic Memory.
Basic Memory is a local-first knowledge management system with three entrypoints:
- API - FastAPI REST server for HTTP access
- MCP - Model Context Protocol server for LLM integration
- CLI - Typer command-line interface
Each entrypoint uses a composition root pattern to manage configuration and dependencies.
A composition root is the single place in an application where dependencies are wired together. In Basic Memory, each entrypoint has its own composition root that:
- Reads configuration from
ConfigManager - Resolves runtime mode (local/test)
- Creates and provides dependencies to downstream code
Key principle: Only composition roots read global configuration. All other modules receive configuration explicitly.
Each entrypoint has a container dataclass in its package:
src/basic_memory/
├── api/
│ └── container.py # ApiContainer
├── mcp/
│ └── container.py # McpContainer
├── cli/
│ └── container.py # CliContainer
└── runtime/
└── mode.py # RuntimeMode enum and resolver
All containers follow the same structure:
@dataclass
class Container:
config: BasicMemoryConfig
mode: RuntimeMode
@classmethod
def create(cls) -> "Container":
"""Create container by reading ConfigManager."""
config = ConfigManager().config
mode = resolve_runtime_mode(is_test_env=config.is_test_env)
return cls(config=config, mode=mode)
@property
def some_computed_property(self) -> bool:
"""Derived values based on config and mode."""
return self.mode.is_local and self.config.some_setting
# Module-level singleton
_container: Container | None = None
def get_container() -> Container:
if _container is None:
raise RuntimeError("Container not initialized")
return _container
def set_container(container: Container) -> None:
global _container
_container = containerThe RuntimeMode enum centralizes mode detection:
class RuntimeMode(Enum):
LOCAL = "local"
CLOUD = "cloud"
TEST = "test"
@property
def is_cloud(self) -> bool:
return self == RuntimeMode.CLOUD
@property
def is_local(self) -> bool:
return self == RuntimeMode.LOCAL
@property
def is_test(self) -> bool:
return self == RuntimeMode.TESTResolution follows this precedence in local app flows: TEST > LOCAL
def resolve_runtime_mode(is_test_env: bool) -> RuntimeMode:
if is_test_env:
return RuntimeMode.TEST
return RuntimeMode.LOCALNote: RuntimeMode determines global behavior (e.g., whether to start file sync).
Per-project routing is orthogonal: individual projects can be set to cloud mode via ProjectMode,
which affects client routing in get_client(project_name=...) without changing global runtime mode.
RuntimeMode.CLOUD may remain for compatibility, but standard local runtime resolution does not select it.
The deps/ package provides FastAPI dependencies organized by feature:
src/basic_memory/deps/
├── __init__.py # Re-exports for backwards compatibility
├── config.py # Configuration access
├── db.py # Database/session management
├── projects.py # Project resolution
├── repositories.py # Data access layer
├── services.py # Business logic layer
└── importers.py # Import functionality
Projects are addressed by external UUID in the URL path; there is one provider
per service (the name-based v1 and integer-id _v2 tiers were removed, #1109):
from basic_memory.deps import EntityServiceV2ExternalDep, ProjectConfigV2ExternalDep
@router.get("/entities/{entity_id}")
async def get_entity(
entity_id: str,
entity_service: EntityServiceV2ExternalDep,
project: ProjectConfigV2ExternalDep,
):
return await entity_service.get_by_external_id(entity_id)get_app_config resolves config from the composition root, never from a
ConfigManager() read of its own:
- API requests read the container the lifespan stored on
app.state. - Requests served without a lifespan (the CLI/MCP local ASGI flow) fall back to
api.container.resolve_container(), which returns the installed container or creates a fresh one — theConfigManagerread stays insideApiContainer.create().
Tests inject config by overriding the provider:
app.dependency_overrides[get_app_config] = lambda: app_config.
Accepted-note orchestration is shared by local and hosted runtimes. Its dependencies flow in one direction:
schemas and runtime values
↓
repositories
↓
portable services and indexing workflows
↓
local or hosted composition roots and adapters
repository/owns explicit-session persistence operations and persisted row values, including accepted-note search rows and vector cleanup. It must not importindexing/workflows.services/owns runtime-neutral note preparation, note-content reads and writes, and delete operations. These modules receive storage and repository capabilities explicitly.indexing/owns portable mutation, reconciliation, materialization, and project-index workflows.index/contains the local runtime's concrete adapters and composition helpers. Shared contracts such as project-index requests and scheduler capabilities live in neutral modules within this package; hosted code must not depend onLocal*implementations.
Accepted Markdown create, replace, and edit operations cross one public persistence boundary:
persist_accepted_note_snapshot. That operation writes the entity snapshot, NoteContent, graph
rows, and entity search row inside the caller's transaction. Move intentionally uses the narrower
persist_accepted_note_move operation because changing a path must not replace observations or
relations.
MCP tools communicate with the API through typed clients that encapsulate HTTP paths and response validation:
src/basic_memory/mcp/clients/
├── __init__.py # Re-exports all clients
├── base.py # BaseClient with common logic
├── knowledge.py # KnowledgeClient - entity CRUD
├── search.py # SearchClient - search operations
├── memory.py # MemoryClient - context building
├── directory.py # DirectoryClient - directory listing
├── resource.py # ResourceClient - resource reading
└── project.py # ProjectClient - project management
Each client encapsulates API paths and validates responses:
class KnowledgeClient(BaseClient):
"""Client for knowledge/entity operations."""
async def resolve_entity(self, identifier: str) -> int:
"""Resolve identifier to entity ID."""
response = await call_get(
self.http_client,
f"{self._base_path}/resolve/{identifier}",
)
return int(response.text)
async def get_entity(self, entity_id: int) -> EntityResponse:
"""Get entity by ID."""
response = await call_get(
self.http_client,
f"{self._base_path}/entities/{entity_id}",
)
return EntityResponse.model_validate(response.json())MCP Tool (thin adapter)
↓
Typed Client (encapsulates paths, validates responses)
↓
HTTP API (FastAPI router)
↓
Service Layer (business logic)
↓
Repository Layer (data access)
Example tool using typed client:
@mcp.tool()
async def search_notes(
query: str,
project: str | None = None,
metadata_filters: dict | None = None,
tags: list[str] | None = None,
status: str | None = None,
) -> SearchResponse:
async with get_project_client(project, context) as (client, active_project):
# Import client inside function to avoid circular imports
from basic_memory.mcp.clients import SearchClient
from basic_memory.schemas.search import SearchQuery
search_query = SearchQuery(
text=query,
metadata_filters=metadata_filters,
tags=tags,
status=status,
)
search_client = SearchClient(client, active_project.external_id)
return await search_client.search(search_query.model_dump())get_project_client() from mcp/project_context.py is an async context manager that:
- Resolves the project name from config (no network call)
- Creates the correctly-routed client based on the project's mode (local ASGI or cloud HTTP with API key)
- Validates the project via the API
- Yields
(client, active_project)tuple
This solves the bootstrap problem: you need the project name to choose the right client (local vs cloud), but you need the client to validate the project exists.
from basic_memory.mcp.project_context import get_project_client
async with get_project_client(project, context) as (client, active_project):
# client is routed based on project's mode (local or cloud)
# active_project is validated via the API
...The WatchCoordinator (index/watch_coordinator.py) owns the local event-index
watcher lifecycle. The API composition root creates it
(ApiContainer.create_watch_coordinator()), the lifespan starts and stops it, and
should_watch/skip_reason come from the container's runtime-mode logic — watching
is skipped in test and cloud modes and when index_changes is disabled.
@dataclass
class WatchCoordinator:
"""Coordinate local event-index watcher lifecycle across entrypoints."""
config: BasicMemoryConfig
should_watch: bool = True
skip_reason: str | None = NoneUnified project selection across all entrypoints:
class ProjectResolver:
"""Resolves which project to use based on context."""
def resolve(
self,
explicit_project: str | None = None,
) -> ResolvedProject:
"""Resolve project using three-tier hierarchy:
1. Explicit project parameter
2. Default project from config
3. Single available project
"""class ResolutionMode(Enum):
EXPLICIT = "explicit" # User specified project
DEFAULT = "default" # Using configured default
SINGLE_PROJECT = "single" # Only one project exists
FALLBACK = "fallback" # Using first availableEach container has corresponding tests:
tests/
├── api/test_api_container.py
├── mcp/test_mcp_container.py
└── cli/test_cli_container.py
Tests verify:
- Container creation from config
- Runtime mode properties
- Container accessor functions (get/set)
When testing MCP tools, mock at the client level:
def test_search_notes(monkeypatch):
import basic_memory.mcp.clients as clients_mod
class MockSearchClient:
async def search(self, query):
return SearchResponse(results=[...])
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)Modules receive configuration explicitly rather than reading globals:
# Good - explicit injection
async def sync_files(config: BasicMemoryConfig):
...
# Avoid - hidden global access
async def sync_files():
config = ConfigManager().config # Hidden couplingEach layer has a clear responsibility:
- Containers: Wire dependencies
- Clients: Encapsulate HTTP communication
- Services: Business logic
- Repositories: Data access
- Tools/Routers: Thin adapters
To avoid circular imports, typed clients are imported inside functions:
async def my_tool():
async with get_client() as client:
# Import here to avoid circular dependency
from basic_memory.mcp.clients import KnowledgeClient
knowledge_client = KnowledgeClient(client, project_id)When refactoring, a re-export shim may bridge a move for downstream callers — but
shims are traps for readers once callers migrate, so they must be deleted promptly
(the 2026-07 post-seam cleanup removed deps.py, basic_memory.cloud, and
create_client(), #1107). A shim ships with its removal plan:
"""
DEPRECATED: Import from basic_memory.new_location instead.
Delete this shim once <downstream repo/release> has migrated.
"""
from basic_memory.new_location import *src/basic_memory/
├── api/
│ ├── container.py # API composition root
│ ├── app.py # FastAPI app + lifespan
│ └── v2/routers/ # v2 routers (the only public API surface)
├── mcp/
│ ├── container.py # MCP composition root
│ ├── clients/ # Typed API clients
│ ├── tools/ # MCP tool definitions
│ └── server.py # MCP server setup
├── cli/
│ ├── container.py # CLI composition root
│ ├── app.py # Typer app
│ └── commands/ # CLI command groups
├── deps/
│ ├── config.py # Config dependencies
│ ├── db.py # Database dependencies
│ ├── projects.py # Project dependencies
│ ├── repositories.py # Repository dependencies
│ ├── services.py # Service dependencies
│ └── importers.py # Importer dependencies
├── index/ # Local runtime adapters + WatchCoordinator
├── indexing/ # Portable indexing runners and planners
├── picoschema/ # Picoschema parsing, resolution, validation, inference, and drift
├── runtime/ # RuntimeMode + runtime Protocol contracts
├── project_resolver.py # Unified project selection
└── config.py # Configuration management