app.py

# app.py
import json
import logging
from datetime import datetime, timezone
from typing import Dict

import gradio as gr

# --------------------------
# Logging setup (quiet + precise)
# --------------------------
class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # Attach any extra dict passed via logger.extra
        for k, v in getattr(record, "__dict__", {}).items():
            if k not in payload and k not in (
                "args", "asctime", "created", "exc_info", "exc_text", "filename",
                "funcName", "levelno", "lineno", "module", "msecs", "msg",
                "name", "pathname", "process", "processName", "relativeCreated",
                "stack_info", "thread", "threadName"
            ):
                payload[k] = v
        return json.dumps(payload, ensure_ascii=False)

def setup_logging() -> logging.Logger:
    root = logging.getLogger()
    root.setLevel(logging.WARNING)  # default higher; opt-in with our logger

    # Silence noisy libs
    logging.getLogger("httpx").setLevel(logging.ERROR)          # suppress asset GETs
    logging.getLogger("uvicorn").setLevel(logging.WARNING)
    logging.getLogger("uvicorn.access").setLevel(logging.ERROR)  # access logs off/noisy

    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())

    app_logger = logging.getLogger("app")
    app_logger.setLevel(logging.INFO)
    # Avoid duplicate handlers if reloaded
    if not any(isinstance(h, logging.StreamHandler) for h in app_logger.handlers):
        app_logger.addHandler(handler)
        app_logger.propagate = False
    return app_logger

log = setup_logging()

# --------------------------
# Time function (LLM-friendly JSON)
# --------------------------
def _time_payload_utc() -> Dict[str, object]:
    """
    Return multiple canonical time representations for LLMs/tools.
    All times are UTC.
    """
    now = datetime.now(timezone.utc)
    iso8601 = now.isoformat(timespec="milliseconds").replace("+00:00", "Z")

    # ISO week/date parts
    iso_year, iso_week, iso_weekday = now.isocalendar()
    yday = int(now.strftime("%j"))

    return {
        # Canonical machine formats
        "utc_iso8601": iso8601,                  # e.g., 2025-11-11T04:15:30.123Z
        "unix_seconds": int(now.timestamp()),    # e.g., 1762824930
        "unix_milliseconds": int(now.timestamp() * 1000),

        # Human-friendly hints (still machine-parseable)
        "date_utc": now.strftime("%Y-%m-%d"),    # 2025-11-11
        "time_utc": now.strftime("%H:%M:%S.%f")[:-3],  # 04:15:30.123
        "weekday_utc": now.strftime("%A"),       # Tuesday
        "rfc_1123_utc": now.strftime("%a, %d %b %Y %H:%M:%S GMT"),

        # Components
        "year": now.year,
        "month": now.month,
        "day": now.day,
        "hour": now.hour,
        "minute": now.minute,
        "second": now.second,
        "millisecond": int(now.microsecond / 1000),

        # ISO calendar context
        "iso_year": iso_year,
        "iso_week": iso_week,
        "iso_weekday": iso_weekday,  # 1=Mon ... 7=Sun
        "yearday": yday,

        # Provenance
        "timezone": "UTC",
        "source": "Datetime MCP Server",
        "note": "All values are UTC. Prefer 'utc_iso8601' for portability."
    }

def now_utc():
    """
    Return current time (UTC) as a rich JSON object.
    Designed to be easily consumed by LLMs/tools.
    """
    payload = _time_payload_utc()
    # Application-level audit log (one line per query)
    log.info("time_requested", extra={"route": "predict", "out_keys": list(payload.keys())})
    return payload

# --------------------------
# Gradio Interface
# --------------------------
demo = gr.Interface(
    fn=now_utc,
    inputs=None,
    outputs=gr.JSON(label="UTC time (LLM-friendly JSON)"),
    title="Datetime MCP Server",
    description="Returns the current UTC datetime in multiple canonical formats."
)

# --------------------------
# Middleware to log only meaningful hits
# --------------------------
def attach_selective_logging_middleware(app):
    """
    Add a lightweight middleware that logs only when an API/predict-like
    endpoint is hit; skips UI asset routes (/_app, /assets, /favicon, etc).
    """
    from fastapi import Request
    from starlette.responses import Response

    ASSET_PREFIXES = ("/_app", "/assets", "/static", "/favicon", "/logo", "/file=")
    INTERESTING_PREFIXES = ("/run", "/queue", "/predict", "/api", "/chat", "/submit")

    @app.middleware("http")
    async def selective_access_log(request: Request, call_next):
        path = request.url.path or "/"
        # Skip common asset paths
        if path.startswith(ASSET_PREFIXES):
            return await call_next(request)

        # Log only interesting interactive endpoints
        if path.startswith(INTERESTING_PREFIXES) or path == "/":
            ua = request.headers.get("user-agent", "")
            client_ip = getattr(request.client, "host", None)
            log.info(
                "incoming_request",
                extra={
                    "method": request.method,
                    "path": path,
                    "client_ip": client_ip,
                    "user_agent": ua[:300],
                },
            )
        response: Response = await call_next(request)
        return response

# Attach middleware to the FastAPI app Gradio runs under
# Gradio 4.x exposes the FastAPI app at demo.server.app
try:
    attach_selective_logging_middleware(demo.server.app)  # type: ignore[attr-defined]
except Exception:
    # Fallback for older/newer Gradio internals; safe no-op if unavailable.
    pass

if __name__ == "__main__":
    # quiet=True reduces banner/noise; mcp_server=True exposes the MCP SSE endpoint.
    demo.launch(
        mcp_server=True,
        quiet=True,            # suppresses gradio banner/extra logs
        show_error=True,       # still useful while developing
        server_name="0.0.0.0",
        server_port=None,
    )

    # https://nimo007-mcp-datetime-server.hf.space/gradio_api/mcp/