Local lifelog search via MCP

README.md

@@ -121,6 +121,7 @@ Environment variables:
    - The object also stores a relative `source` path to the processed original file so you can reference images or audio later.
    - Non-text media should be ingested first so a text summary is available.
    - Completed memories are moved to `PERSONA_DIR/archive` after `/complete_interview` so they won't be processed twice.
+  - The MCP server runs on `http://localhost:8900`. Visit `/` for a quick status check, `/docs` for the UI, or `/openapi.json` for the spec. CORS is enabled so other tools can fetch `/openapi.json`. When the Limitless plugin is enabled the server also serves its plugin manifest from `/.well-known/ai-plugin.json`. Call `POST /limitless/lifelogs` with a JSON body containing optional `start`, `end`, `keyword`, and `speakerName` fields to search previously fetched Limitless lifelogs stored in `PERSONA_DIR/processed`. You may append `?api_key=YOUR_KEY` but it is only required when the server fetches data from Limitless directly.
 
 ### Sample Data
 

src/digital_persona/limitless_mcp_server.py

@@ -0,0 +1,25 @@
+import os
+from fastapi import FastAPI
+
+from .mcp_server import create_app
+from .mcp_service import DEFAULT_PLUGINS
+
+
+def create_limitless_app() -> FastAPI:
+    app = create_app(DEFAULT_PLUGINS)
+    app.title = "limitless_mcp"
+
+    return app
+
+
+def _cli() -> None:
+    import uvicorn
+
+    app = create_limitless_app()
+    port = int(os.getenv("MCP_PORT", "8900"))
+    uvicorn.run(app, host="0.0.0.0", port=port)
+
+
+if __name__ == "__main__":
+    _cli()
+

src/digital_persona/mcp_plugins/limitless.py

@@ -4,13 +4,23 @@
 from datetime import datetime, timedelta, UTC
 import asyncio
 import httpx
-from fastapi import APIRouter, FastAPI
+from fastapi import APIRouter, FastAPI, Security, Request
+from fastapi.security import APIKeyQuery
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
 from digital_persona.utils.filename import sanitize_filename
+from digital_persona.secure_storage import (
+    get_fernet,
+    save_json_encrypted,
+    load_json_encrypted,
+)
 
 from digital_persona import config as dp_config
 
 dp_config.load_env()
-from digital_persona.ingest import INPUT_DIR, _persona_dir
+from digital_persona.ingest import INPUT_DIR, PROCESSED_DIR, _persona_dir
+
+FERNET = get_fernet(_persona_dir())
 
 STATE_FILE = _persona_dir() / "limitless_state.json"
 API_URL = os.getenv("LIMITLESS_API_URL", "https://api.limitless.ai/v1")
@@ -28,6 +38,39 @@
 
 router = APIRouter()
 
+api_key_query = APIKeyQuery(name="api_key", auto_error=False)
+
+
+class LifelogParams(BaseModel):
+    """Parameters accepted by the lifelogs endpoint."""
+
+    start: str | None = Field(
+        default=None,
+        description="ISO 8601 start timestamp",
+        examples=["2025-07-19T00:00:00Z"],
+    )
+    end: str | None = Field(
+        default=None,
+        description="ISO 8601 end timestamp",
+        examples=["2025-07-20T00:00:00Z"],
+    )
+    keyword: str | None = Field(
+        default=None,
+        description="Filter entries containing this text",
+        examples=["meeting"],
+    )
+    speaker_name: str | None = Field(
+        default=None,
+        alias="speakerName",
+        description="Filter entries attributed to this speaker",
+        examples=["Alice"],
+    )
+
+    model_config = {
+        "populate_by_name": True,
+        "extra": "ignore",
+    }
+
 
 def setup(app: FastAPI) -> None:
     """Attach background ingest task to *app* startup."""
@@ -40,6 +83,28 @@ async def _loop() -> None:
 
     app.add_event_handler("startup", lambda: asyncio.create_task(_loop()))
 
