Add automatic per-request MCP selection
This commit is contained in:
Mikei386
@@ -0,0 +1,233 @@
|
||||
"""
|
||||
title: MikeAI Auto Tool Selector
|
||||
author: MikeAI
|
||||
version: 1.0.0
|
||||
description: Selects a small, relevant set of MCP servers for each user request.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class Filter:
|
||||
class Valves(BaseModel):
|
||||
priority: int = 25
|
||||
enabled: bool = True
|
||||
max_automatic_tools: int = 2
|
||||
show_selection_status: bool = True
|
||||
|
||||
TOOL_IDS = {
|
||||
"web": "server:mcp:web-local",
|
||||
"github": "server:mcp:github-local",
|
||||
"homeassistant": "server:mcp:homeassistant-local",
|
||||
"arr": "server:mcp:arr-local",
|
||||
"unraid": "server:mcp:unraid-readonly-local",
|
||||
"navidrome": "server:mcp:navidrome-local",
|
||||
"platform": "server:mcp:athena-platform",
|
||||
}
|
||||
|
||||
LABELS = {
|
||||
"web": "Web",
|
||||
"github": "GitHub",
|
||||
"homeassistant": "Home Assistant",
|
||||
"arr": "Sonarr/Radarr",
|
||||
"unraid": "Unraid-Diagnose",
|
||||
"navidrome": "Navidrome",
|
||||
"platform": "Athena-Plattformwissen",
|
||||
}
|
||||
|
||||
def __init__(self):
|
||||
self.valves = self.Valves()
|
||||
self.toggle = False
|
||||
|
||||
@staticmethod
|
||||
def _normalise(text: str) -> str:
|
||||
text = unicodedata.normalize("NFKC", text or "").casefold()
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
@staticmethod
|
||||
def _matches(text: str, patterns: tuple[str, ...]) -> bool:
|
||||
return any(re.search(pattern, text, flags=re.IGNORECASE) for pattern in patterns)
|
||||
|
||||
@staticmethod
|
||||
def _latest_user_text(body: dict) -> str:
|
||||
for message in reversed(body.get("messages") or []):
|
||||
if message.get("role") != "user":
|
||||
continue
|
||||
content = message.get("content", "")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
return " ".join(
|
||||
part.get("text", "")
|
||||
for part in content
|
||||
if isinstance(part, dict) and isinstance(part.get("text"), str)
|
||||
)
|
||||
return str(content)
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _add_system_rule(body: dict, labels: list[str]) -> None:
|
||||
rule = (
|
||||
"MikeAI automatic tool selection made these capabilities available for "
|
||||
f"this request: {', '.join(labels)}. Availability is not authorization. "
|
||||
"First inspect read-only state. Never perform a state-changing action "
|
||||
"unless the user's current message explicitly requests that exact change "
|
||||
"and every required preview, confirmation, backup, and validation rule "
|
||||
"of the tool is satisfied. Do not call unrelated tools merely because "
|
||||
"they are available. If the selected tool cannot verify the claim, say so."
|
||||
)
|
||||
messages = body.setdefault("messages", [])
|
||||
for message in messages:
|
||||
if message.get("role") == "system" and isinstance(message.get("content"), str):
|
||||
message["content"] += "\n\n" + rule
|
||||
return
|
||||
messages.insert(0, {"role": "system", "content": rule})
|
||||
|
||||
def _classify(self, raw_text: str) -> list[str]:
|
||||
text = self._normalise(raw_text)
|
||||
if not text:
|
||||
return []
|
||||
|
||||
selected: list[str] = []
|
||||
|
||||
platform = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bathena\b", r"\bmikeai\b", r"\bki[- ]host\b",
|
||||
r"\bai[- ]profile[- ]router\b", r"\bprofil[- ]router\b",
|
||||
r"\brecovery[- ](?:koffer|bundle|skript)\b",
|
||||
r"\b(?:disaster|bare metal)[- ]recovery\b",
|
||||
r"\b(?:installations?|reinstall|setup)[- ]skript\b",
|
||||
r"\bplattform(?:wissen|dokumentation)?\b",
|
||||
),
|
||||
)
|
||||
homeassistant = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bhome\s*assistant\b", r"\bhomeassistant\b", r"\bhass\b",
|
||||
r"\bautomatisierung(?:en)?\b", r"\bentit[aä]t(?:en)?\b",
|
||||
r"\b(?:sensor|light|switch|climate|automation)\.[\w.-]+",
|
||||
r"\b(?:temperatur|luftfeuchtigkeit|wie warm)\b.*\b"
|
||||
r"(?:k[uü]che|wohnzimmer|schlafzimmer|bad|toilette|keller|haus)\b",
|
||||
),
|
||||
)
|
||||
arr = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bsonarr\b", r"\bradarr\b", r"\barr[- ]mcp\b",
|
||||
r"\b(?:staffel|episode|folge)(?:n)?\b.*\b"
|
||||
r"(?:fehl|such|download|indexer|release|serie)\w*",
|
||||
r"\b(?:serie|film)\b.*\b(?:sonarr|radarr|indexer|release)\b",
|
||||
),
|
||||
)
|
||||
navidrome = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bnavidrome\b", r"\bmusikbibliothek\b", r"\bplaylist\b",
|
||||
r"\b(?:titel|song|track)\b.*\b(?:[aä]hnlich|empfehl|playlist|bibliothek)\w*",
|
||||
r"\b(?:album|k[uü]nstler|artist)\b.*\b(?:navidrome|bibliothek|playlist)\b",
|
||||
),
|
||||
)
|
||||
unraid = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bunraid\b", r"\bhome\s*server\b", r"\bhomeserver\b",
|
||||
r"\bparity\b", r"\barray\b", r"\bmua\b",
|
||||
r"\bdocker[- ]container\b", r"\bcontainer(?:n)?\b.*\b"
|
||||
r"(?:docker|log|update|l[aä]uft|laufen|gestoppt|unraid)\b",
|
||||
r"\b(?:docker|container)\b.*\b(?:unraid|homeserver|nas)\b",
|
||||
),
|
||||
)
|
||||
github = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\bgithub\b", r"github\.com/", r"\brepositor(?:y|ium|ies)\b",
|
||||
r"\bgit[- ]repo\b", r"\bsource[- ]tree\b",
|
||||
r"\b(?:readme|quellcode|sourcecode|code)\b.*\b(?:repo|github)\b",
|
||||
),
|
||||
)
|
||||
|
||||
# A specialist source is more precise than public web search. Platform
|
||||
# wins over a generic Docker mention when Athena is explicitly named.
|
||||
if platform:
|
||||
selected.append("platform")
|
||||
elif homeassistant:
|
||||
selected.append("homeassistant")
|
||||
elif arr:
|
||||
selected.append("arr")
|
||||
elif navidrome:
|
||||
selected.append("navidrome")
|
||||
elif unraid:
|
||||
selected.append("unraid")
|
||||
elif github:
|
||||
selected.append("github")
|
||||
|
||||
explicit_web = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\b(?:im|ins|das|dem) (?:internet|netz|web)\b",
|
||||
r"\bwebsuche\b", r"\bonline (?:such|nachschau|recherch|pr[uü]f)\w*",
|
||||
r"\b(?:internet|web) (?:such|nachschau|recherch|pr[uü]f)\w*",
|
||||
),
|
||||
)
|
||||
current_public = self._matches(
|
||||
text,
|
||||
(
|
||||
r"\b(?:wetter|regen|regnet|regnen|regenradar|vorhersage)\b",
|
||||
r"\b(?:nachrichten|news|schlagzeilen)\b",
|
||||
r"\b(?:preis|preise|verf[uü]gbar|verf[uü]gbarkeit)\b",
|
||||
r"\b(?:neueste|neuestes|neuerungen|release)\b",
|
||||
),
|
||||
)
|
||||
if (explicit_web or (current_public and not selected)) and "web" not in selected:
|
||||
selected.append("web")
|
||||
|
||||
# A direct GitHub reference should still use GitHub even if broader web
|
||||
# research is also requested.
|
||||
if github and "github" not in selected:
|
||||
selected.insert(0, "github")
|
||||
|
||||
return selected[: max(0, self.valves.max_automatic_tools)]
|
||||
|
||||
async def _notify(self, emitter, labels: list[str]) -> None:
|
||||
if emitter is None or not self.valves.show_selection_status:
|
||||
return
|
||||
try:
|
||||
await emitter(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {
|
||||
"description": "Automatisch bereitgestellt: " + ", ".join(labels),
|
||||
"done": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def inlet(self, body: dict, __event_emitter__=None, **kwargs) -> dict:
|
||||
if not self.valves.enabled:
|
||||
return body
|
||||
|
||||
selected = self._classify(self._latest_user_text(body))
|
||||
if not selected:
|
||||
return body
|
||||
|
||||
existing = body.get("tool_ids")
|
||||
if not isinstance(existing, list):
|
||||
existing = []
|
||||
merged = list(existing)
|
||||
for category in selected:
|
||||
tool_id = self.TOOL_IDS[category]
|
||||
if tool_id not in merged:
|
||||
merged.append(tool_id)
|
||||
body["tool_ids"] = merged
|
||||
|
||||
labels = [self.LABELS[category] for category in selected]
|
||||
self._add_system_rule(body, labels)
|
||||
await self._notify(__event_emitter__, labels)
|
||||
return body
|
||||
@@ -9,7 +9,7 @@ ACTION_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/actions" && pwd)
|
||||
|
||||
die() { printf 'FEHLER: %s\n' "$*" >&2; exit 1; }
|
||||
[[ $EUID -eq 0 ]] || die "Bitte als root ausführen."
|
||||
for file in reasoning_default_off.py thinking.py stability_guard.py secret_redaction.py spoken_tool_status.py local_performance_metrics.py; do
|
||||
for file in reasoning_default_off.py thinking.py auto_tool_selector.py stability_guard.py secret_redaction.py spoken_tool_status.py local_performance_metrics.py; do
|
||||
[[ -s $FILTER_DIR/$file ]] || die "Filterdatei fehlt: $file"
|
||||
done
|
||||
[[ -s $ACTION_DIR/quick_actions.py ]] || die "Actiondatei fehlt: quick_actions.py"
|
||||
@@ -67,9 +67,9 @@ if not {"key", "value", "updated_at"} <= config_columns:
|
||||
owner = requested_owner
|
||||
if not owner:
|
||||
existing = con.execute(
|
||||
"select user_id from function where id in (?, ?, ?, ?, ?, ?, ?) order by id limit 1",
|
||||
"select user_id from function where id in (?, ?, ?, ?, ?, ?, ?, ?) order by id limit 1",
|
||||
(
|
||||
"reasoning_default_off", "thinking", "stability_guard",
|
||||
"reasoning_default_off", "thinking", "auto_tool_selector", "stability_guard",
|
||||
"secret_redaction", "spoken_tool_status", "local_performance_metrics",
|
||||
"quick_actions",
|
||||
),
|
||||
@@ -88,6 +88,11 @@ now = int(time.time())
|
||||
functions = [
|
||||
("reasoning_default_off", "Reasoning Default Off", "filter", 10, filter_dir, ""),
|
||||
("thinking", "Thinking", "filter", 20, filter_dir, ""),
|
||||
(
|
||||
"auto_tool_selector", "MikeAI Auto Tool Selector", "filter", 25, filter_dir,
|
||||
"Stellt pro Anfrage höchstens zwei passende MCP-Werkzeuge bereit. "
|
||||
"Die Auswahl ist keine Freigabe für schreibende Aktionen.",
|
||||
),
|
||||
("stability_guard", "MikeAI Stability Guard", "filter", 30, filter_dir, ""),
|
||||
("secret_redaction", "MikeAI Secret Redaction", "filter", 40, filter_dir, ""),
|
||||
(
|
||||
@@ -312,7 +317,7 @@ with con:
|
||||
),
|
||||
)
|
||||
print(
|
||||
"OpenWebUI konfiguriert: Default Off=10, Thinking=20, "
|
||||
"OpenWebUI konfiguriert: Default Off=10, Thinking=20, Auto Tool Selector=25, "
|
||||
"Stability Guard=30, Secret Redaction=40, Spoken Tool Status=80, Local Metrics=90, "
|
||||
"Quick Actions=100, Folgefragen=aus, Piper-TTS=aktiv, Werkzeugwahl=optimiert"
|
||||
)
|
||||
|
||||
@@ -93,17 +93,16 @@ now = int(time.time())
|
||||
filter_ids = [
|
||||
"reasoning_default_off",
|
||||
"thinking",
|
||||
"auto_tool_selector",
|
||||
"stability_guard",
|
||||
"secret_redaction",
|
||||
"local_performance_metrics",
|
||||
]
|
||||
|
||||
# Open WebUI addresses a global MCP server as server:mcp:<connection-id>.
|
||||
# Attach the bounded web relay and, when its dedicated secret exists, the four
|
||||
# read-only official GitHub tools. This is not Open WebUI's built-in web search.
|
||||
default_tool_ids = ["server:mcp:web-local"]
|
||||
if github_enabled:
|
||||
default_tool_ids.append("server:mcp:github-local")
|
||||
# The Auto Tool Selector attaches a bounded, relevant MCP set per request.
|
||||
# Keeping this list empty prevents unrelated schemas from consuming context.
|
||||
# Users can still enable additional tools manually in a chat.
|
||||
default_tool_ids = []
|
||||
|
||||
def capabilities(vision: bool) -> dict:
|
||||
return {
|
||||
@@ -310,8 +309,7 @@ with con:
|
||||
"tags": [{"name": tag} for tag in profile["tags"]],
|
||||
"toolIds": default_tool_ids,
|
||||
# Built-in features remain available but are not forced on every
|
||||
# request. Only the bounded web MCP and four read-only GitHub
|
||||
# repository tools are attached by default.
|
||||
# request. The Auto Tool Selector supplies relevant MCPs per turn.
|
||||
"defaultFeatureIds": [],
|
||||
"filterIds": filter_ids,
|
||||
"actionIds": ["quick_actions"],
|
||||
|
||||
Reference in New Issue
Block a user