src/server.py

"""MCP server that generates charts and returns static image data."""

from __future__ import annotations

import base64
import inspect
import os
from typing import Any, Literal

from mcp.server.fastmcp import FastMCP, Image

from src.charting import render_chart_png_base64

DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 7860
DEFAULT_MCP_PATH = "/mcp"


def _read_port() -> int:
    raw_port = os.getenv("PORT", str(DEFAULT_PORT)).strip()
    try:
        return int(raw_port)
    except ValueError:
        return DEFAULT_PORT


def _fastmcp_ctor_kwargs_from_env() -> dict[str, Any]:
    ctor_params = inspect.signature(FastMCP).parameters
    kwargs: dict[str, Any] = {}

    host = os.getenv("HOST", DEFAULT_HOST).strip() or DEFAULT_HOST
    port = _read_port()
    path = os.getenv("MCP_PATH", DEFAULT_MCP_PATH).strip() or DEFAULT_MCP_PATH

    if "host" in ctor_params:
        kwargs["host"] = host
    if "port" in ctor_params:
        kwargs["port"] = port
    if "streamable_http_path" in ctor_params:
        kwargs["streamable_http_path"] = path

    return kwargs


def _create_mcp_server() -> FastMCP:
    server = FastMCP("mcp-charts", **_fastmcp_ctor_kwargs_from_env())
    _register_tools(server)
    return server


def generate_chart(
    chart_type: Literal["line", "bar", "scatter", "pie", "grouped_bar", "stacked_bar"],
    x_values: list[float | str],
    y_values: list[float],
    y_series: list[list[float]] | None = None,
    series_labels: list[str] | None = None,
    pie_labels: list[str] | None = None,
    title: str = "Generated Chart",
    x_label: str = "X",
    y_label: str = "Y",
    color: str = "#1f77b4",
) -> Image:
    """Return a typed MCP image payload so clients can render directly."""

    image_base64 = render_chart_png_base64(
        chart_type=chart_type,
        x_values=x_values,
        y_values=y_values,
        y_series=y_series,
        series_labels=series_labels,
        pie_labels=pie_labels,
        title=title,
        x_label=x_label,
        y_label=y_label,
        color=color,
    )
    return Image(data=base64.b64decode(image_base64), format="png")


def _register_tools(server: FastMCP) -> None:
    server.tool(
        name="generate_chart",
        description=(
            "Generate a chart image (line, bar, scatter, pie, grouped_bar, stacked_bar) and return a typed PNG image."
        ),
    )(generate_chart)


mcp = _create_mcp_server()


def main() -> None:
    run_params = inspect.signature(mcp.run).parameters
    transport = os.getenv("MCP_TRANSPORT", "streamable-http").strip().lower()

    if transport == "stdio":
        kwargs = {"transport": "stdio"} if "transport" in run_params else {}
        try:
            mcp.run(**kwargs)
        except TypeError:
            mcp.run()
        return

    kwargs = {"transport": "streamable-http"} if "transport" in run_params else {}

    try:
        mcp.run(**kwargs)
    except TypeError:
        mcp.run()


if __name__ == "__main__":
    main()