app.py

import base64
import json
import mimetypes
import os
from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlparse

import gradio as gr
import requests


CONFIG_FILE = "mcpuploadclient.config.json"


@dataclass(frozen=True)
class ClientConfig:
    endpoint: str
    access_token: str
    request_timeout_seconds: int = 300


def load_config() -> ClientConfig:
    config_path = os.path.join(os.getcwd(), CONFIG_FILE)
    if not os.path.exists(config_path):
        raise FileNotFoundError(
            f"Config file not found: {config_path}. "
            f"Create {CONFIG_FILE} next to app.py (working directory)."
        )

    with open(config_path, "r", encoding="utf-8") as f:
        raw = json.load(f)

    endpoint = raw.get("Endpoint") or raw.get("endpoint")
    access_token = raw.get("AccessToken") or raw.get("access_token") or raw.get("Access_Token")
    timeout = raw.get("RequestTimeoutSeconds") or raw.get("request_timeout_seconds") or 300

    if not endpoint or not isinstance(endpoint, str):
        raise ValueError("Config must include 'Endpoint' (string).")
    if not access_token or not isinstance(access_token, str):
        raise ValueError("Config must include 'AccessToken' (string).")

    try:
        timeout_int = int(timeout)
    except Exception:
        timeout_int = 300

    return ClientConfig(endpoint=endpoint, access_token=access_token, request_timeout_seconds=timeout_int)


ALLOWED_EXTS = {".png", ".jpg", ".jpeg", ".webp"}


def _filename_from_url(image_url: str) -> str:
    parsed = urlparse(image_url)
    name = os.path.basename(parsed.path)
    if not name:
        raise ValueError(
            "Image URL must end with a filename (e.g. https://.../foo.png). "
            "URLs without a path filename are not supported."
        )
    return name


def _mime_from_filename(filename: str) -> str:
    ext = os.path.splitext(filename)[1].lower()
    if ext not in ALLOWED_EXTS:
        raise ValueError(f"Unsupported image extension '{ext}'. Allowed: {sorted(ALLOWED_EXTS)}")

    if ext == ".png":
        return "image/png"
    if ext in (".jpg", ".jpeg"):
        return "image/jpeg"
    if ext == ".webp":
        return "image/webp"

    guessed, _ = mimetypes.guess_type(filename)
    return guessed or "application/octet-stream"


def _parse_first_sse_data_line(sse_text: str) -> Dict[str, Any]:
    for line in sse_text.splitlines():
        line = line.strip()
        if not line:
            continue
        if line.startswith("data:"):
            payload = line[5:].strip()
            if not payload:
                continue
            return json.loads(payload)
    raise ValueError("No SSE data lines found in server response.")


def mcp_upload(image_url: str, json_text: str) -> str:
    image_url = (image_url or "").strip()
    json_text = (json_text or "").strip()

    if not image_url:
        return "ERROR: image_url is required."
    if not json_text:
        return "ERROR: json_text is required."

    try:
        cfg = load_config()
    except Exception as ex:
        return f"ERROR: failed to load config: {ex}"

    try:
        image_file_name = _filename_from_url(image_url)
        image_mime_type = _mime_from_filename(image_file_name)

        base = os.path.splitext(image_file_name)[0]
        json_file_name = base + ".json"

        # Download image bytes
        r = requests.get(image_url, timeout=cfg.request_timeout_seconds)
        r.raise_for_status()
        image_bytes = r.content
        image_b64 = base64.b64encode(image_bytes).decode("ascii")

        # Build JSON-RPC tool call
        body = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "upload_image_and_json",
                "arguments": {
                    "imageFileName": image_file_name,
                    "imageMimeType": image_mime_type,
                    "imageBase64": image_b64,
                    "jsonFileName": json_file_name,
                    "jsonText": json_text,
                },
            },
        }

        headers = {
            "ProtocolVersion": "2025-06-18",
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Authorization": f"Bearer {cfg.access_token}",
        }

        resp = requests.post(
            cfg.endpoint,
            headers=headers,
            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
            timeout=cfg.request_timeout_seconds,
        )
        resp.raise_for_status()

        rpc = _parse_first_sse_data_line(resp.text)

        # Extract tool-returned text
        tool_text = (
            rpc.get("result", {})
            .get("content", [{}])[0]
            .get("text")
        )
        if not tool_text:
            return f"ERROR: server response did not include result.content[0].text. Raw first data: {json.dumps(rpc, ensure_ascii=False)}"

        return tool_text

    except requests.RequestException as ex:
        return f"ERROR: HTTP request failed: {ex}"
    except Exception as ex:
        return f"ERROR: {ex}"


with gr.Blocks(title="McpUploadServer Client") as demo:
    gr.Markdown("# McpUploadServer Client (Gradio)\n\nEnter an **image URL** and a **JSON text**. Click Upload to send them to McpUploadServer via MCP.")

    image_url = gr.Textbox(label="Image URL", placeholder="https://example.com/image.png")
    json_text = gr.Textbox(label="JSON text", lines=12, placeholder='{"Category":"portrait", ... }')

    upload_btn = gr.Button("Upload")
    result = gr.Textbox(label="Result", lines=12)

    upload_btn.click(fn=mcp_upload, inputs=[image_url, json_text], outputs=[result])


if __name__ == "__main__":
    launch_kwargs = {
        "server_port": int(os.environ.get("PORT", "7860")),
    }

    # NOTE: Setting server_name="0.0.0.0" can cause Gradio's internal startup check
    # (GET /gradio_api/startup-events) to hang locally because it tries to connect
    # to http://0.0.0.0:PORT, which isn't a routable address.
    # HuggingFace Spaces typically sets GRADIO_SERVER_NAME=0.0.0.0 for you.
    server_name = os.environ.get("GRADIO_SERVER_NAME")
    if not server_name:
        # Common HuggingFace Spaces environment variables. If we detect Spaces and the
        # host isn't explicitly set, bind to 0.0.0.0 so the reverse proxy can reach us.
        is_spaces = any(
            os.environ.get(k)
            for k in (
                "SPACE_ID",
                "SPACE_REPO_NAME",
                "SPACE_AUTHOR_NAME",
                "HF_SPACE_ID",
            )
        )
        if is_spaces:
            server_name = "0.0.0.0"

    if server_name:
        launch_kwargs["server_name"] = server_name

    demo.launch(**launch_kwargs,mcp_server=True)