"""Synchroner Zugang zum offiziellen MCP-SDK für Web und Worker (→ Konzept 7.2). Jeder Vorgang öffnet eine eigene Verbindung (Streamable HTTP) und schließt sie wieder; so hält kein Prozess Zustand eines MCP-Servers. Fehler werden zu ToolError mit einer Meldung ohne Token oder interne Details. """ import asyncio import logging from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any import httpx2 from mcp import Client, MCPError from mcp.client.streamable_http import streamable_http_client from mcp.shared._httpx_utils import create_mcp_http_client from app.tools.registry import ToolError log = logging.getLogger(__name__) MAX_PAGES = 20 INSTRUCTIONS_CHARS = 8000 @dataclass(frozen=True) class Target: url: str token: str | None timeout: float # Zusätzliche Header je Aufruf, z. B. Zugangsdaten für den DB-Dienst headers: dict[str, str] = field(default_factory=dict, repr=False) @dataclass(frozen=True) class RemoteTool: name: str description: str input_schema: dict[str, Any] annotations: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class Discovery: tools: list[RemoteTool] server_info: dict[str, Any] # Austauschbar für Tests: liefert den Client für ein Ziel (z. B. einen Server im selben Prozess) def connect(target: Target) -> Client: headers = {**target.headers, **({"Authorization": f"Bearer {target.token}"} if target.token else {})} http = create_mcp_http_client(headers=headers, timeout=httpx2.Timeout(target.timeout)) return Client(streamable_http_client(target.url, http_client=http), read_timeout_seconds=target.timeout, cache=None) def run(target: Target, work: Callable[[Client], Awaitable[Any]]) -> Any: async def main() -> Any: async with connect(target) as client: return await asyncio.wait_for(work(client), target.timeout) try: return asyncio.run(main()) except (KeyboardInterrupt, SystemExit): raise except BaseException as raised: error = unwrap(raised) if isinstance(error, ToolError): raise error from raised if isinstance(error, TimeoutError): raise ToolError(f"Der MCP-Server antwortet nicht innerhalb von {target.timeout:g} Sekunden.") from raised if isinstance(error, MCPError): raise ToolError(f"Der MCP-Server meldet einen Fehler: {str(error)[:300]}") from raised log.warning("MCP-Server %s nicht erreichbar", target.url, exc_info=True) raise ToolError(f"Der MCP-Server ist nicht erreichbar ({describe(error)}).") from raised def unwrap(error: BaseException) -> BaseException: """Den eigentlichen Fehler aus (verschachtelten) ExceptionGroups holen.""" while isinstance(error, BaseExceptionGroup) and error.exceptions: error = error.exceptions[0] return error def describe(error: BaseException) -> str: """Kurze Ursache ohne URL-Bestandteile wie Tokens.""" if isinstance(error, httpx2.HTTPStatusError): return f"HTTP {error.response.status_code}" return type(error).__name__ def discover(target: Target) -> Discovery: async def work(client: Client) -> Discovery: tools, cursor = [], None for _ in range(MAX_PAGES): page = await client.list_tools(cursor=cursor) for tool in page.tools: annotations = tool.annotations.model_dump(exclude_none=True, by_alias=True) if tool.annotations else {} tools.append(RemoteTool(tool.name, tool.description or "", dict(tool.input_schema), annotations)) cursor = page.next_cursor if not cursor: break info = client.server_info server_info = {"name": info.name, "version": info.version} if info else {} if client.instructions: # Anleitung des Servers für den System-Prompt (→ Konzept 7.6); begrenzt server_info["instructions"] = client.instructions[:INSTRUCTIONS_CHARS] return Discovery(tools, server_info) return run(target, work) def call(target: Target, tool: str, arguments: dict[str, Any]) -> str: """Tool aufrufen; Ergebnis als Text. Meldet der Server isError, wird daraus ein ToolError.""" async def work(client: Client) -> str: result = await client.call_tool(tool, arguments) text = as_text(result) if result.is_error: raise ToolError(f"Das Tool meldet einen Fehler: {text[:1000]}") return text return run(target, work) def call_result(target: Target, tool: str, arguments: dict[str, Any]) -> tuple[bool, str, Any]: """Tool aufrufen; liefert isError, Text und strukturiertes Ergebnis (structuredContent).""" async def work(client: Client) -> tuple[bool, str, Any]: result = await client.call_tool(tool, arguments) return bool(result.is_error), as_text(result), getattr(result, "structured_content", None) return run(target, work) def as_text(result: Any) -> str: parts = [] for block in result.content: kind = getattr(block, "type", "") if kind == "text": parts.append(block.text) elif kind == "resource" and getattr(block.resource, "text", None): parts.append(block.resource.text) else: parts.append(f"[{kind or 'Inhalt'} ausgelassen]") if not parts and getattr(result, "structured_content", None) is not None: import json parts.append(json.dumps(result.structured_content, ensure_ascii=False)) return "\n".join(parts)