mg-mcp/email_server.py

"""
mg-mcp — Mailgun MCP server (Streamable HTTPS).

Public URL once deployed: https://optical-dev.oliver.solutions/mg-mcp/

Layout inside this container (FastAPI):
  - GET /api/health        → 200 (no auth, used by deploy script + Apache health proxy)
  - mount("/", FastMCP)    → MCP JSON-RPC streamable-HTTP endpoint at /mcp
                              (clients call POST /mcp with the protocol)
All endpoints except /api/health require `Authorization: Bearer <MCP_BEARER_KEY>`.

Apache strips the `/mg-mcp/` URL prefix before proxying, so externally:
  https://optical-dev.oliver.solutions/mg-mcp/mcp        → MCP endpoint
  https://optical-dev.oliver.solutions/mg-mcp/api/health → health
"""

import os
import sys
from contextlib import asynccontextmanager

import requests
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
from starlette.middleware.base import BaseHTTPMiddleware


def _require_env(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        sys.stderr.write(f"FATAL: {name} not set in environment.\n")
        sys.exit(1)
    return val


MAILGUN_API_KEY = _require_env("MAILGUN_API_KEY")
MAILGUN_DOMAIN = _require_env("MAILGUN_DOMAIN")
MAILGUN_FROM = _require_env("MAILGUN_FROM")
MCP_BEARER_KEY = _require_env("MCP_BEARER_KEY")

# Comma-separated list of public hostnames the MCP server is reachable as.
# Required because the SDK's DNS-rebinding protection rejects requests whose
# Host header isn't in its allowlist (default only covers 127.0.0.1/localhost).
# When deployed behind Apache with ProxyPreserveHost On, the container sees
# the real public hostname and it must be allow-listed here.
PUBLIC_HOSTS = [
    h.strip()
    for h in os.environ.get("PUBLIC_HOSTS", "optical-dev.oliver.solutions").split(",")
    if h.strip()
]

HEALTH_PATH = "/api/health"


class BearerAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path == HEALTH_PATH:
            return await call_next(request)

        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return JSONResponse({"error": "Missing Bearer token"}, status_code=401)
        if auth[len("Bearer "):].strip() != MCP_BEARER_KEY:
            return JSONResponse({"error": "Invalid API key"}, status_code=401)
        return await call_next(request)


_allowed_hosts = ["127.0.0.1", "127.0.0.1:*", "localhost", "localhost:*"] + PUBLIC_HOSTS
_allowed_origins = [f"https://{h}" for h in PUBLIC_HOSTS] + [f"http://{h}" for h in PUBLIC_HOSTS]

mcp = FastMCP(
    "mg-mcp",
    stateless_http=True,
    transport_security=TransportSecuritySettings(
        allowed_hosts=_allowed_hosts,
        allowed_origins=_allowed_origins,
    ),
)


@mcp.tool()
def send_email(
    to_email: str,
    subject: str,
    body: str = "",
    html_body: str = "",
) -> str:
    """
    Send a plain-text or HTML email via Mailgun.

    Provide `body` for plain text, `html_body` for HTML, or both (recommended —
    clients that don't render HTML fall back to the text version).

    Args:
        to_email:  Recipient email address.
        subject:   Email subject line.
        body:      Plain-text body. Optional if `html_body` is provided.
        html_body: HTML body. Optional if `body` is provided.

    Returns:
        Human-readable success or failure string.
    """
    if not body and not html_body:
        return "Error: provide at least one of `body` or `html_body`."

    data = {
        "from": MAILGUN_FROM,
        "to": [to_email],
        "subject": subject,
    }
    if body:
        data["text"] = body
    if html_body:
        data["html"] = html_body

    r = requests.post(
        f"https://api.mailgun.net/v3/{MAILGUN_DOMAIN}/messages",
        auth=("api", MAILGUN_API_KEY),
        data=data,
        timeout=30,
    )

    if r.status_code == 200:
        parts = (["plain text"] if body else []) + (["HTML"] if html_body else [])
        return f"Email sent to {to_email} ({' + '.join(parts)})"
    return f"Mailgun error {r.status_code}: {r.text}"


# FastMCP's streamable-HTTP session manager owns an anyio task group that
# must be started for the lifetime of the app. Mounted sub-apps don't get
# their own lifespan run, so we nest it in the parent FastAPI lifespan.
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with mcp.session_manager.run():
        yield


app = FastAPI(title="mg-mcp", description="Mailgun MCP server", lifespan=lifespan)
app.add_middleware(BearerAuthMiddleware)


@app.get(HEALTH_PATH)
async def health():
    return {"status": "ok", "service": "mg-mcp"}


# Order matters: register the health route before mounting FastMCP so the
# route resolver matches /api/health first. The mount catches everything else
# and exposes the MCP streamable-HTTP endpoint at /mcp.
app.mount("/", mcp.streamable_http_app())


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)