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 composition roots (api/container.py, mcp/container.py, cli/container.py)

Browse Source 
Adds composition root containers for API, MCP, and CLI entrypoints to
centralize configuration loading, logging initialization, and runtime
mode selection. This reduces coupling between modules and runtime concerns.

Changes:
- Add api/container.py: API composition root with config + logging
- Add mcp/container.py: MCP composition root with config + logging + client factories
- Add cli/container.py: CLI composition root with config + logging + client factories
- Refactor api/app.py to use APIContainer.create()
- Refactor mcp/server.py to use MCPContainer.create()
- Refactor mcp/async_client.py to delegate to MCPContainer for runtime mode

Containers own:
- Reading ConfigManager + environment variables
- Selecting runtime mode (cloud/local/test)
- Providing factories (httpx clients)
- Initializing logging for each entrypoint

This is foundational work that enables further refactoring to remove
direct ConfigManager access from downstream modules.

Co-authored-by: Paul Hernandez <phernandez@users.noreply.github.com>
Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
This commit is contained in:
claude[bot]
2025-12-31 22:04:51 +00:00
parent 4ce21984a4
commit 08c9273328
6 changed files with 374 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/api/app.py
+5 -5
View File
@@ -30,7 +30,7 @@ from basic_memory.api.v2.routers import (
prompt_router as v2_prompt,
importer_router as v2_importer,
)
from basic_memory.config import ConfigManager, init_api_logging
from basic_memory.api.container import APIContainer
from basic_memory.services.initialization import initialize_file_sync, initialize_app
@@ -38,10 +38,10 @@ from basic_memory.services.initialization import initialize_file_sync, initializ
async def lifespan(app: FastAPI): # pragma: no cover
"""Lifecycle manager for the FastAPI app. Not called in stdio mcp mode"""
# Initialize logging for API (stdout in cloud mode, file otherwise)
init_api_logging()
app_config = ConfigManager().config
# --- Composition Root ---
# Container handles: config loading, logging init, runtime mode selection
container = APIContainer.create()
app_config = container.config
logger.info("Starting Basic Memory API")
await initialize_app(app_config)
src/basic_memory/api/container.py
+69
View File
@@ -0,0 +1,69 @@
"""Composition root for API entrypoint.
This module owns:
- Reading ConfigManager + environment variables
- Selecting runtime mode (cloud/local/test)
- Providing factories (httpx clients, repositories, services)
- Initializing logging for API
This centralizes composition concerns and reduces coupling between
modules and runtime environment decisions.
"""
from dataclasses import dataclass
from basic_memory.config import BasicMemoryConfig, ConfigManager, init_api_logging
@dataclass
class APIContainer:
"""Composition root for API entrypoint.
Responsibilities:
- Configuration loading and caching
- Logging initialization
- Runtime mode determination
The container is built once at startup and provides access to
configuration throughout the API lifecycle.
"""
config: BasicMemoryConfig
@classmethod
def create(cls) -> "APIContainer":
"""Build container with all dependencies.
This is the single point where we:
1. Initialize logging for API mode
2. Load configuration from file + environment
3. Determine runtime mode (cloud/local/test)
Returns:
Configured APIContainer ready for use
"""
# Initialize logging first (cloud mode: stdout, local: file)
init_api_logging()
# Load configuration (merges file + environment variables)
config = ConfigManager().config
return cls(config=config)
@property
def is_cloud_mode(self) -> bool:
"""Check if running in cloud mode.
Returns:
True if cloud mode enabled via env var or config file
"""
return self.config.cloud_mode_enabled
@property
def is_test_env(self) -> bool:
"""Check if running in test environment.
Returns:
True if test environment detected
"""
return self.config.is_test_env
src/basic_memory/cli/container.py
+141
View File
@@ -0,0 +1,141 @@
"""Composition root for CLI entrypoint.
This module owns:
- Reading ConfigManager + environment variables
- Selecting runtime mode (cloud/local/test)
- Providing factories (httpx clients, repositories, services)
- Initializing logging for CLI
This centralizes composition concerns and reduces coupling between
modules and runtime environment decisions.
"""
from contextlib import asynccontextmanager, AbstractAsyncContextManager
from dataclasses import dataclass
from typing import AsyncIterator, Callable, Optional
from httpx import ASGITransport, AsyncClient, Timeout
from loguru import logger
from basic_memory.config import BasicMemoryConfig, ConfigManager, init_cli_logging
@dataclass
class CLIContainer:
"""Composition root for CLI entrypoint.
Responsibilities:
- Configuration loading and caching
- Logging initialization
- Runtime mode determination
- HTTP client factory provision
The container is built once at CLI startup and provides factories
for creating properly configured dependencies.
"""
config: BasicMemoryConfig
@classmethod
def create(cls) -> "CLIContainer":
"""Build container with all dependencies.
This is the single point where we:
1. Initialize logging for CLI mode (file only, no stdout)
2. Load configuration from file + environment
3. Determine runtime mode (cloud/local/test)
Returns:
Configured CLIContainer ready for use
"""
# Initialize logging first (CLI: file only, avoid interfering with output)
init_cli_logging()
# Load configuration (merges file + environment variables)
config = ConfigManager().config
return cls(config=config)
@property
def is_cloud_mode(self) -> bool:
"""Check if running in cloud mode.
Returns:
True if cloud mode enabled via env var or config file
"""
return self.config.cloud_mode_enabled
@property
def is_test_env(self) -> bool:
"""Check if running in test environment.
Returns:
True if test environment detected
"""
return self.config.is_test_env
def get_client_factory(
self, override_factory: Optional[Callable[[], AbstractAsyncContextManager[AsyncClient]]] = None
) -> Callable[[], AbstractAsyncContextManager[AsyncClient]]:
"""Get HTTP client factory based on runtime mode.
Priority order:
1. Override factory (for dependency injection in tests)
2. Cloud mode: HTTP client with auth
3. Local mode: ASGI transport for in-process calls
Args:
override_factory: Optional factory override for testing
Returns:
Async context manager factory that yields configured AsyncClient
"""
if override_factory:
return override_factory
# Return a factory that creates clients based on current mode
@asynccontextmanager
async def _client_factory() -> AsyncIterator[AsyncClient]:
"""Factory that creates client based on runtime mode."""
timeout = Timeout(
connect=10.0, # 10 seconds for connection
read=30.0, # 30 seconds for reading response
write=30.0, # 30 seconds for writing request
pool=30.0, # 30 seconds for connection pool
)
if self.is_cloud_mode:
# Cloud mode: inject auth when creating client
from basic_memory.cli.auth import CLIAuth
auth = CLIAuth(
client_id=self.config.cloud_client_id, authkit_domain=self.config.cloud_domain
)
token = await auth.get_valid_token()
if not token:
raise RuntimeError(
"Cloud mode enabled but not authenticated. "
"Run 'basic-memory cloud login' first."
)
# Auth header set ONCE at client creation
proxy_base_url = f"{self.config.cloud_host}/proxy"
logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")
async with AsyncClient(
base_url=proxy_base_url,
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
) as client:
yield client
else:
# Local mode: ASGI transport for in-process calls
from basic_memory.api.app import app as fastapi_app
logger.info("Creating 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
return _client_factory
src/basic_memory/mcp/async_client.py
+13 -42
View File
@@ -57,50 +57,21 @@ async def get_client() -> AsyncIterator[AsyncClient]:
Raises:
RuntimeError: If cloud mode is enabled but user is not authenticated
"""
if _client_factory:
# Use injected factory (cloud app, tests)
async with _client_factory() as client:
yield client
else:
# Default: create based on config
config = ConfigManager().config
timeout = Timeout(
connect=10.0, # 10 seconds for connection
read=30.0, # 30 seconds for reading response
write=30.0, # 30 seconds for writing request
pool=30.0, # 30 seconds for connection pool
)
# --- Composition Root Pattern ---
# Delegate to container for runtime mode selection
# Note: We create container each time but skip logging init since that's
# already done in the entrypoint (MCP server lifespan or CLI command)
from basic_memory.config import ConfigManager
from basic_memory.mcp.container import MCPContainer
if config.cloud_mode_enabled:
# CLI cloud mode: inject auth when creating client
from basic_memory.cli.auth import CLIAuth
# Create lightweight container without re-initializing logging
# Logging is initialized once at entrypoint startup
config = ConfigManager().config
container = MCPContainer(config=config)
factory = container.get_client_factory(override_factory=_client_factory)
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
token = await auth.get_valid_token()
if not token:
raise RuntimeError(
"Cloud mode enabled but not authenticated. "
"Run 'basic-memory cloud login' first."
)
# Auth header set ONCE at client creation
proxy_base_url = f"{config.cloud_host}/proxy"
logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")
async with AsyncClient(
base_url=proxy_base_url,
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
) as client:
yield client
else:
# Local mode: ASGI transport for in-process calls
# Note: ASGI transport does NOT trigger FastAPI lifespan, so no special handling needed
logger.info("Creating 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
async with factory() as client:
yield client
def create_client() -> AsyncClient:
src/basic_memory/mcp/container.py
+141
View File
@@ -0,0 +1,141 @@
"""Composition root for MCP entrypoint.
This module owns:
- Reading ConfigManager + environment variables
- Selecting runtime mode (cloud/local/test)
- Providing factories (httpx clients, repositories, services)
- Initializing logging for MCP
This centralizes composition concerns and reduces coupling between
modules and runtime environment decisions.
"""
from contextlib import asynccontextmanager, AbstractAsyncContextManager
from dataclasses import dataclass
from typing import AsyncIterator, Callable, Optional
from httpx import ASGITransport, AsyncClient, Timeout
from loguru import logger
from basic_memory.config import BasicMemoryConfig, ConfigManager, init_mcp_logging
@dataclass
class MCPContainer:
"""Composition root for MCP entrypoint.
Responsibilities:
- Configuration loading and caching
- Logging initialization
- Runtime mode determination
- HTTP client factory provision
The container is built once at startup and provides factories
for creating properly configured dependencies.
"""
config: BasicMemoryConfig
@classmethod
def create(cls) -> "MCPContainer":
"""Build container with all dependencies.
This is the single point where we:
1. Initialize logging for MCP mode (file only, no stdout)
2. Load configuration from file + environment
3. Determine runtime mode (cloud/local/test)
Returns:
Configured MCPContainer ready for use
"""
# Initialize logging first (MCP: file only, never stdout)
init_mcp_logging()
# Load configuration (merges file + environment variables)
config = ConfigManager().config
return cls(config=config)
@property
def is_cloud_mode(self) -> bool:
"""Check if running in cloud mode.
Returns:
True if cloud mode enabled via env var or config file
"""
return self.config.cloud_mode_enabled
@property
def is_test_env(self) -> bool:
"""Check if running in test environment.
Returns:
True if test environment detected
"""
return self.config.is_test_env
def get_client_factory(
self, override_factory: Optional[Callable[[], AbstractAsyncContextManager[AsyncClient]]] = None
) -> Callable[[], AbstractAsyncContextManager[AsyncClient]]:
"""Get HTTP client factory based on runtime mode.
Priority order:
1. Override factory (for dependency injection in tests/cloud)
2. Cloud mode: HTTP client with auth
3. Local mode: ASGI transport for in-process calls
Args:
override_factory: Optional factory override for testing/cloud
Returns:
Async context manager factory that yields configured AsyncClient
"""
if override_factory:
return override_factory
# Return a factory that creates clients based on current mode
@asynccontextmanager
async def _client_factory() -> AsyncIterator[AsyncClient]:
"""Factory that creates client based on runtime mode."""
timeout = Timeout(
connect=10.0, # 10 seconds for connection
read=30.0, # 30 seconds for reading response
write=30.0, # 30 seconds for writing request
pool=30.0, # 30 seconds for connection pool
)
if self.is_cloud_mode:
# Cloud mode: inject auth when creating client
from basic_memory.cli.auth import CLIAuth
auth = CLIAuth(
client_id=self.config.cloud_client_id, authkit_domain=self.config.cloud_domain
)
token = await auth.get_valid_token()
if not token:
raise RuntimeError(
"Cloud mode enabled but not authenticated. "
"Run 'basic-memory cloud login' first."
)
# Auth header set ONCE at client creation
proxy_base_url = f"{self.config.cloud_host}/proxy"
logger.info(f"Creating HTTP client for cloud proxy at: {proxy_base_url}")
async with AsyncClient(
base_url=proxy_base_url,
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
) as client:
yield client
else:
# Local mode: ASGI transport for in-process calls
from basic_memory.api.app import app as fastapi_app
logger.info("Creating 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
return _client_factory
src/basic_memory/mcp/server.py
+5 -2
View File
@@ -9,7 +9,7 @@ from fastmcp import FastMCP
from loguru import logger
from basic_memory import db
from basic_memory.config import ConfigManager
from basic_memory.mcp.container import MCPContainer
from basic_memory.services.initialization import initialize_app, initialize_file_sync
from basic_memory.telemetry import show_notice_if_needed, track_app_started
@@ -24,7 +24,10 @@ async def lifespan(app: FastMCP):
- File sync in background (if enabled and not in cloud mode)
- Proper cleanup on shutdown
"""
app_config = ConfigManager().config
# --- Composition Root ---
# Container handles: config loading, logging init, runtime mode selection
container = MCPContainer.create()
app_config = container.config
logger.info("Starting Basic Memory MCP server")
# Show telemetry notice (first run only) and track startup