mission-control-v2/mcp/mcp_memory.py

#!/usr/bin/env python3
"""
Mission Control 2.0 — Memory MCP Server (geteiltes Gedächtnis für ALLE Tools).

Stdio-MCP-Server. Läuft als Subprocess von Cline/Claude Code/OpenCode/Hermes.
Ruft die /api/memory-Endpunkte von Mission Control via HTTP auf.

Tool-Beschreibungen bewusst als GUARDS formuliert (konditional, nicht imperativ),
damit kleine Modelle (z.B. Hermes 14B) nicht in Aufruf-Schleifen laufen.

Env: MC_URL (default http://192.168.178.151:9001), MC_TOKEN (optional).
Install (auf dem Rechner des Tools): pip install mcp httpx
"""

import os

import httpx
from mcp.server.fastmcp import FastMCP

MC_URL = os.environ.get("MC_URL", "http://192.168.178.151:9001").rstrip("/")
MC_TOKEN = os.environ.get("MC_TOKEN", "")

mcp = FastMCP("mission-control-memory")


def _h() -> dict:
    return {"X-MC-Token": MC_TOKEN} if MC_TOKEN else {}


def _get(path: str, **params):
    r = httpx.get(f"{MC_URL}{path}", headers=_h(), params=params, timeout=10)
    r.raise_for_status()
    return r.json()


def _post(path: str, data: dict):
    r = httpx.post(f"{MC_URL}{path}", headers=_h(), json=data, timeout=10)
    r.raise_for_status()
    return r.json()


def _put(path: str, data: dict):
    r = httpx.put(f"{MC_URL}{path}", headers=_h(), json=data, timeout=10)
    r.raise_for_status()
    return r.json()


def _delete(path: str):
    r = httpx.delete(f"{MC_URL}{path}", headers=_h(), timeout=10)
    r.raise_for_status()
    return r.json()


@mcp.tool()
def get_memories(category: str = "") -> str:
    """Lädt gespeicherte Fakten/Entscheidungen aus dem geteilten Gedächtnis.
    Nutze dies EINMAL am Session-Beginn, wenn du Projekt-Kontext brauchst — nicht wiederholt.
    category: user | instruction | stable | versioned | ephemeral | (leer = alle)"""
    items = _get("/api/memory", **({"category": category} if category else {}))
    if not items:
        return "Keine Memories gespeichert."
    icon = {"stable": "🔵", "versioned": "🟡", "ephemeral": "⏱", "user": "👤", "instruction": "📋"}
    return "\n".join(
        f"{icon.get(m['category'], '·')} [{m['category']}] {m['content']}  (ID: {m['id'][:8]})"
        for m in items
    )


@mcp.tool()
def search_memories(q: str) -> str:
    """Sucht SEMANTISCH (nach Bedeutung, nicht nur Stichwort) in den Fakten. Nutze dies
    VOR add_memory (Dubletten-Check) oder wenn du eine konkrete frühere Entscheidung suchst.
    Treffer sind nach Relevanz sortiert."""
    items = _get("/api/memory", q=q)
    if not items:
        return f"Keine Treffer für '{q}'."
    def _line(m):
        sc = f" {m['score']:.2f}" if isinstance(m.get("score"), (int, float)) else ""
        return f"[{m['category']}{sc}] {m['content']}  (ID: {m['id'][:8]})"
    return "\n".join(_line(m) for m in items)


@mcp.tool()
def add_memory(content: str, category: str = "stable", source: str = "agent") -> str:
    """Speichert EINEN dauerhaften Fakt im geteilten Gedächtnis (alle Tools sehen ihn).
    Nur aufrufen, WENN gerade etwas Dauerhaftes entstanden ist (Konvention, Architektur-
    Entscheidung, Tech-Version, Nutzer-Präferenz) UND es noch nicht existiert (vorher
    search_memories!). Knapp & atomar. Existiert ein passender Eintrag → update_memory.
    category: user | instruction | stable | versioned | ephemeral"""
    m = _post("/api/memory", {"content": content, "category": category, "source": source})
    return f"Gespeichert (ID: {m['id'][:8]}): {content}"


@mcp.tool()
def update_memory(memory_id: str, content: str = "", category: str = "") -> str:
    """Aktualisiert einen bestehenden Eintrag (nur gesetzte Felder)."""
    data = {k: v for k, v in (("content", content), ("category", category)) if v}
    if not data:
        return "Nichts zu aktualisieren."
    m = _put(f"/api/memory/{memory_id}", data)
    return f"Aktualisiert: {m['content']}"


@mcp.tool()
def delete_memory(memory_id: str) -> str:
    """Löscht einen veralteten/falschen Eintrag anhand seiner ID."""
    _delete(f"/api/memory/{memory_id}")
    return f"Eintrag {memory_id[:8]} gelöscht."


@mcp.tool()
def learn(text: str) -> str:
    """Lässt das Gedächtnis AUTOMATISCH aus einem Gesprächsausschnitt lernen: reiche
    rohe Nutzer-Aussagen/Turns durch — das Gedächtnis EXTRAHIERT die dauerhaften Fakten
    selbst und mischt sie ein (ohne Dubletten). Nutze dies, wenn der Nutzer beiläufig etwas
    Dauerhaftes über sich/seine Vorlieben/das Projekt erwähnt hat — du musst nicht selbst
    den Fakt formulieren. Für einen bereits fertig formulierten Einzel-Fakt → add_memory."""
    res = _post("/api/memory/learn", {"text": text})
    items = res.get("results", []) if isinstance(res, dict) else []
    learned = [r for r in items if r.get("event") in (None, "ADD", "UPDATE")]
    if not learned:
        return "Nichts Neues gelernt (keine dauerhaften Fakten erkannt)."
    return "Gelernt:\n" + "\n".join(f"· {r.get('content')}" for r in learned)


if __name__ == "__main__":
    mcp.run()