agent-platform/mcp_tools/server.py

"""MCP-Tool-Server für die Agent Platform."""
import os
import ast
import operator
import ipaddress
import socket
from datetime import datetime, timezone, timedelta
from urllib.parse import urlparse

import httpx
from fastmcp import FastMCP


mcp = FastMCP("Agent Platform Tools")


@mcp.tool()
def echo(text: str) -> dict:
    """Echo-Tool zum Testen der MCP-Verbindung."""
    return {"input": text, "timestamp": datetime.utcnow().isoformat()}


@mcp.tool()
def get_time(timezone_name: str = "UTC") -> dict:
    """Gibt aktuelle Zeit in angegebener Zeitzone zurück."""
    tz_offsets = {"UTC": 0, "CET": 1, "CEST": 2, "EST": -5, "PST": -8, "JST": 9}
    offset = tz_offsets.get(timezone_name.upper(), 0)
    tz = timezone(timedelta(hours=offset))
    return {"timezone": timezone_name, "time": datetime.now(tz).isoformat()}


@mcp.tool()
def calculate(expression: str) -> dict:
    """Berechnet einen mathematischen Ausdruck (sicher, nur Zahlen).

    Erlaubt: +, -, *, /, **, %.
    Beispiel: calculate("2 + 3 * 4") -> 14
    """
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.Mod: operator.mod,
    }

    try:
        tree = ast.parse(expression, mode="eval")
    except SyntaxError:
        return {"error": "Ungültiger Ausdruck"}

    def _eval(node):
        if isinstance(node, ast.Constant):
            if not isinstance(node.value, (int, float)):
                raise ValueError("Nur Zahlen erlaubt")
            return node.value
        if isinstance(node, ast.BinOp):
            if type(node.op) not in ops:
                raise ValueError("Operation nicht erlaubt")
            return ops[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp):
            if isinstance(node.op, ast.USub):
                return -_eval(node.operand)
            if isinstance(node.op, ast.UAdd):
                return _eval(node.operand)
        raise ValueError("Ungültiger Knoten")

    try:
        result = _eval(tree.body)
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"expression": expression, "error": str(e)}


@mcp.tool()
def http_get(url: str, timeout: int = 10) -> dict:
    """Holt eine URL via HTTP. SSRF-Schutz aktiv.

    Nur http/https, keine Auth, keine privaten IPs.
    Beispiel: http_get("https://example.com")
    """
    if not url.startswith(("http://", "https://")):
        return {"error": "Nur http/https URLs erlaubt"}

    parsed = urlparse(url)
    try:
        ip = socket.gethostbyname(parsed.hostname)
        ip_obj = ipaddress.ip_address(ip)
        if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved:
            return {"error": f"IP {ip} blockiert (privat/loopback/reserved)"}
    except Exception:
        pass

    try:
        with httpx.Client(timeout=timeout, follow_redirects=True) as client:
            r = client.get(url)
            return {
                "status": r.status_code,
                "url": str(r.url),
                "content_type": r.headers.get("content-type", ""),
                "length": len(r.text),
                "text_preview": r.text[:2000],
            }
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def list_env(prefix: str = "") -> dict:
    """Listet Umgebungsvariablen mit optionalem Prefix-Filter.

    Sensitive Werte (KEY/SECRET/TOKEN/PASSWORD) werden maskiert.
    """
    env = {}
    for k, v in os.environ.items():
        if prefix and not k.startswith(prefix.upper()):
            continue
        upper = k.upper()
        if any(s in upper for s in ["KEY", "SECRET", "TOKEN", "PASSWORD"]):
            env[k] = "***MASKED***"
        else:
            env[k] = v[:200]
    return {"count": len(env), "vars": env}


@mcp.tool()
def json_format(data: str, indent: int = 2) -> dict:
    """Formatiert einen JSON-String lesbar.

    Beispiel: json_format('{"a":1}')
    """
    try:
        parsed = json.loads(data)
        return {"formatted": json.dumps(parsed, indent=indent, ensure_ascii=False)}
    except Exception as e:
        return {"error": str(e)}


if __name__ == "__main__":
    host = os.getenv("MCP_HOST", "0.0.0.0")
    port = int(os.getenv("MCP_PORT", "8501"))
    mcp.run(transport="streamable-http", host=host, port=port)