+    @app.get("/.well-known/ai-plugin.json", include_in_schema=False)
+    def ai_plugin(request: Request) -> JSONResponse:
+        """Return Open WebUI plugin manifest."""
+        base = str(request.base_url).rstrip("/")
+        manifest = {
+            "schema_version": "v1",
+            "name_for_human": "Limitless MCP",
+            "name_for_model": "limitless_mcp",
+            "description_for_human": "Search your stored Limitless lifelogs",
+            "description_for_model": "Search previously ingested lifelogs via the MCP server",
+            "auth": {"type": "none"},
+            "api": {
+                "type": "openapi",
+                "url": f"{base}{app.openapi_url}",
+                "is_user_authenticated": False,
+            },
+            "logo_url": f"{base}/logo.png",
+            "contact_email": "support@example.com",
+            "legal_info_url": "https://example.com/legal",
+        }
+        return JSONResponse(content=manifest)
+
 
 def _load_state() -> dict:
     if STATE_FILE.exists():
@@ -54,10 +119,12 @@ def _save_state(state: dict) -> None:
     STATE_FILE.write_text(json.dumps(state))
 
 
-def _fetch_entries(*, start: str | None = None, cursor: str | None = None) -> tuple[list[dict], str | None]:
+def _fetch_entries(
+    *, start: str | None = None, cursor: str | None = None, api_key: str = API_KEY
+) -> tuple[list[dict], str | None]:
     """Return lifelog entries and the next cursor."""
 
-    headers = {"X-API-Key": API_KEY}
+    headers = {"X-API-Key": api_key}
     params = {}
     if start:
         params["start"] = start
@@ -84,14 +151,92 @@ def _save_entry(entry: dict) -> None:
     entry_id = sanitize_filename(str(entry_id))
     out = _get_entry_filename(entry_id)
     obj = {k: v for k, v in entry.items()}
-    out.write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8")
+    save_json_encrypted(obj, out, FERNET)
     logger.info("Saved %s", out.name)
 
+
 def _get_entry_filename(entry_id: str) -> os.PathLike:
     """Construct the file path for a given entry ID."""
     return INPUT_DIR / f"limitless-{entry_id}.json"
 
 
+def _contains_speaker(obj: object, speaker_name: str) -> bool:
+    """Return True if *obj* or nested values contain the speaker name."""
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if key in {"speakerName", "speaker"}:
+                if value and speaker_name.lower() in str(value).lower():
+                    return True
+            if key == "metadata" and isinstance(value, dict):
+                val = value.get("speakerName") or value.get("speaker")
+                if val and speaker_name.lower() in str(val).lower():
+                    return True
+            if _contains_speaker(value, speaker_name):
+                return True
+    elif isinstance(obj, list):
+        return any(_contains_speaker(v, speaker_name) for v in obj)
+    return False
+
+
+def _search_local_entries(
+    *,
+    start: str | None = None,
+    end: str | None = None,
+    keyword: str | None = None,
+    speaker_name: str | None = None,
+    limit: int = 100,
+) -> list[dict]:
+    """Return stored Limitless entries from the persona directory."""
+
+    files = sorted(
+        list(PROCESSED_DIR.glob("limitless-*.json"))
+        + list(INPUT_DIR.glob("limitless-*.json"))
+    )
+    start_dt = None
+    end_dt = None
+    if start:
+        try:
+            start_dt = datetime.fromisoformat(start.replace("Z", "+00:00"))
+        except Exception:
+            start_dt = None
+    if end:
+        try:
+            end_dt = datetime.fromisoformat(end.replace("Z", "+00:00"))
+        except Exception:
+            end_dt = None
+
+    items: list[dict] = []
+    for path in files:
+        try:
+            obj = load_json_encrypted(path, FERNET)
+        except Exception:
+            continue
+        ts = obj.get("updatedAt") or obj.get("timestamp") or obj.get("endTime")
+        if ts:
+            try:
+                ts_dt = datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
+                if start_dt and ts_dt < start_dt:
+                    continue
+                if end_dt and ts_dt > end_dt:
+                    continue
+            except Exception:
+                pass
+        if keyword:
+            text = json.dumps(obj, ensure_ascii=False).lower()
+            if keyword.lower() not in text:
+                continue
+        if speaker_name:
+            try:
+                if not _contains_speaker(obj, speaker_name):
+                    continue
+            except Exception:
+                continue
+        items.append(obj)
+        if len(items) >= limit:
+            break
+    return items
+
+
 def run_once() -> None:
     state = _load_state()
     last_id: str | None = state.get("last_id")
