FastAPI Application

Overview

The Mem0 FastAPI application serves as the core backend service for managing, searching, and organizing AI memory artifacts through a robust REST API. Built on FastAPI and SQLAlchemy, the server orchestrates application lifecycle initialization, dynamic runtime configuration changes across vector and LLM backends, and multi-tenant authentication supporting JWT bearer tokens, user API keys, and legacy admin credentials. It integrates custom request logging, rate limiting via Slowapi, and upstream error translation to expose clean endpoints for user management, API key rotation, request telemetry browsing, and entity manipulation.

Sources: server/main.py:146-173, server/errors.py:71-84, server/auth.py:144-176

Application Bootstrap and State Initialization

Server State and Session Wiring

The server bootstrap sequence initializes environment configuration, default connection parameters, and global thread-safe state before instantiating the FastAPI application. Environment variables govern Postgres vector storage connection targets, OpenAI API keys, history paths, and model selections.

Sources: server/main.py:108-139

Server state management relies on thread-locked configuration containers and factory registration in server_state.py. Before the FastAPI application object is created, set_session_factory binds the SQLAlchemy SessionLocal maker, and initialize_state loads any persisted configuration overrides from the database before building the initial Memory instance via Memory.from_config.

Copydef initialize_state(default_config: Dict[str, Any]) -> None:
    global _current_config, _memory_instance
    with _state_lock:
        _current_config = deepcopy(default_config)
        overrides = _load_overrides()
        if overrides:
            _current_config = _merge_config(_current_config, overrides)
        _memory_instance = Memory.from_config(_current_config)
 

Sources: server/server_state.py:76-84

Sources: server/server_state.py:103-107

Configuration Validation and Unconfigured Warnings

Startup checks enforce authentication constraints based on AUTH_DISABLED, JWT_SECRET, and ADMIN_API_KEY. If AUTH_DISABLED is false and JWT_SECRET is missing, initialization aborts immediately with a RuntimeError. If ADMIN_API_KEY is not provided, _warn_if_unconfigured() queries the user table via the session factory to verify whether any administrator accounts exist.

Copydef _warn_if_unconfigured() -> None:
    """Pre-auth deployments upgrading into this build will 401 everywhere until
    an admin key or admin user exists. Surface the fix before the support tickets."""
    try:
        with SessionLocal() as session:
            if session.scalar(select(func.count(User.id))) > 0:
                return
    except Exception:
        return
 
    logging.warning(
        "\n%s\n"
        "  Auth is enabled by default and this server has no admin configured.\n"
        "  Protected endpoints will return 401 until you either:\n"
        "    1. Set ADMIN_API_KEY=<long-random-value>  (fastest, no client changes)\n"
        "    2. Register an admin at http://<host>:3000/setup\n"
        "    3. Set AUTH_DISABLED=true                 (local development only)\n"
        "  Docs: https://docs.mem0.ai/open-source/features/rest-api#authentication\n"
        "%s",
        "=" * 72,
        "=" * 72,
    )
 

Sources: server/main.py:66-88

CORS and Request Pipeline Middleware

CORS Policy and Telemetry Nudging

The FastAPI application integrates CORS policies through CORSMiddleware, restricting allowed origins to the configured dashboard URL (DASHBOARD_URL), which defaults to http://localhost:3000. Credentials and all HTTP methods and headers are permitted for cross-origin interactions.

