Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
442 lines (339 loc) · 12.3 KB

ARCHITECTURE.md

File metadata and controls

442 lines (339 loc) · 12.3 KB

Basic Memory Architecture

This document describes the architectural patterns and composition structure of Basic Memory.

Overview

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.

Composition Roots

What is a Composition Root?

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:

  1. Reads configuration from ConfigManager
  2. Resolves runtime mode (local/test)
  3. Creates and provides dependencies to downstream code

Key principle: Only composition roots read global configuration. All other modules receive configuration explicitly.

Container Structure

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.py            # RuntimeMode enum and resolver

Container Pattern

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 = container

Runtime Mode Resolution

The 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.TEST

Resolution 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.LOCAL

Note: 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.

Dependencies Package

Structure

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

Usage in Routers

from basic_memory.deps.services import get_entity_service
from basic_memory.deps.projects import get_project_config

@router.get("/entities/{id}")
async def get_entity(
    id: int,
    entity_service: EntityService = Depends(get_entity_service),
    project: ProjectConfig = Depends(get_project_config),
):
    return await entity_service.get(id)

Backwards Compatibility

The old deps.py file still exists as a thin re-export shim:

# deps.py - backwards compatibility shim
from basic_memory.deps import *

New code should import from specific submodules (basic_memory.deps.services) for clarity.

MCP Tools Architecture

Typed API Clients

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

Client Pattern

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())

Tool → Client → API Flow

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())

Per-Project Client Routing

get_project_client() from mcp/project_context.py is an async context manager that:

  1. Resolves the project name from config (no network call)
  2. Creates the correctly-routed client based on the project's mode (local ASGI or cloud HTTP with API key)
  3. Validates the project via the API
  4. 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
    ...

Sync Coordination

SyncCoordinator

The SyncCoordinator centralizes sync/watch lifecycle management:

@dataclass
class SyncCoordinator:
    """Coordinates file sync and watch operations."""

    status: SyncStatus = SyncStatus.NOT_STARTED
    sync_task: asyncio.Task | None = None
    watch_service: WatchService | None = None

    async def start(self, ...):
        """Start sync and watch operations."""

    async def stop(self):
        """Stop all sync operations gracefully."""

    def get_status_info(self) -> dict:
        """Get current sync status for observability."""

Status Enum

class SyncStatus(Enum):
    NOT_STARTED = "not_started"
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    STOPPED = "stopped"
    ERROR = "error"

Project Resolution

ProjectResolver

Unified 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
        """

Resolution Modes

class ResolutionMode(Enum):
    EXPLICIT = "explicit"           # User specified project
    DEFAULT = "default"             # Using configured default
    SINGLE_PROJECT = "single"       # Only one project exists
    FALLBACK = "fallback"           # Using first available

Testing Patterns

Container Testing

Each 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)

Mocking Typed Clients

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)

Design Principles

1. Explicit Dependencies

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 coupling

2. Single Responsibility

Each layer has a clear responsibility:

  • Containers: Wire dependencies
  • Clients: Encapsulate HTTP communication
  • Services: Business logic
  • Repositories: Data access
  • Tools/Routers: Thin adapters

3. Deferred Imports

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)

4. Backwards Compatibility

When refactoring, maintain backwards compatibility via shims:

# Old module becomes a shim
from basic_memory.new_location import *

# Docstring explains migration path
"""
DEPRECATED: Import from basic_memory.new_location instead.
This shim will be removed in a future version.
"""

File Organization

src/basic_memory/
├── api/
│   ├── container.py          # API composition root
│   ├── routers/              # FastAPI routers
│   └── ...
├── 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
├── sync/
│   ├── coordinator.py        # SyncCoordinator
│   └── ...
├── runtime.py                # RuntimeMode resolution
├── project_resolver.py       # Unified project selection
└── config.py                 # Configuration management