diff --git a/README.md b/README.md index 0ff0390..23ef0f9 100644 --- a/README.md +++ b/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 diff --git a/src/digital_persona/limitless_mcp_server.py b/src/digital_persona/limitless_mcp_server.py new file mode 100644 index 0000000..cb6bea1 --- /dev/null +++ b/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() + diff --git a/src/digital_persona/mcp_plugins/limitless.py b/src/digital_persona/mcp_plugins/limitless.py index be72557..1afb5ba 100644 --- a/src/digital_persona/mcp_plugins/limitless.py +++ b/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: diff --git a/src/digital_persona/mcp_server.py b/src/digital_persona/mcp_server.py index 142a08a..57c769c 100644 --- a/src/digital_persona/mcp_server.py +++ b/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) diff --git a/src/digital_persona/mcp_service.py b/src/digital_persona/mcp_service.py new file mode 100644 index 0000000..ee94292 --- /dev/null +++ b/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() + diff --git a/tests/test_limitless.py b/tests/test_limitless.py index 481ad78..90746c8 100644 --- a/tests/test_limitless.py +++ b/tests/test_limitless.py @@ -11,6 +11,7 @@ def setup_limitless(monkeypatch, tmp_path: Path): monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) monkeypatch.setenv("LIMITLESS_API_KEY", "test-key") import digital_persona.mcp_plugins.limitless as limitless + limitless = importlib.reload(limitless) return limitless @@ -41,6 +42,7 @@ def test_requires_api_key(monkeypatch, tmp_path): monkeypatch.delenv("LIMITLESS_API_KEY", raising=False) with pytest.raises(RuntimeError): import importlib as _imp + _imp.reload(_imp.import_module("digital_persona.mcp_plugins.limitless")) @@ -58,7 +60,9 @@ def fake_fetch(*, start=None, cursor=None): limitless.run_once() assert captured["start"] is not None - delta = datetime.now(UTC) - datetime.fromisoformat(captured["start"].replace("Z", "+00:00")) + delta = datetime.now(UTC) - datetime.fromisoformat( + captured["start"].replace("Z", "+00:00") + ) assert 0 <= delta.days <= 2 @@ -66,7 +70,9 @@ def test_missing_file_triggers_redownload(monkeypatch, tmp_path): limitless = setup_limitless(monkeypatch, tmp_path) state_file = tmp_path / "limitless_state.json" - state_file.write_text(json.dumps({"last_id": "99", "start": "2025-07-01T00:00:00Z"})) + state_file.write_text( + json.dumps({"last_id": "99", "start": "2025-07-01T00:00:00Z"}) + ) captured = {} def fake_fetch(*, start=None, cursor=None): @@ -80,3 +86,27 @@ def fake_fetch(*, start=None, cursor=None): assert captured["start"] is not None assert captured["cursor"] is None + +def test_search_local_entries_by_speaker(monkeypatch, tmp_path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("LIMITLESS_API_KEY", "x") + monkeypatch.setenv("PLAINTEXT_MEMORIES", "1") + import importlib + import digital_persona.mcp_plugins.limitless as limitless + + limitless = importlib.reload(limitless) + + entry1 = {"id": "1", "contents": [{"speakerName": "You", "content": "hi"}]} + entry2 = {"id": "2", "contents": [{"speakerName": "Alice", "content": "yo"}]} + + limitless.INPUT_DIR.mkdir(parents=True, exist_ok=True) + save_json_encrypted = limitless.save_json_encrypted + fernet = limitless.FERNET + save_json_encrypted(entry1, limitless.INPUT_DIR / "limitless-1.json", fernet) + save_json_encrypted(entry2, limitless.INPUT_DIR / "limitless-2.json", fernet) + + items = limitless._search_local_entries(speaker_name="You") + assert len(items) == 1 and items[0]["id"] == "1" + + items = limitless._search_local_entries(speaker_name="Alice") + assert len(items) == 1 and items[0]["id"] == "2" diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index ac26ad3..d091032 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -10,12 +10,133 @@ def test_limitless_route(monkeypatch, tmp_path: Path): monkeypatch.setenv("MCP_PLUGINS", "digital_persona.mcp_plugins.limitless") import digital_persona.mcp_plugins.limitless as limitless - monkeypatch.setattr(limitless, "_fetch_entries", lambda start=None, cursor=None: ([{"id": "1", "content": "hi"}], None)) + monkeypatch.setattr( + limitless, + "_search_local_entries", + lambda start=None, end=None, keyword=None, speaker_name=None: [ + {"id": "1", "content": "hi"} + ], + ) from digital_persona import mcp_server importlib.reload(mcp_server) app = mcp_server.create_app() client = TestClient(app) - resp = client.get("/limitless/lifelogs") + resp = client.post("/limitless/lifelogs", json={}) assert resp.status_code == 200 assert resp.json()["items"][0]["content"] == "hi" + + +def test_limitless_ignores_string_params(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("LIMITLESS_API_KEY", "x") + monkeypatch.setenv("MCP_PLUGINS", "digital_persona.mcp_plugins.limitless") + + import digital_persona.mcp_plugins.limitless as limitless + + def fake_load(start=None, end=None, keyword=None, speaker_name=None): + assert start is None + assert end is None + assert keyword is None + assert speaker_name is None + return [{"id": "1"}] + + monkeypatch.setattr(limitless, "_search_local_entries", fake_load) + + from digital_persona import mcp_server + importlib.reload(mcp_server) + app = mcp_server.create_app() + client = TestClient(app) + + resp = client.post( + "/limitless/lifelogs", + json={"start": "string", "end": "string", "keyword": "string", "speakerName": "string"}, + ) + assert resp.status_code == 200 + assert resp.json()["items"][0]["id"] == "1" + + +def test_limitless_http_error(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("LIMITLESS_API_KEY", "x") + monkeypatch.setenv("MCP_PLUGINS", "digital_persona.mcp_plugins.limitless") + + import digital_persona.mcp_plugins.limitless as limitless + from fastapi import HTTPException + + def fake_load(**_): + raise HTTPException(status_code=400, detail="bad") + + monkeypatch.setattr(limitless, "_search_local_entries", fake_load) + + from digital_persona import mcp_server + importlib.reload(mcp_server) + app = mcp_server.create_app() + client = TestClient(app) + + resp = client.post("/limitless/lifelogs", json={}) + assert resp.status_code == 400 + + +def test_root_route(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("MCP_PLUGINS", "") + + from digital_persona import mcp_server + importlib.reload(mcp_server) + app = mcp_server.create_app([]) + client = TestClient(app) + + resp = client.get("/") + assert resp.status_code == 200 + assert resp.json()["message"].startswith("MCP server") + + +def test_ai_plugin(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("MCP_PLUGINS", "") + + from digital_persona import limitless_mcp_server + importlib.reload(limitless_mcp_server) + app = limitless_mcp_server.create_limitless_app() + client = TestClient(app) + + resp = client.get("/.well-known/ai-plugin.json") + assert resp.status_code == 200 + data = resp.json() + assert data["name_for_model"] == "limitless_mcp" + + +def test_ai_plugin_via_create_app(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("LIMITLESS_API_KEY", "x") + monkeypatch.setenv("MCP_PLUGINS", "digital_persona.mcp_plugins.limitless") + + from digital_persona import mcp_server + importlib.reload(mcp_server) + app = mcp_server.create_app() + client = TestClient(app) + + resp = client.get("/.well-known/ai-plugin.json") + assert resp.status_code == 200 + assert resp.json()["name_for_model"] == "limitless_mcp" + + +def test_cors_options(monkeypatch, tmp_path: Path): + monkeypatch.setenv("PERSONA_DIR", str(tmp_path)) + monkeypatch.setenv("MCP_PLUGINS", "") + + from digital_persona import mcp_server + importlib.reload(mcp_server) + app = mcp_server.create_app([]) + client = TestClient(app) + + resp = client.options( + "/openapi.json", + headers={ + "Origin": "http://example.com", + "Access-Control-Request-Method": "GET", + }, + ) + assert resp.status_code in {200, 204} + assert resp.headers.get("access-control-allow-origin") == "*"