CopyDASHBOARD_URL = os.environ.get("DASHBOARD_URL", "http://localhost:3000")
app.add_middleware(
    CORSMiddleware,
    allow_origins=[DASHBOARD_URL],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Sources: server/main.py:160-167

Before the HTTP pipeline processes incoming connections, telemetry logging status is initialized via telemetry.log_status().

Copytelemetry.log_status()

Sources: server/main.py:106-106

Request ID Tagging and Logging Filters

Request traceability is enforced across execution contexts using a ContextVar named request_id_var defaulting to "-". Unique 8-character hexadecimal identifiers are generated via new_request_id() using UUID4.

Copyrequest_id_var: ContextVar[str] = ContextVar("request_id", default="-")
…
def new_request_id() -> str:
    return uuid.uuid4().hex[:8]

Sources: server/errors.py:9-9, server/errors.py:67-68

The logging record factory is monkey-patched via install_request_id_logging() to automatically inject the current context's request_id into every log record. Standard logging is configured with a formatting string that includes the request identifier tag.

Copydef install_request_id_logging() -> None:
    old_factory = logging.getLogRecordFactory()
 
    def factory(*args, **kwargs):
        record = old_factory(*args, **kwargs)
        record.request_id = request_id_var.get()
        return record
 
    logging.setLogRecordFactory(factory)
…
install_request_id_logging()
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - [%(request_id)s] %(message)s")

Sources: server/errors.py:87-95, server/main.py:45-46

Certain paths and prefixes are explicitly excluded from request logging telemetry to prevent noise from health inspections and documentation endpoints.

Sources: server/main.py:59-60

Sources: server/main.py:59-60

Modular Subsystem Router Composition

Subsystem Router Integration

The FastAPI application instance consolidates its API surface by including modular subsystem routers defined across the project architecture. These include authentication management, API key provisioning, entity browsing, and request telemetry inspection routes. Each imported router is mounted directly onto the central app instance.

Copyfrom routers import api_keys as api_keys_router
from routers import auth as auth_router
from routers import entities as entities_router
from routers import requests as requests_router
…
app.include_router(auth_router.router)
app.include_router(api_keys_router.router)
app.include_router(entities_router.router)
app.include_router(requests_router.router)

Sources: server/main.py:25-28, server/main.py:169-172

Sources: server/main.py:25-28, server/main.py:169-172

Sources: server/main.py:146-156

Dynamic Memory Configuration Management

Runtime State and Configuration Merging

The application manages runtime configuration through thread-safe state wrappers using threading.RLock() to coordinate access across concurrent requests. Base configurations defined during server bootstrap are merged recursively with persistent overrides stored in the database. The _merge_config() function executes a deep copy of the base dictionary and iterates through update keys, recursively updating nested dictionaries such as vector store and LLM provider parameters while overwriting scalar properties.

Copy_state_lock = threading.RLock()
_current_config: Dict[str, Any] = {}
…
def _merge_config(base: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]:
    merged = deepcopy(base)
 
    for key, value in updates.items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            merged[key] = _merge_config(merged[key], value)
        else:
            merged[key] = value
 
    return merged

Sources: server/server_state.py:9-10, server/server_state.py:64-73

Sources: server/server_state.py:86-96

Database Persistence and Redaction

Persistent configuration overrides are serialized to JSON format and stored in the database using the Settings model under the primary key "config_overrides". The _save_overrides() function utilizes SQLAlchemy's PostgreSQL upsert construct (on_conflict_do_update) to insert or update override records atomically. If the session factory is unconfigured or a database exception occurs, the failure is caught and logged without crashing the request flow.

Copydef _save_overrides(overrides: Dict[str, Any]) -> None:
    try:
        if _session_factory is None:
            return
        from models import Settings
        from sqlalchemy.dialects.postgresql import insert
 
        session = _session_factory()
        try:
            serialized = json.dumps(overrides)
            stmt = (
                insert(Settings)
                .values(key="config_overrides", value=serialized)
                .on_conflict_do_update(
                    index_elements=[Settings.key],
                    set_={"value": serialized},
                )
            )
            session.execute(stmt)
            session.commit()
        finally:
            session.close()
    except Exception:
        logging.warning("Failed to persist config overrides to database", exc_info=True)
 

Sources: server/server_state.py:38-62

To prevent accidental exposure of sensitive keys in logs and API responses, the server defines a strict set of sensitive configuration keys that must be redacted or filtered during configuration inspection.

Sources: server/main.py:49-58

Rate Limiting Architecture

Limiter Integration and Client Identification

The application integrates rate limiting functionality through the slowapi library to control request volume across the API endpoints. A global Limiter instance is constructed by leveraging get_remote_address as the key function to identify incoming client requests based on their remote network address.

Copyfrom slowapi import Limiter
from slowapi.util import get_remote_address
 

Sources: server/rate_limit.py:1-3

This rate limiter instance is bound directly to the FastAPI application state during bootstrap and configured with an exception handler to intercept rate violation events.

Copyapp.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Sources: server/main.py:157-158

Sources: server/main.py:37-38, server/main.py:158

Exception Classification and Error Translation

Upstream Provider Fault Mapping

The server intercepts runtime exceptions raised by upstream LLM providers, embedding services, vector databases, and underlying storage backends, translating them into structured HTTP 502 Bad Gateway responses via the UpstreamError class. The classification engine inspects exception types and exception class names by walking the cause and context chains (__cause__ and __context__) to correctly identify wrapped provider faults.

Copyclass UpstreamError(HTTPException):
    def __init__(self, code: str, detail: str, request_id: str) -> None:
        super().__init__(status_code=502, detail=detail)
        self.code = code
        self.request_id = request_id
 
…
def _classify(exc: BaseException | None) -> tuple[str, str]:
    # Walk the cause/context chain so wrapped provider errors still classify correctly.
    seen: set[int] = set()
    current = exc
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        result = _classify_one(current)
        if result[0] != "unknown":
            return result
        current = current.__cause__ or current.__context__
    return ("unknown", "Upstream provider error.")
 

Sources: server/errors.py:12-17, server/errors.py:54-65

Sources: server/errors.py:19-51

Sources: server/errors.py:56-64

Error Response Handlers and Correlation

When an UpstreamError is intercepted by upstream_error_handler, the FastAPI application formats a JSON response containing the error detail, error code, and request correlation ID, while attaching the request ID as an X-Request-ID header.

Copyasync def upstream_error_handler(_: Request, exc: UpstreamError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "code": exc.code, "request_id": exc.request_id},
        headers={"X-Request-ID": exc.request_id},
    )

Sources: server/errors.py:79-84

The request identifier is managed across asynchronous boundaries using context variables (request_id_var) and injected into all log records via a custom logging record factory installed during application startup.

Copyrequest_id_var: ContextVar[str] = ContextVar("request_id", default="-")
 

Sources: server/errors.py:9-10

Copyapp.add_exception_handler(UpstreamError, upstream_error_handler)

Sources: server/main.py:159

Related

• Server Authentication
• Server API Endpoints
• Database and Migrations