@@ -102,7 +247,11 @@ def run_once() -> None:
         last_id = None
 
     if not cursor and not start:
-        start = (datetime.now(UTC) - timedelta(days=LOOKBACK_DAYS)).isoformat().replace("+00:00", "Z")
+        start = (
+            (datetime.now(UTC) - timedelta(days=LOOKBACK_DAYS))
+            .isoformat()
+            .replace("+00:00", "Z")
+        )
 
     try:
         entries, next_cursor = _fetch_entries(start=start, cursor=cursor)
@@ -132,11 +281,34 @@ def run_once() -> None:
     _save_state(state)
 
 
-@router.get("/lifelogs")
-async def api_lifelogs(start: str | None = None, cursor: str | None = None) -> dict:
+@router.post(
+    "/lifelogs",
+    name="limitless_lifelogs",
+    description="Fetch Limitless lifelog entries",
+    operation_id="limitless_lifelogs",
+)
+async def api_lifelogs(
+    params: LifelogParams | None = None,
+    api_key: str | None = Security(api_key_query),
+) -> dict:
     """Return Limitless entries via the MCP server."""
-    items, next_cursor = _fetch_entries(start=start, cursor=cursor)
-    return {"items": items, "next_cursor": next_cursor}
+    start = params.start if params else None
+    end = params.end if params else None
+    keyword = params.keyword if params else None
+    speaker_name = params.speaker_name if params else None
+    # OpenAPI tooling may send literal "string" when no value is provided
+    if start == "string":
+        start = None
+    if end == "string":
+        end = None
+    if keyword == "string":
+        keyword = None
+    if speaker_name == "string":
+        speaker_name = None
+    items = _search_local_entries(
+        start=start, end=end, keyword=keyword, speaker_name=speaker_name
+    )
+    return {"items": items}
 
 
 def _cli() -> None:

src/digital_persona/mcp_server.py

@@ -2,23 +2,32 @@
 import importlib
 import logging
 from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 
-DEFAULT_PLUGINS = ["digital_persona.mcp_plugins.limitless"]
+from .mcp_service import get_plugin_names
 
 
 def create_app(plugin_names: list[str] | None = None) -> FastAPI:
     if plugin_names is None:
-        env = os.getenv("MCP_PLUGINS")
-        if env:
-            plugin_names = [p.strip() for p in env.split(",") if p.strip()]
-        else:
-            plugin_names = DEFAULT_PLUGINS
+        plugin_names = get_plugin_names()
 
     logger = logging.getLogger(__name__)
     if not logger.handlers:
         logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
 
     app = FastAPI(title="MCP Server")
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.get("/", include_in_schema=False)
+    def root() -> dict:
+        """Basic health check message."""
+        return {"message": "MCP server running", "docs": "/docs"}
+
     for name in plugin_names:
         try:
             mod = importlib.import_module(name)

src/digital_persona/mcp_service.py

@@ -0,0 +1,10 @@
+import os
+
+DEFAULT_PLUGINS = ["digital_persona.mcp_plugins.limitless"]
+
+def get_plugin_names() -> list[str]:
+    env = os.getenv("MCP_PLUGINS")
+    if env:
+        return [p.strip() for p in env.split(",") if p.strip()]
+    return DEFAULT_PLUGINS.copy()
+