Hitonabi
47f7a85510
Die Ampel-Nachruestung deckte 317/337 vorbestehende ruff-Verstoesse im ganzen Repo auf. Aufgeraeumt: - ruff.toml: intentionale Muster als Projekt-Politik ausgenommen (BLE001 blind-except, S110/S112 try-except-pass/continue, PLW1510 subprocess-best-effort, B008 FastAPI- Depends/File-Idiom, EXE001 Shebang, + wenige Stil-Regeln). __init__.py-Re-Exports geschuetzt (F401). - ruff --fix: 128 mechanische (Import-Sortierung, PEP585/604-Annotationen, tote Imports, ueberfluessige noqa) auto-behoben. - 12 echte Reste von Hand: PERF402/102, PLC3002 (Lambda->walrus), ISC004 (String-Concat geklammert), F841/RUF059 (ungenutzte Vars), PIE810 (startswith-Tuple), UP031 (f-string), UP035 (veraltete typing-Imports). Ergebnis: 'ruff check .' = 0, 'compileall' grün. Kein Verhaltenswechsel (nur Stil/Modernisierung). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mission Control 2.0 — Web Fetch MCP Server.
|
|
|
|
Stdio-MCP-Server. Läuft als Subprocess von Hermes Gateway (AI Box).
|
|
Fetcht URLs und extrahiert sauberen Text — kein API-Key nötig.
|
|
"""
|
|
|
|
import httpx
|
|
from guard import wrap_untrusted # Injection-Schutz für untrusted Web-Inhalt (Stufe 0 Security)
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("hermes-web-fetch")
|
|
|
|
HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
)
|
|
}
|
|
|
|
|
|
@mcp.tool()
|
|
def fetch_url(url: str, max_chars: int = 8000) -> str:
|
|
"""Lädt eine URL und extrahiert den Haupttext (Artikel, News, Seiteninhalte).
|
|
Nutze dies wenn du den vollen Inhalt einer Webseite lesen willst.
|
|
Gibt sauberen Text zurück — kein HTML-Rauschen.
|
|
max_chars: maximale Zeichenanzahl des Ergebnisses (default 8000)."""
|
|
try:
|
|
import trafilatura
|
|
|
|
resp = httpx.get(url, headers=HEADERS, timeout=20, follow_redirects=True)
|
|
resp.raise_for_status()
|
|
|
|
text = trafilatura.extract(
|
|
resp.text,
|
|
include_comments=False,
|
|
include_tables=True,
|
|
no_fallback=False,
|
|
)
|
|
|
|
if not text:
|
|
# Fallback: strip HTML tags manually
|
|
import re
|
|
text = re.sub(r"<[^>]+>", " ", resp.text)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
|
|
if len(text) > max_chars:
|
|
text = text[:max_chars] + f"\n\n[... gekürzt auf {max_chars} Zeichen]"
|
|
|
|
# Untrusted Web-Inhalt als DATEN markieren (+ Warnung bei Injection-Mustern).
|
|
return wrap_untrusted(text or "Kein Text extrahierbar.", label=f"WEBSEITE {url}")
|
|
|
|
except httpx.TimeoutException:
|
|
return f"FEHLER: Timeout beim Laden von {url}"
|
|
except httpx.HTTPStatusError as e:
|
|
return f"FEHLER: HTTP {e.response.status_code} für {url}"
|
|
except Exception as e:
|
|
return f"FEHLER: {e}"
|
|
|
|
|
|
@mcp.tool()
|
|
def web_search(query: str, max_results: int = 6) -> str:
|
|
"""Websuche (DuckDuckGo, kein API-Key). Liefert Titel, URL und Kurzbeschreibung je
|
|
Treffer. Nutze dies für aktuelle Informationen, Dokumentation oder Fehlermeldungen —
|
|
und lies vielversprechende Treffer danach mit fetch_url im Volltext."""
|
|
try:
|
|
from ddgs import DDGS
|
|
|
|
rows = list(DDGS().text(query, max_results=max_results) or [])
|
|
if not rows:
|
|
return "Keine Treffer."
|
|
lines = [f"- {r.get('title', '(ohne Titel)')}\n {r.get('href', '')}\n {r.get('body', '')}"
|
|
for r in rows]
|
|
return wrap_untrusted("\n".join(lines), label=f"WEBSUCHE {query}")
|
|
except Exception as e:
|
|
return f"FEHLER: {e}"
|
|
|
|
|
|
@mcp.tool()
|
|
def fetch_urls(urls: list[str], max_chars_each: int = 4000) -> str:
|
|
"""Lädt mehrere URLs gleichzeitig und gibt die Texte zurück.
|
|
Nützlich um mehrere Suchergebnisse auf einmal zu lesen.
|
|
urls: Liste von URLs (max 5)
|
|
max_chars_each: max Zeichen pro Seite"""
|
|
if len(urls) > 5:
|
|
urls = urls[:5]
|
|
|
|
results = []
|
|
for url in urls:
|
|
text = fetch_url(url, max_chars_each)
|
|
results.append(f"=== {url} ===\n{text}")
|
|
|
|
return "\n\n".join(results)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|