577 lines
29 KiB
Python
577 lines
29 KiB
Python
"""MCP Client Manager (→ Konzept 7.2): Registrierung, Discovery, Freischaltung je Workspace,
|
||
Verbindungen der Workspaces und die entdeckten Tools als Teil des Tool-Katalogs.
|
||
|
||
Verbindungen eines Workspace: angelegt vom Workspace-Admin, nur dort nutzbar. Ist die URL schon
|
||
freigegeben (zentraler Server oder freigegebene Verbindung), ist die Verbindung sofort aktiv, sonst
|
||
wartet sie auf die Freigabe eines Plattform-Admins. Gleiche Tools desselben Dienstes (Fingerabdruck)
|
||
erben die Einstufung (app.tools.service.sync).
|
||
|
||
Die Tools werden aus dem Stand der letzten Discovery gebaut; ein Run ruft den
|
||
Server erst beim Tool-Aufruf an. Ein nicht erreichbarer Server führt so zu einem
|
||
Tool-Fehler, nicht zu einem abgebrochenen Run.
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from functools import partial
|
||
from typing import Any
|
||
from urllib.parse import urlsplit
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app import audit
|
||
from app.ai.service import LOCAL_HOSTS
|
||
from app.core.config import Settings
|
||
from app.core.crypto import Vault
|
||
from app.core.database import now
|
||
from app.core.errors import Forbidden, Invalid, NotFound
|
||
from app.data_sources.service import check_token
|
||
from app.identity import Principal
|
||
from app.identity.admin import require as require_platform_admin
|
||
from app.identity.models import User
|
||
from app.mcp import client
|
||
from app.mcp.models import APPROVED, PENDING, REJECTED, McpServer, McpTool, WorkspaceMcpServer
|
||
from app.tools.registry import Proposal, TableResult, Tool, ToolError
|
||
from app.workspaces import service as workspaces
|
||
|
||
NAME = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$")
|
||
TOOL_NAME_LIMIT = 64 # Vorgabe der Responses API für Funktionsnamen
|
||
COMBINATORS = {"anyOf", "oneOf", "allOf", "enum", "not", "const"}
|
||
|
||
|
||
# Art des Servers: allgemein, Gedächtnis (Munin, → Konzept 7.6) oder Datenbank-Treiber. Ein Treiber
|
||
# (DB-Dienst aus services/db-mcp) hat keine eigenen Daten; Workspaces legen damit Datenbanken an
|
||
# (app.mcp.databases). Treiber verwaltet die eigene Seite /admin/databases.
|
||
# Dokumenten-Auslese: liest Office, PDF und Bilder für abgeglichene Ordner aus (app.documents); welches Tool
|
||
# und welche Parameter, legt der Plattform-Admin in config fest
|
||
ROLES = {"": "Allgemein", "memory": "Gedächtnis (z. B. Munin)", "database": "Datenbank-Treiber",
|
||
"documents": "Dokumenten-Auslese"}
|
||
DATABASE, DOCUMENTS = "database", "documents"
|
||
SERVER_ROLES = {k: v for k, v in ROLES.items() if k != DATABASE}
|
||
HEADER_NAME = re.compile(r"^[A-Za-z0-9-]{1,64}$")
|
||
# Diese Header setzt Bifröst bzw. der Transport selbst
|
||
RESERVED_HEADERS = {"authorization", "host", "content-type", "content-length", "accept", "cookie", "connection",
|
||
"transfer-encoding", "mcp-protocol-version", "mcp-session-id"}
|
||
MAX_HEADERS = 20
|
||
|
||
def kind(server_name: str) -> str:
|
||
return f"mcp:{server_name}"
|
||
|
||
|
||
# --- Registrierung (Plattform-Admin) ---
|
||
|
||
def check_url(url: str, settings: Settings) -> str:
|
||
url = url.strip()
|
||
parts = urlsplit(url)
|
||
if parts.username or parts.password:
|
||
raise Invalid("Keine Zugangsdaten in der URL. Das Token wird separat hinterlegt.")
|
||
host = (parts.hostname or "").lower()
|
||
if parts.scheme == "http" and host in LOCAL_HOSTS and settings.mcp_allow_local:
|
||
return url
|
||
if parts.scheme != "https" or not host:
|
||
raise Invalid("Nur https:// (http:// nur für localhost mit AIW_MCP_ALLOW_LOCAL).")
|
||
if host not in settings.allowed_mcp_hosts:
|
||
raise Invalid(f"Host {host} ist nicht freigegeben (AIW_MCP_ALLOWED_HOSTS).")
|
||
return url
|
||
|
||
|
||
def servers(session: Session) -> list[McpServer]:
|
||
"""Zentrale Server ohne Datenbank-Treiber."""
|
||
return list(session.scalars(select(McpServer).where(McpServer.workspace_id.is_(None), McpServer.role != DATABASE)
|
||
.order_by(McpServer.name)))
|
||
|
||
|
||
def drivers(session: Session) -> list[McpServer]:
|
||
"""Datenbank-Treiber, auch inaktive."""
|
||
return list(session.scalars(select(McpServer).where(McpServer.workspace_id.is_(None), McpServer.role == DATABASE)
|
||
.order_by(McpServer.label)))
|
||
|
||
|
||
def origins(session: Session) -> dict[str, McpServer]:
|
||
"""Tool-Art → Server, für die Tool-Übersicht."""
|
||
return {kind(s.name): s for s in session.scalars(select(McpServer))}
|
||
|
||
|
||
def connections(session: Session, workspace_id: int | None = None) -> list[McpServer]:
|
||
"""Verbindungen der Workspaces (eines Workspace oder aller)."""
|
||
query = select(McpServer).where(McpServer.workspace_id.is_not(None))
|
||
if workspace_id is not None:
|
||
query = query.where(McpServer.workspace_id == workspace_id)
|
||
return list(session.scalars(query.order_by(McpServer.name)))
|
||
|
||
|
||
def check_name(session: Session, name: str) -> str:
|
||
"""Kurzname, Präfix der Tool-Namen; über alle Workspaces eindeutig (Einstufung und Regeln hängen daran)."""
|
||
name = name.strip().lower()
|
||
if not NAME.match(name) or not 2 <= len(name) <= 16:
|
||
raise Invalid("Kurzname: 2–16 Zeichen, Kleinbuchstaben, Ziffern und einzelne Unterstriche.")
|
||
if session.scalar(select(McpServer.id).where(McpServer.name == name)):
|
||
raise Invalid(f"Den Kurznamen {name} gibt es schon (auch andere Workspaces zählen); bitte einen anderen "
|
||
"wählen.")
|
||
return name
|
||
|
||
|
||
def check_headers(text: str) -> dict[str, str]:
|
||
"""Zeilen „Name: Wert“; leere Zeilen werden ignoriert."""
|
||
headers: dict[str, str] = {}
|
||
for line in text.splitlines():
|
||
if not line.strip():
|
||
continue
|
||
name, sep, value = line.partition(":")
|
||
name, value = name.strip(), value.strip()
|
||
if not sep or not HEADER_NAME.match(name) or not value:
|
||
raise Invalid(f"Header bitte als „Name: Wert“, je Zeile einer: {name[:40]!r}")
|
||
if name.lower() in RESERVED_HEADERS:
|
||
raise Invalid(f"Den Header {name} setzt Bifröst selbst; das Token gehört ins Feld Token.")
|
||
if any(ord(c) < 32 for c in value) or len(value) > 4000:
|
||
raise Invalid(f"Ungültiger Wert für {name}.")
|
||
headers[name] = value
|
||
if len(headers) > MAX_HEADERS:
|
||
raise Invalid(f"Höchstens {MAX_HEADERS} Header.")
|
||
return headers
|
||
|
||
|
||
def get(session: Session, server_id: int) -> McpServer:
|
||
server = session.get(McpServer, server_id)
|
||
if server is None:
|
||
raise NotFound
|
||
return server
|
||
|
||
|
||
def tools_of(session: Session, server_id: int) -> list[McpTool]:
|
||
return list(session.scalars(select(McpTool).where(McpTool.server_id == server_id).order_by(McpTool.name)))
|
||
|
||
|
||
def save(session: Session, principal: Principal, settings: Settings, vault: Vault, server_id: int | None, *,
|
||
name: str, label: str, url: str, token: str = "", remove_token: bool = False,
|
||
active: bool = True, role: str = "") -> McpServer:
|
||
require_platform_admin(principal)
|
||
url, token = check_url(url, settings), check_token(token)
|
||
if role not in ROLES:
|
||
raise Invalid("Unbekannte Art des Servers.")
|
||
if server_id is None:
|
||
server = McpServer(name=check_name(session, name), created_by=principal.user_id, label="", url="")
|
||
session.add(server)
|
||
else:
|
||
server = get(session, server_id)
|
||
if server.workspace_id is not None:
|
||
raise Invalid("Das ist eine Verbindung eines Workspace; sie wird dort bearbeitet.")
|
||
if (server.role == DATABASE) != (role == DATABASE):
|
||
raise Invalid("Ein Datenbank-Treiber bleibt ein Treiber und ein Server ein Server.")
|
||
if server.url and server.url != url:
|
||
server.status = "new" # neuer Endpunkt: Tools neu abrufen
|
||
for connection_ in session.scalars(select(McpServer).where(McpServer.template_id == server.id)):
|
||
connection_.url = url
|
||
server.label, server.url, server.active = label.strip()[:200] or server.name, url, active
|
||
server.role = role
|
||
if token:
|
||
server.secret, server.secret_hint = vault.encrypt(token), token[-4:]
|
||
elif remove_token:
|
||
server.secret, server.secret_hint = None, ""
|
||
session.flush()
|
||
audit.record(session, "mcp.server_saved", user_id=principal.user_id, target_kind="mcp_server",
|
||
target_id=server.id, name=server.name, url=url, active=active, role=role)
|
||
return server
|
||
|
||
|
||
def set_documents(session: Session, principal: Principal, server_id: int, tool: str, content: str,
|
||
filename: str = "") -> McpServer:
|
||
"""Dokumenten-Auslese festlegen: welches Tool, welcher Parameter bekommt den Inhalt (Base64), welcher den
|
||
Dateinamen. Geprüft gegen das Schema aus der letzten Discovery."""
|
||
require_platform_admin(principal)
|
||
server = get(session, server_id)
|
||
if server.role != DOCUMENTS or server.workspace_id is not None:
|
||
raise Invalid("Nur für zentrale Server mit der Art „Dokumenten-Auslese“.")
|
||
found = session.scalar(select(McpTool).where(McpTool.server_id == server.id, McpTool.remote_name == tool))
|
||
if found is None:
|
||
raise Invalid("Dieses Tool hat der Server bei der letzten Discovery nicht gemeldet.")
|
||
properties = (found.input_schema or {}).get("properties", {})
|
||
if content not in properties:
|
||
raise Invalid(f"Das Tool hat keinen Parameter {content!r}.")
|
||
if filename and (filename not in properties or filename == content):
|
||
raise Invalid(f"Ungültiger Parameter für den Dateinamen: {filename!r}.")
|
||
server.config = {**(server.config or {}), "tool": tool, "content": content, "filename": filename}
|
||
audit.record(session, "mcp.documents_configured", user_id=principal.user_id, target_kind="mcp_server",
|
||
target_id=server.id, tool=tool, content=content, filename=filename)
|
||
return server
|
||
|
||
|
||
def uses(session: Session, template_id: int) -> int:
|
||
"""Wie viele Datenbanken diesen Treiber nutzen."""
|
||
return len(list(session.scalars(select(McpServer.id).where(McpServer.template_id == template_id))))
|
||
|
||
|
||
def delete(session: Session, principal: Principal, server_id: int) -> None:
|
||
require_platform_admin(principal)
|
||
server = get(session, server_id)
|
||
if uses(session, server.id):
|
||
raise Invalid("Datenbanken von Workspaces nutzen diesen Treiber; erst diese entfernen.")
|
||
audit.record(session, "mcp.server_deleted", user_id=principal.user_id, target_kind="mcp_server",
|
||
target_id=server.id, name=server.name)
|
||
session.delete(server)
|
||
|
||
|
||
def target(server: McpServer, vault: Vault, settings: Settings, secret: str | None = None,
|
||
template: McpServer | None = None) -> client.Target:
|
||
"""Ziel eines Aufrufs. Token: eigenes des Workspace, sonst das des Servers bzw. bei Datenbanken der Schlüssel
|
||
des Treibers. Header: die eigenen und bei Datenbanken die Verbindungsdaten (X-DB-*)."""
|
||
encrypted = secret or server.secret or (template.secret if template else None)
|
||
headers = json.loads(vault.decrypt(server.headers)) if server.headers else {}
|
||
if template is not None:
|
||
from app.mcp import databases
|
||
headers = {**databases.headers(server), **headers}
|
||
url = template.url if template is not None else server.url
|
||
return client.Target(url, vault.decrypt(encrypted) if encrypted else None, settings.mcp_timeout_seconds, headers)
|
||
|
||
|
||
# --- Discovery ---
|
||
|
||
def model_name(server: McpServer, remote: str) -> str:
|
||
return f"{server.name}__{re.sub(r'[^A-Za-z0-9_-]', '_', remote)}"
|
||
|
||
|
||
def fingerprint(url: str, remote: str, description: str, schema: dict[str, Any]) -> str:
|
||
"""Gleiches Tool desselben Dienstes, unabhängig von Kurzname und Bezeichnung der Verbindung."""
|
||
data = json.dumps({"url": url.rstrip("/"), "tool": remote, "description": description, "schema": schema},
|
||
sort_keys=True, ensure_ascii=False)
|
||
return hashlib.sha256(data.encode()).hexdigest()
|
||
|
||
|
||
def normalize(schema: Any) -> dict[str, Any]:
|
||
"""Schema so, wie es die Responses API für Funktionen verlangt: oben ein Objekt."""
|
||
if not isinstance(schema, dict):
|
||
raise ValueError("kein Objekt")
|
||
if COMBINATORS & set(schema) or schema.get("type", "object") != "object":
|
||
raise ValueError("oben kein einfaches Objekt (anyOf/oneOf/…)")
|
||
schema = {k: v for k, v in schema.items() if k not in ("$schema", "title")}
|
||
return {"type": "object", "properties": {}, **schema}
|
||
|
||
|
||
def discover(session: Session, principal: Principal, settings: Settings, vault: Vault,
|
||
server_id: int) -> tuple[str, list[str]]:
|
||
"""Tools eines zentralen Servers abrufen (Plattform-Admin)."""
|
||
require_platform_admin(principal)
|
||
server = get(session, server_id)
|
||
error, notes = refresh(session, settings, vault, server, principal.user_id)
|
||
if not error and server.role == DATABASE:
|
||
from app.mcp import databases
|
||
databases.driver_refreshed(session, server)
|
||
return error, notes
|
||
|
||
|
||
def refresh(session: Session, settings: Settings, vault: Vault, server: McpServer,
|
||
user_id: int | None) -> tuple[str, list[str]]:
|
||
"""Tools abrufen und speichern. Liefert (Fehler, Hinweise zu übersprungenen Tools); ein Fehler
|
||
wird am Server vermerkt, nicht geworfen. Geänderte Tools verlieren über den Hash ihre Einstufung."""
|
||
template = session.get(McpServer, server.template_id) if server.template_id else None
|
||
try:
|
||
found = client.discover(target(server, vault, settings, template=template))
|
||
except ToolError as e:
|
||
server.status, server.last_error = "error", str(e)
|
||
audit.record(session, "mcp.discovery_failed", user_id=user_id, target_kind="mcp_server",
|
||
target_id=server.id, error=str(e)[:300])
|
||
return str(e), []
|
||
existing = {t.remote_name: t for t in tools_of(session, server.id)}
|
||
taken = set(session.scalars(select(McpTool.name).where(McpTool.server_id != server.id)))
|
||
notes, seen = [], set()
|
||
for remote in found.tools:
|
||
name = model_name(server, remote.name)
|
||
try:
|
||
schema = normalize(remote.input_schema)
|
||
except ValueError as e:
|
||
notes.append(f"{remote.name}: Schema nicht verwendbar ({e}).")
|
||
continue
|
||
if len(name) > TOOL_NAME_LIMIT or name in taken or name in seen:
|
||
notes.append(f"{remote.name}: Name zu lang oder nicht eindeutig.")
|
||
continue
|
||
seen.add(name)
|
||
row = existing.pop(remote.name, None) or McpTool(server_id=server.id, remote_name=remote.name)
|
||
row.name, row.description, row.input_schema = name, remote.description[:4000], schema
|
||
row.annotations = remote.annotations
|
||
row.fingerprint = fingerprint(server.url, remote.name, row.description, schema)
|
||
session.add(row)
|
||
for gone in existing.values():
|
||
session.delete(gone)
|
||
server.status, server.last_error, server.discovered_at = "ok", "", now()
|
||
server.server_info = found.server_info
|
||
session.flush()
|
||
audit.record(session, "mcp.discovered", user_id=user_id, target_kind="mcp_server",
|
||
target_id=server.id, tools=len(seen), skipped=len(notes))
|
||
return "", notes
|
||
|
||
|
||
# --- Tools für den Katalog ---
|
||
|
||
def suggest(annotations: dict[str, Any]) -> str:
|
||
"""Vorschlag aus den Annotations; der Server ist nicht vertrauenswürdig, entschieden wird von Hand."""
|
||
if annotations.get("readOnlyHint") is True:
|
||
return "read"
|
||
if annotations.get("destructiveHint") is True:
|
||
return "destructive"
|
||
return "write_external"
|
||
|
||
|
||
def run_tool(server_name: str, remote: str, context: Any, arguments: dict[str, Any]) -> str:
|
||
"""Ergebnis als Text; liefert der Server Spalten und Zeilen (structuredContent), zusätzlich als Tabelle
|
||
für Ansicht, CSV und Diagramm."""
|
||
target_ = context.mcp.get(server_name)
|
||
if target_ is None:
|
||
raise ToolError(f"Der MCP-Server {server_name} ist für diesen Workspace nicht freigeschaltet.")
|
||
failed, text, data = client.call_result(target_, remote, arguments)
|
||
if failed:
|
||
raise ToolError(f"Das Tool meldet einen Fehler: {re.sub(r'^Error executing tool [^:]+: ', '', text)[:1000]}")
|
||
return as_table(text, data, context.settings.query_max_rows)
|
||
|
||
|
||
def as_table(text: str, data: Any, max_rows: int) -> str:
|
||
if not isinstance(data, dict):
|
||
return text
|
||
columns, rows = data.get("columns"), data.get("rows")
|
||
if not (isinstance(columns, list) and isinstance(rows, list) and all(isinstance(c, str) for c in columns)
|
||
and all(isinstance(r, list) and len(r) == len(columns) for r in rows)):
|
||
return text
|
||
return TableResult(text, columns, rows[:max_rows], bool(data.get("truncated")) or len(rows) > max_rows)
|
||
|
||
|
||
def prepare_tool(server_name: str, remote: str, context: Any, arguments: dict[str, Any]) -> Proposal:
|
||
"""Freigabe für MCP-Tools: der Antrag zeigt die geprüften Argumente, die genau so gesendet werden."""
|
||
if server_name not in context.mcp:
|
||
raise ToolError(f"Der MCP-Server {server_name} ist für diesen Workspace nicht freigeschaltet.")
|
||
return Proposal(source_id=None, arguments=arguments,
|
||
preview=json.dumps(arguments, indent=2, ensure_ascii=False, sort_keys=True),
|
||
preconditions={"server": server_name, "tool": remote})
|
||
|
||
|
||
def execute_tool(server_name: str, remote: str, context: Any, snapshot: dict[str, Any]) -> tuple[str, dict]:
|
||
try:
|
||
return run_tool(server_name, remote, context, snapshot["arguments"]), {}
|
||
except ToolError as e:
|
||
# MCP-Aufrufe sind nicht idempotent: nie automatisch wiederholen
|
||
raise ToolError(f"{e} Ob die Aktion im Zielsystem ausgeführt wurde, ist unklar; bitte dort prüfen. "
|
||
"Sie wird nicht automatisch wiederholt.") from e
|
||
|
||
|
||
def catalog_tools(session: Session) -> dict[str, Tool]:
|
||
rows = session.execute(select(McpTool, McpServer).join(McpServer, McpServer.id == McpTool.server_id)
|
||
.where(McpServer.active.is_(True), McpServer.approval == APPROVED)).all()
|
||
result = {}
|
||
for row, server in rows:
|
||
bound = (server.name, row.remote_name)
|
||
result[row.name] = Tool(
|
||
row.name, kind(server.name), f"[{server.label}] {row.description}".strip(), None,
|
||
suggest(row.annotations), run=partial(run_tool, *bound), prepare=partial(prepare_tool, *bound),
|
||
execute=partial(execute_tool, *bound), input_schema=row.input_schema,
|
||
fingerprint=row.fingerprint or None)
|
||
return result
|
||
|
||
|
||
# --- Freischaltung je Workspace ---
|
||
|
||
def workspace_servers(session: Session, workspace_id: int) -> list[tuple[McpServer, WorkspaceMcpServer | None]]:
|
||
"""Zentrale Server zum Freischalten; Treiber nicht, die nutzt man über Datenbanken."""
|
||
links = {link.server_id: link for link in session.scalars(
|
||
select(WorkspaceMcpServer).where(WorkspaceMcpServer.workspace_id == workspace_id))}
|
||
return [(s, links.get(s.id)) for s in servers(session) if s.active or s.id in links]
|
||
|
||
|
||
def enable(session: Session, principal: Principal, vault: Vault, workspace_id: int, server_id: int,
|
||
token: str = "") -> None:
|
||
workspaces.require_admin(session, principal, workspace_id)
|
||
server = get(session, server_id)
|
||
if not server.active:
|
||
raise Invalid("Dieser Server ist deaktiviert.")
|
||
if server.workspace_id is not None or server.role == DATABASE:
|
||
raise Invalid("Dieser Server lässt sich nicht freischalten.")
|
||
token = check_token(token)
|
||
link = session.get(WorkspaceMcpServer, (workspace_id, server_id)) or WorkspaceMcpServer(
|
||
workspace_id=workspace_id, server_id=server_id, enabled_by=principal.user_id)
|
||
if token:
|
||
link.secret, link.secret_hint = vault.encrypt(token), token[-4:]
|
||
session.add(link)
|
||
audit.record(session, "mcp.workspace_enabled", user_id=principal.user_id, workspace_id=workspace_id,
|
||
target_kind="mcp_server", target_id=server_id, own_token=bool(token))
|
||
|
||
|
||
def disable(session: Session, principal: Principal, workspace_id: int, server_id: int) -> None:
|
||
workspaces.require_admin(session, principal, workspace_id)
|
||
link = session.get(WorkspaceMcpServer, (workspace_id, server_id))
|
||
if link is not None:
|
||
session.delete(link)
|
||
audit.record(session, "mcp.workspace_disabled", user_id=principal.user_id, workspace_id=workspace_id,
|
||
target_kind="mcp_server", target_id=server_id)
|
||
|
||
|
||
def enabled(session: Session, workspace_id: int) -> list[tuple[McpServer, WorkspaceMcpServer | None]]:
|
||
"""Was ein Run dieses Workspace nutzen darf: freigeschaltete zentrale Server und die eigenen,
|
||
freigegebenen Verbindungen (ohne Freischaltung, daher None)."""
|
||
central = list(session.execute(
|
||
select(McpServer, WorkspaceMcpServer).join(WorkspaceMcpServer, WorkspaceMcpServer.server_id == McpServer.id)
|
||
.where(WorkspaceMcpServer.workspace_id == workspace_id, McpServer.active.is_(True),
|
||
McpServer.workspace_id.is_(None), McpServer.role != DATABASE)).all())
|
||
own = session.scalars(select(McpServer).where(McpServer.workspace_id == workspace_id, McpServer.active.is_(True),
|
||
McpServer.approval == APPROVED).order_by(McpServer.name))
|
||
return [(s, link) for s, link in central] + [(s, None) for s in own]
|
||
|
||
|
||
def kinds(session: Session, workspace_id: int) -> set[str]:
|
||
return {kind(server.name) for server, _ in enabled(session, workspace_id)}
|
||
|
||
|
||
def targets(session: Session, vault: Vault, settings: Settings, workspace_id: int) -> dict[str, client.Target]:
|
||
"""Ziele für einen Run; ein eigenes Token des Workspace hat Vorrang vor dem des Servers."""
|
||
result = {}
|
||
for server, link in enabled(session, workspace_id):
|
||
template = session.get(McpServer, server.template_id) if server.template_id else None
|
||
result[server.name] = target(server, vault, settings, link.secret if link else None, template)
|
||
return result
|
||
|
||
|
||
def foreign_kinds(session: Session, workspace_id: int) -> set[str]:
|
||
"""Tool-Arten, die dieser Workspace nie sieht: Verbindungen anderer Workspaces und Treiber."""
|
||
rows = session.scalars(select(McpServer.name).where(
|
||
((McpServer.workspace_id.is_not(None)) & (McpServer.workspace_id != workspace_id))
|
||
| (McpServer.role == DATABASE)))
|
||
return {kind(name) for name in rows}
|
||
|
||
|
||
# --- Verbindungen eines Workspace ---
|
||
|
||
def approved_url(session: Session, url: str, exclude: int | None = None) -> bool:
|
||
"""Die Plattform nutzt diese URL schon: zentraler Server oder freigegebene Verbindung."""
|
||
query = select(McpServer.id).where(McpServer.url == url, McpServer.approval == APPROVED)
|
||
if exclude is not None:
|
||
query = query.where(McpServer.id != exclude)
|
||
return session.scalar(query.limit(1)) is not None
|
||
|
||
|
||
def connection(session: Session, principal: Principal, server_id: int) -> McpServer:
|
||
"""Verbindung eines Workspace, nur für dessen Admins (NotFound für Nichtmitglieder)."""
|
||
server = get(session, server_id)
|
||
if server.workspace_id is None:
|
||
raise NotFound
|
||
if not workspaces.get(session, principal, server.workspace_id).is_admin:
|
||
raise Forbidden
|
||
return server
|
||
|
||
|
||
def set_headers(server: McpServer, vault: Vault, headers: dict[str, str]) -> None:
|
||
server.headers = vault.encrypt(json.dumps(headers)) if headers else None
|
||
server.header_names = sorted(headers)
|
||
|
||
|
||
def create_connection(session: Session, principal: Principal, settings: Settings, vault: Vault, workspace_id: int,
|
||
*, name: str, label: str, url: str, token: str = "", headers: str = "",
|
||
exposure_confirmed: bool = False) -> McpServer:
|
||
workspaces.require_admin(session, principal, workspace_id)
|
||
if not exposure_confirmed:
|
||
raise Invalid("Bitte bestätigen, dass alle Mitglieder des Workspace sehen dürfen, was dieser Server liefert.")
|
||
name, url, token, parsed = check_name(session, name), check_url(url, settings), check_token(token), \
|
||
check_headers(headers)
|
||
server = McpServer(workspace_id=workspace_id, name=name, label=label.strip()[:200] or name, url=url,
|
||
approval=APPROVED if approved_url(session, url) else PENDING,
|
||
exposure_confirmed_by=principal.user_id, exposure_confirmed_at=now(),
|
||
created_by=principal.user_id)
|
||
if token:
|
||
server.secret, server.secret_hint = vault.encrypt(token), token[-4:]
|
||
set_headers(server, vault, parsed)
|
||
session.add(server)
|
||
session.flush()
|
||
audit.record(session, "mcp.connection_created", user_id=principal.user_id, workspace_id=workspace_id,
|
||
target_kind="mcp_server", target_id=server.id, name=name, url=url, approval=server.approval,
|
||
token=bool(token), headers=server.header_names)
|
||
audit.record(session, "source.exposure_confirmed", user_id=principal.user_id, workspace_id=workspace_id,
|
||
target_kind="mcp_server", target_id=server.id)
|
||
if server.approval == PENDING:
|
||
request_approval(session, server)
|
||
return server
|
||
|
||
|
||
def update_connection(session: Session, principal: Principal, vault: Vault, server_id: int, *, label: str,
|
||
token: str = "", remove_token: bool = False, headers: str = "", remove_headers: bool = False,
|
||
active: bool = True) -> McpServer:
|
||
"""Bezeichnung, Token, Header und aktiv. Die URL bleibt: eine andere URL ist eine neue Verbindung."""
|
||
server = connection(session, principal, server_id)
|
||
token, parsed = check_token(token), check_headers(headers)
|
||
server.label, server.active = label.strip()[:200] or server.name, active
|
||
if token:
|
||
server.secret, server.secret_hint = vault.encrypt(token), token[-4:]
|
||
elif remove_token:
|
||
server.secret, server.secret_hint = None, ""
|
||
if parsed:
|
||
set_headers(server, vault, parsed)
|
||
elif remove_headers:
|
||
set_headers(server, vault, {})
|
||
if token or remove_token or parsed or remove_headers:
|
||
server.status, server.status_message = "new", "Geändert, bitte neu testen."
|
||
audit.record(session, "mcp.connection_updated", user_id=principal.user_id, workspace_id=server.workspace_id,
|
||
target_kind="mcp_server", target_id=server.id, active=active,
|
||
token="ersetzt" if token else "entfernt" if remove_token else "unverändert",
|
||
headers=server.header_names)
|
||
return server
|
||
|
||
|
||
def delete_connection(session: Session, principal: Principal, server_id: int) -> int:
|
||
server = connection(session, principal, server_id)
|
||
workspace_id = server.workspace_id
|
||
audit.record(session, "mcp.connection_deleted", user_id=principal.user_id, workspace_id=workspace_id,
|
||
target_kind="mcp_server", target_id=server.id, name=server.name, url=server.url)
|
||
session.delete(server)
|
||
return workspace_id
|
||
|
||
|
||
def test_connection(session: Session, principal: Principal, settings: Settings, vault: Vault,
|
||
server_id: int) -> None:
|
||
"""Tools abrufen; bei Datenbanken zusätzlich die Verbindung zur Datenbank prüfen."""
|
||
server = connection(session, principal, server_id)
|
||
if server.approval != APPROVED:
|
||
server.status_message = "Wartet auf die Freigabe durch einen Plattform-Admin." \
|
||
if server.approval == PENDING else "Von einem Plattform-Admin abgelehnt."
|
||
return
|
||
if server.template_id:
|
||
from app.mcp import databases
|
||
databases.test(session, settings, vault, server, principal.user_id)
|
||
return
|
||
error, notes = refresh(session, settings, vault, server, principal.user_id)
|
||
count = len(tools_of(session, server.id))
|
||
server.status_message = error or f"{count} Tool{'s' if count != 1 else ''} gefunden." + (
|
||
" Übersprungen: " + " ".join(notes) if notes else "")
|
||
|
||
|
||
# --- Freigabe durch Plattform-Admins ---
|
||
|
||
def pending(session: Session) -> list[McpServer]:
|
||
return list(session.scalars(select(McpServer).where(McpServer.approval == PENDING).order_by(McpServer.created)))
|
||
|
||
|
||
def request_approval(session: Session, server: McpServer) -> None:
|
||
from app.notify import service as notify
|
||
from app.workspaces.models import Workspace
|
||
workspace = session.get(Workspace, server.workspace_id)
|
||
subject = f"MCP-Verbindung wartet auf Freigabe: {server.label} ({notify.clean(workspace.name, 80)})"
|
||
body = (f"Im Workspace „{workspace.name}“ wurde eine MCP-Verbindung zu einer neuen Adresse angelegt:\n\n"
|
||
f"{server.label} ({server.name})\n{server.url}\n\nErst nach der Freigabe nutzt Bifröst sie. Zur "
|
||
f"Entscheidung: {notify.link('/admin/mcp')}")
|
||
for admin in session.scalars(select(User).where(User.is_admin, User.active)):
|
||
notify.email(session, admin, "mcp.connection_requested", subject, body, workspace.id)
|
||
|
||
|
||
def decide(session: Session, principal: Principal, settings: Settings, vault: Vault, server_id: int,
|
||
approve: bool, note: str = "") -> McpServer:
|
||
"""Plattform-Admin gibt eine Verbindung frei (danach werden die Tools abgerufen) oder lehnt sie ab."""
|
||
require_platform_admin(principal)
|
||
server = get(session, server_id)
|
||
if server.workspace_id is None or server.approval != PENDING:
|
||
raise Invalid("Diese Verbindung wartet nicht auf eine Freigabe.")
|
||
server.approval, server.approval_note = (APPROVED if approve else REJECTED), note.strip()[:500]
|
||
audit.record(session, "mcp.connection_approved" if approve else "mcp.connection_rejected",
|
||
user_id=principal.user_id, workspace_id=server.workspace_id, target_kind="mcp_server",
|
||
target_id=server.id, url=server.url, note=server.approval_note)
|
||
if approve:
|
||
error, _ = refresh(session, settings, vault, server, principal.user_id)
|
||
count = len(tools_of(session, server.id))
|
||
server.status_message = error or f"Freigegeben, {count} Tool{'s' if count != 1 else ''} gefunden."
|
||
else:
|
||
server.status_message = "Abgelehnt" + (f": {server.approval_note}" if server.approval_note else ".")
|
||
return server
|