Minimal FastAPI proxy in front of MemoriLabs/Memori — OpenAI-compatible /v1/chat/completions for n8n & friends

.env.example

# Token clients send as `Authorization: Bearer ...`
MEMORI_PROXY_API_KEY=replace-with-a-long-random-string

# Memori storage (BYODB) — Memori auto-creates its schema on first run
MEMORI_POSTGRES_URL=postgresql://user:pass@host:5432/memori?sslmode=require

# Upstream LLM (OpenAI direct, or any OpenAI-compatible endpoint)
OPENAI_API_KEY=sk-...
UPSTREAM_BASE_URL=https://api.openai.com/v1

docker-compose.yml

services:
  memori-api:
    build: .
    container_name: memori-api
    restart: unless-stopped
    ports:
      - "8012:8000"
    env_file: .env
    deploy:
      resources:
        limits: { memory: 1G }
        reservations: { memory: 256M }

Dockerfile

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update \
    && apt-get install -y --no-install-recommends libpq5 curl \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py /app/main.py

RUN useradd -m -u 1001 appuser && chown -R appuser:appuser /app
USER appuser

HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
    CMD curl -fsS http://localhost:8000/health || exit 1

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

main.py

"""
Minimal FastAPI proxy in front of the MemoriLabs Memori SDK.

Exposes an OpenAI-compatible /v1/chat/completions and /v1/models so any
OpenAI-shaped client (n8n, Open WebUI, LibreChat, etc.) can talk to it.

Per-request scoping comes from either:
  - HTTP headers: X-Memori-{Entity,Process,Session}-Id   (preferred)
  - JSON body:    "memori_attribution": {entity_id, process_id, session_id}

Memory is stored in your own Postgres (BYODB) — set MEMORI_POSTGRES_URL.
"""

import os
from fastapi import FastAPI, Request, HTTPException, Header
from fastapi.responses import StreamingResponse, JSONResponse
from openai import OpenAI
from memori import Memori

PROXY_KEY    = os.environ["MEMORI_PROXY_API_KEY"]
PG_URL       = os.environ["MEMORI_POSTGRES_URL"]
UPSTREAM     = os.environ.get("UPSTREAM_BASE_URL", "https://api.openai.com/v1")
UPSTREAM_KEY = os.environ["OPENAI_API_KEY"]

app = FastAPI()
client = OpenAI(base_url=UPSTREAM, api_key=UPSTREAM_KEY)


def _auth(authorization: str | None) -> None:
    if authorization != f"Bearer {PROXY_KEY}":
        raise HTTPException(401, "bad token")


def _attribution(body: dict, headers) -> dict:
    body_attr = body.pop("memori_attribution", {}) or {}
    return {
        "entity_id":  headers.get("x-memori-entity-id")  or body_attr.get("entity_id"),
        "process_id": headers.get("x-memori-process-id") or body_attr.get("process_id"),
        "session_id": headers.get("x-memori-session-id") or body_attr.get("session_id"),
    }


@app.get("/health")
def health():
    return {"status": "healthy"}


@app.get("/v1/models")
def models():
    return JSONResponse(client.models.list().model_dump())


@app.post("/v1/chat/completions")
async def chat(req: Request, authorization: str | None = Header(None)):
    _auth(authorization)
    body = await req.json()
    attr = _attribution(body, req.headers)

    mem = Memori(database_connect=PG_URL, namespace=attr["process_id"] or "default")
    mem.enable()
    mem.attribution(entity_id=attr["entity_id"], process_id=attr["process_id"])
    mem.llm.register(client)

    if body.get("stream"):
        def gen():
            for chunk in client.chat.completions.create(**body):
                yield f"data: {chunk.model_dump_json()}\n\n"
            yield "data: [DONE]\n\n"
        return StreamingResponse(gen(), media_type="text/event-stream")

    resp = client.chat.completions.create(**body)
    return JSONResponse(resp.model_dump())

requirements.txt

memori==3.3.1
fastapi==0.115.6
uvicorn[standard]==0.32.1
openai==1.58.1
psycopg[binary]==3.2.3
SQLAlchemy==2.0.36
sentence-transformers==5.4.1
python-dotenv==1.0.1