Hitonabi
3facb6c3ad
- Phase 7: model.default = Hermes-4-14B (Q6_K, llama-swap-Alias hermes, --jinja), verifiziert (sauberer Tool-Call). A/B vs Qwen3-Coder umschaltbar. - Phase 9: mcp_memory.py Tool-Beschreibungen auf proaktives Protokoll (get→Session-Start, add→proaktiv+dedup, search→vor add) → geteiltes MC-SQLite waechst autonom aus allen Tool-Sessions mit. - Docs: Memory-Architektur (2 komplementaere Schichten) festgeschrieben. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mission Control — Memory MCP Server
|
|
|
|
Stdio-MCP-Server fuer persistentes Gedaechtnis.
|
|
Laeuft als Subprocess von Cline / Claude Code / OpenCode auf dem Client-Rechner.
|
|
Ruft die /api/memory-Endpunkte von Mission Control via HTTP auf.
|
|
|
|
Konfiguration via Env-Vars:
|
|
MC_URL Mission Control URL (default: http://192.168.178.151:9000)
|
|
MC_TOKEN Auth-Token (default: leer)
|
|
|
|
Einmalig installieren (auf dem Rechner wo das KI-Tool laeuft):
|
|
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:9000").rstrip("/")
|
|
MC_TOKEN = os.environ.get("MC_TOKEN", "")
|
|
|
|
mcp = FastMCP("mission-control-memory")
|
|
|
|
|
|
def _headers() -> dict:
|
|
return {"X-MC-Token": MC_TOKEN} if MC_TOKEN else {}
|
|
|
|
|
|
def _get(path: str, **params) -> list | dict:
|
|
r = httpx.get(f"{MC_URL}{path}", headers=_headers(), params=params, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _post(path: str, data: dict) -> dict:
|
|
r = httpx.post(f"{MC_URL}{path}", headers=_headers(), json=data, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _put(path: str, data: dict) -> dict:
|
|
r = httpx.put(f"{MC_URL}{path}", headers=_headers(), json=data, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _delete(path: str) -> dict:
|
|
r = httpx.delete(f"{MC_URL}{path}", headers=_headers(), timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
@mcp.tool()
|
|
def get_memories(category: str = "") -> str:
|
|
"""Laedt gespeicherte Fakten und Entscheidungen aus dem geteilten Gedaechtnis.
|
|
RUFE DAS ZU BEGINN JEDER SESSION AUF, um vorhandenen Kontext (Konventionen,
|
|
Entscheidungen, Nutzer-Praeferenzen) zu laden, BEVOR du handelst oder nachfragst.
|
|
category: user | instruction | stable | versioned | ephemeral | (leer = alle)"""
|
|
params = {"category": category} if category else {}
|
|
items = _get("/api/memory", **params)
|
|
if not items:
|
|
return "Keine Memories gespeichert."
|
|
cat_icon = {"stable": "🔵", "versioned": "🟡", "ephemeral": "⏱"}
|
|
return "\n".join(
|
|
f"{cat_icon.get(m['category'], '·')} [{m['category']}] {m['content']}"
|
|
f" (Quelle: {m['source']}, ID: {m['id'][:8]})"
|
|
for m in items
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def add_memory(content: str, category: str = "stable", source: str = "agent") -> str:
|
|
"""Speichert einen Fakt/eine Entscheidung dauerhaft im GETEILTEN Gedaechtnis (alle Tools sehen es).
|
|
RUFE DAS PROAKTIV AUF, sobald waehrend der Arbeit etwas Dauerhaftes entsteht — eine
|
|
Konvention, Architektur-Entscheidung, Tech-Version, Nutzer-Praeferenz oder ein Projektfakt.
|
|
Warte NICHT, bis man dich bittet.
|
|
Mache VORHER `search_memories` mit Stichworten, um Dubletten zu vermeiden; existiert ein
|
|
passender Eintrag, nutze `update_memory` statt neu anzulegen. Halte Eintraege knapp & atomar.
|
|
category: user=Fakten ueber den User (Praeferenzen/Arbeitsweise), instruction=Verhaltensregeln fuer alle Tools, stable=Projektfakten, versioned=Tech-Versionen, ephemeral=temporaer/Session-Notiz (7 Tage, dann weg).
|
|
"""
|
|
m = _post("/api/memory", {"content": content, "category": category, "source": source})
|
|
return f"Gespeichert (ID: {m['id'][:8]}): {content}"
|
|
|
|
|
|
@mcp.tool()
|
|
def search_memories(q: str) -> str:
|
|
"""Sucht per Stichwort in allen gespeicherten Fakten. Nutze das VOR `add_memory`
|
|
(Dubletten-Check) und immer dann, wenn du fruehere Entscheidungen/Kontext brauchst."""
|
|
items = _get("/api/memory", q=q)
|
|
if not items:
|
|
return f"Keine Treffer fuer '{q}'."
|
|
return "\n".join(
|
|
f"[{m['category']}] {m['content']} (ID: {m['id'][:8]})"
|
|
for m in items
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def update_memory(memory_id: str, content: str = "", category: str = "") -> str:
|
|
"""Aktualisiert einen bestehenden Eintrag (z.B. nach Tech-Update). Nur gesetzte Felder werden geaendert."""
|
|
data: dict = {}
|
|
if content:
|
|
data["content"] = content
|
|
if category:
|
|
data["category"] = category
|
|
if not data:
|
|
return "Nichts zu aktualisieren (content und category sind leer)."
|
|
m = _put(f"/api/memory/{memory_id}", data)
|
|
return f"Aktualisiert: {m['content']}"
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_memory(memory_id: str) -> str:
|
|
"""Loescht einen veralteten oder falschen Eintrag anhand seiner ID."""
|
|
_delete(f"/api/memory/{memory_id}")
|
|
return f"Eintrag {memory_id[:8]} geloescht."
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|