Hello World MCP FastAPI Endpoint

import re
from datetime import datetime, timedelta
from typing import Optional

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCP
from starlette.middleware.proxy_headers import ProxyHeadersMiddleware

app = FastAPI(title="Hello World MCP")

# Trust Cloud Run/GFE proxy so scheme/host are correct in redirects
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])

# --- MCP server over Streamable HTTP ---
mcp = FastMCP("HelloWorldMCP", stateless_http=True)


@mcp.tool()
def ping() -> str:
    """Connectivity test tool for Agent Builder."""
    return "pong"


@mcp.tool()
def server_time(fmt: Optional[str] = None) -> str:
    """Return current UTC time. Optional strftime format."""
    now = datetime.utcnow()
    return now.strftime(fmt or "%Y-%m-%dT%H:%M:%SZ")


@mcp.tool()
def echo(text: str) -> str:
    """Echo back the provided text."""
    return text


@mcp.tool()
def date_math(expr: str) -> str:
    """
    Apply simple relative time offsets to UTC 'now'.
    Examples: '+2h', '-15m', '+1d 30m', '-3h 10m'.
    Returns ISO8601 UTC string.
    """
    now = datetime.utcnow()
    delta = timedelta()
    for token in re.findall(r"([+-]?\d+)\s*([smhdw])", expr.lower()):
        val = int(token[0])
        unit = token[1]
        if unit == "s":
            delta += timedelta(seconds=val)
        elif unit == "m":
            delta += timedelta(minutes=val)
        elif unit == "h":
            delta += timedelta(hours=val)
        elif unit == "d":
            delta += timedelta(days=val)
        elif unit == "w":
            delta += timedelta(weeks=val)
    if expr.strip().startswith("-") and delta > timedelta(0):
        return (now - delta).strftime("%Y-%m-%dT%H:%M:%SZ")
    return (now + delta).strftime("%Y-%m-%dT%H:%M:%SZ")


# Build the MCP ASGI app
mcp_app = mcp.streamable_http_app()

# Wrap the sub-app with proxy headers middleware (ASGI-safe)
mcp_app = ProxyHeadersMiddleware(mcp_app, trusted_hosts=["*"])

# Wrap the sub-app with CORS middleware exposing Mcp-Session-Id
mcp_app = CORSMiddleware(
    app=mcp_app,
    allow_origins=["*"],  # tighten later
    allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
    allow_headers=["*"],
    expose_headers=["Mcp-Session-Id"],
)

# Disable 307 slash redirects only if the router supports it
if hasattr(mcp_app, "router") and hasattr(mcp_app.router, "redirect_slashes"):
    mcp_app.router.redirect_slashes = False

# Mount MCP at /mcp
app.mount("/mcp", mcp_app)

# Optional: CORS for top-level REST routes (keep if needed by your UI)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["*"],
)


# --- Regular REST routes ---
@app.get("/")
def root() -> dict[str, str]:
    return {"message": "Hello World MCP server is running"}


@app.get("/healthz")
def health() -> dict[str, bool]:
    return {"ok": True}