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}