Prompt-Diät (Faden 7): schlanker Voice-MCP-Server
mcp_voice.py = Sprach-Lane-Teilmenge: reminder_create/list/delete + box_status. Die api_server-Plattform (Lucy) bekommt per platform_toolsets.api_server nur diesen statt der vollen mission-control-stack-Control-Plane (26 Tools) — CLI/Telegram/Werkstatt bleiben unangetastet. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,100 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Mission Control 2.0 — schlanker Voice-MCP-Server (für Lucys Sprach-Lane).
|
||||||
|
|
||||||
|
Prompt-Diät (Faden 7): die api_server-Plattform (Lucy Desktop) bekommt per
|
||||||
|
platform_toolsets.api_server NUR diesen dünnen Server statt der vollen
|
||||||
|
mission-control-stack-Control-Plane (26 Tools, ~große Schemas). Enthalten ist,
|
||||||
|
was man einer gesprochenen Assistentin zuruft: Erinnerungen + „geht's der Box gut?".
|
||||||
|
CLI/Telegram/Werkstatt behalten den vollen Stack (mcp_mc.py) — hier NICHTS ergänzen,
|
||||||
|
was nicht in ein Sprachgespräch gehört.
|
||||||
|
|
||||||
|
Env: MC_URL (default http://127.0.0.1:9001).
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
from mcp.server.fastmcp import FastMCP
|
||||||
|
|
||||||
|
MC_URL = os.environ.get("MC_URL", "http://127.0.0.1:9001").rstrip("/")
|
||||||
|
|
||||||
|
mcp = FastMCP("mission-control-voice")
|
||||||
|
|
||||||
|
|
||||||
|
def _get(path: str, **params):
|
||||||
|
r = httpx.get(f"{MC_URL}{path}", params=params, timeout=30)
|
||||||
|
r.raise_for_status()
|
||||||
|
return r.json()
|
||||||
|
|
||||||
|
|
||||||
|
def _post(path: str, data: dict):
|
||||||
|
r = httpx.post(f"{MC_URL}{path}", json=data, timeout=30)
|
||||||
|
r.raise_for_status()
|
||||||
|
return r.json()
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def reminder_create(text: str, when: str, repeat: str = "") -> str:
|
||||||
|
"""Legt eine Erinnerung an: zur angegebenen Zeit sagt die Box dem Commander den Text AKTIV
|
||||||
|
an (Lucy spricht + Telegram). NUR nutzen, wenn der Commander ausdrücklich erinnert werden
|
||||||
|
will ('erinner mich', 'sag mir um', 'jeden Morgen').
|
||||||
|
when = ISO-8601 mit Datum UND Uhrzeit in der Lokalzeit des Commanders, z.B. 2026-07-04T09:00.
|
||||||
|
Relative Angaben ('in 20 Minuten', 'morgen früh') selbst aus der aktuellen Zeit im Kontext
|
||||||
|
umrechnen. repeat: '' einmalig | daily (täglich) | weekdays (Mo–Fr) | weekly (wöchentlich).
|
||||||
|
text = die Ansage selbst, kurz und direkt (ohne 'Erinnerung:' davor — das ergänzt die Box)."""
|
||||||
|
try:
|
||||||
|
item = _post("/api/reminders", {"text": text, "when": when, "repeat": repeat})["item"]
|
||||||
|
except httpx.HTTPStatusError as e:
|
||||||
|
return f"Fehlgeschlagen: {e.response.json().get('detail', e.response.text[:200])}"
|
||||||
|
rep = {"daily": " (täglich)", "weekdays": " (werktags)", "weekly": " (wöchentlich)"}.get(item["repeat"], "")
|
||||||
|
return f"Erinnerung #{item['id']} angelegt: {item['when_local']}{rep} — „{item['text']}“"
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def reminder_list() -> str:
|
||||||
|
"""Zeigt alle anstehenden Erinnerungen/Routinen (Nummer, Zeit, Text). Vor dem Löschen
|
||||||
|
hiermit die Nummer ermitteln."""
|
||||||
|
items = _get("/api/reminders")["items"]
|
||||||
|
if not items:
|
||||||
|
return "Keine Erinnerungen angelegt."
|
||||||
|
rep = {"daily": " · täglich", "weekdays": " · werktags", "weekly": " · wöchentlich"}
|
||||||
|
return "\n".join(f"#{i['id']} · {i['when_local']}{rep.get(i['repeat'], '')} — {i['text']}" for i in items)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def reminder_delete(reminder_id: int) -> str:
|
||||||
|
"""Löscht eine Erinnerung/Routine endgültig (Nummer aus reminder_list). NUR auf
|
||||||
|
ausdrücklichen Wunsch des Commanders."""
|
||||||
|
try:
|
||||||
|
r = httpx.delete(f"{MC_URL}/api/reminders/{reminder_id}", timeout=30)
|
||||||
|
r.raise_for_status()
|
||||||
|
gone = r.json()["deleted"]
|
||||||
|
except httpx.HTTPStatusError as e:
|
||||||
|
return f"Fehlgeschlagen: {e.response.json().get('detail', e.response.text[:200])}"
|
||||||
|
return f"Erinnerung #{gone['id']} gelöscht ({gone['when_local']} — „{gone['text']}“)."
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def box_status() -> str:
|
||||||
|
"""Kurzer Gesundheits-Check der Box (KI-Server): Engine, Agent-Hirn, Kern-Dienste.
|
||||||
|
Nutze dies, wenn der Commander fragt, ob mit der Box/dem System alles in Ordnung ist."""
|
||||||
|
try:
|
||||||
|
h = _get("/api/health")
|
||||||
|
except Exception as exc:
|
||||||
|
return f"MC2 antwortet nicht ({exc}) — das Backend selbst könnte down sein."
|
||||||
|
brain = h.get("brain") or {}
|
||||||
|
lines = [
|
||||||
|
f"{'✅' if h.get('engine_reachable') else '❌'} Modell-Engine",
|
||||||
|
f"{'✅' if brain.get('ready') else '❌'} Agent-Hirn ({brain.get('model', '?')})",
|
||||||
|
]
|
||||||
|
try:
|
||||||
|
svcs = _get("/api/system/services").get("services", [])
|
||||||
|
lines += [f"{'✅' if s.get('ok') else '❌'} {s.get('name')}" for s in svcs]
|
||||||
|
except Exception:
|
||||||
|
lines.append("(Dienste-Liste nicht abrufbar)")
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
mcp.run()
|
||||||
Reference in New Issue
Block a user