import json
import math
import os
import re
import threading
import traceback
from typing import Any

import gradio as gr
import requests


EI_HOST = os.getenv("EI_HOST", "https://studio.edgeimpulse.com/v1").rstrip("/")
REQUEST_TIMEOUT_SECONDS = int(os.getenv("EI_REQUEST_TIMEOUT_SECONDS", "20"))
MAX_RESPONSE_CHARS = int(os.getenv("EI_MAX_RESPONSE_CHARS", "12000"))
AUTH_HELP = (
    "Use an Edge Impulse JWT from "
    "[the Studio login API docs](https://docs.edgeimpulse.com/apis/studio/login/get-jwt-token?playground=open), "
    "or open an Edge Impulse project and create a project API key if you want to limit this demo's access to that project."
)
SKILL_DOC_URL = "https://docs.edgeimpulse.com/tutorials/topics/ai-agents/create-edge-impulse-skill"
DOCS_INDEX_URL = "https://docs.edgeimpulse.com/llms.txt"
DEFAULT_CHAT_MODEL = os.getenv("CHAT_MODEL", "Qwen/Qwen2.5-0.5B-Instruct")
ENABLE_LOCAL_QWEN = os.getenv("ENABLE_LOCAL_QWEN", "true").lower() in {"1", "true", "yes"}
PRELOAD_QWEN = os.getenv("PRELOAD_QWEN", "true").lower() in {"1", "true", "yes"}
LOCAL_LLM_MAX_NEW_TOKENS = int(os.getenv("LOCAL_LLM_MAX_NEW_TOKENS", "220"))
PROVIDER_SPACE_QWEN = "Space-hosted Qwen 2.5 0.5B"
PROVIDER_EXTERNAL = "External OpenAI-compatible API"
PROVIDER_NONE = "No LLM"
LLM_PROVIDER_CHOICES = [PROVIDER_SPACE_QWEN, PROVIDER_EXTERNAL, PROVIDER_NONE]
SLASH_SKILLS = {
    "edge-impulse": {
        "summary": "General Edge Impulse Studio control-plane skill.",
        "examples": [
            "/edge-impulse list projects",
            "/edge-impulse project 69300 list active jobs",
            "/edge-impulse retrain project 69300",
        ],
    },
    "project-info": {
        "summary": "Inspect project metadata and deployment targets.",
        "examples": [
            "/project-info 69300",
            "/project-info project 69300 deployment targets",
        ],
    },
    "monitor-job": {
        "summary": "Check job status or recent logs.",
        "examples": [
            "/monitor-job project 69300 job 123 status",
            "/monitor-job project 69300 job 123 logs",
        ],
    },
    "train-ei-model": {
        "summary": "Start training-oriented workflows such as retrain, evaluate, feature generation, and Keras training.",
        "examples": [
            "/train-ei-model retrain project 69300",
            "/train-ei-model train Keras project 69300 learn block 34 with 40 cycles",
            "/train-ei-model generate features project 69300 dsp 12",
        ],
    },
    "deploy-impulse": {
        "summary": "Build deployment artifacts for a project.",
        "examples": [
            "/deploy-impulse build deployment project 69300 target arduino engine tflite-eon",
        ],
    },
}
_LOCAL_PIPELINE = None
_LOCAL_PIPELINE_MODEL = None
_LAST_QWEN_ERROR = ""

APP_CSS = """
#chatbot {
  border-radius: 14px;
}
#message_box textarea {
  font-size: 15px;
}
.settings-panel {
  border-radius: 12px;
}
.tool-console {
  max-height: 720px;
  overflow-y: auto;
}
"""


class EdgeImpulseError(Exception):
    pass


class LlmError(Exception):
    pass


def _auth_headers(auth_mode: str = "Use Space secret", credential: str = "") -> dict[str, str]:
    credential = credential.strip()

    if credential:
        if auth_mode == "JWT token":
            return {"x-jwt-token": credential}
        if auth_mode == "API key":
            return {"x-api-key": credential}
        raise EdgeImpulseError("Choose API key or JWT token when entering a credential.")

    jwt_token = os.getenv("EI_JWT_TOKEN", "").strip()
    api_key = os.getenv("EI_API_KEY", "").strip()

    if jwt_token:
        return {"x-jwt-token": jwt_token}
    if api_key:
        return {"x-api-key": api_key}

    raise EdgeImpulseError(
        "Missing Edge Impulse credentials. Enter an API key/JWT in the UI or configure EI_API_KEY/EI_JWT_TOKEN as a Space Secret."
    )


def _request(
    path: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    auth_mode: str = "Use Space secret",
    credential: str = "",
    json_body: dict[str, Any] | None = None,
) -> Any:
    url = f"{EI_HOST}{path}"
    try:
        response = requests.request(
            method,
            url,
            headers=_auth_headers(auth_mode, credential),
            params=params,
            json=json_body,
            timeout=REQUEST_TIMEOUT_SECONDS,
        )
    except requests.RequestException as exc:
        raise EdgeImpulseError(f"Request to Edge Impulse failed: {exc}") from exc

    if response.status_code >= 400:
        detail = response.text[:1000] if response.text else response.reason
        raise EdgeImpulseError(
            f"Edge Impulse returned HTTP {response.status_code} for {path}: {detail}"
        )

    content_type = response.headers.get("content-type", "")
    if "application/json" in content_type:
        return response.json()
    return response.text


def _format_response(data: Any) -> str:
    if isinstance(data, str):
        text = data
    else:
        text = json.dumps(data, indent=2, sort_keys=True)

    if len(text) > MAX_RESPONSE_CHARS:
        return (
            text[:MAX_RESPONSE_CHARS]
            + f"\n\n... truncated at {MAX_RESPONSE_CHARS} characters. Narrow the request for more detail."
        )
    return text


def _positive_int(value: int | float, name: str) -> int:
    try:
        if isinstance(value, float) and math.isnan(value):
            raise ValueError("NaN")
        parsed = int(value)
    except (TypeError, ValueError) as exc:
        raise EdgeImpulseError(f"{name} must be a positive integer.") from exc

    if parsed < 1:
        raise EdgeImpulseError(f"{name} must be a positive integer.")
    return parsed


def _is_blank(value: Any) -> bool:
    if value is None:
        return True
    if isinstance(value, float) and math.isnan(value):
        return True
    if isinstance(value, str) and value.strip().lower() in {"", "nan", "none", "null"}:
        return True
    return False


def _parse_json_object(value: str, name: str) -> dict[str, Any]:
    value = value.strip()
    if not value:
        return {}

    try:
        parsed = json.loads(value)
    except json.JSONDecodeError as exc:
        raise EdgeImpulseError(f"{name} must be valid JSON: {exc}") from exc

    if not isinstance(parsed, dict):
        raise EdgeImpulseError(f"{name} must be a JSON object.")
    return parsed


def _optional_impulse_params(impulse_id: int | float | None) -> dict[str, Any] | None:
    if _is_blank(impulse_id):
        return None
    return {"impulseId": _positive_int(impulse_id, "impulse_id")}


def _safe_call(
    path: str,
    method: str = "GET",
    params: dict[str, Any] | None = None,
    auth_mode: str = "Use Space secret",
    credential: str = "",
    json_body: dict[str, Any] | None = None,
) -> str:
    try:
        return _format_response(
            _request(
                path,
                method=method,
                params=params,
                auth_mode=auth_mode,
                credential=credential,
                json_body=json_body,
            )
        )
    except EdgeImpulseError as exc:
        return f"Error: {exc}"


def _safe_action(
    confirm: bool,
    path: str,
    method: str = "POST",
    params: dict[str, Any] | None = None,
    auth_mode: str = "Use Space secret",
    credential: str = "",
    json_body: dict[str, Any] | None = None,
) -> str:
    if not confirm:
        return "Error: This action starts or changes work in Edge Impulse. Set confirm to true to run it."
    return _safe_call(
        path,
        method=method,
        params=params,
        auth_mode=auth_mode,
        credential=credential,
        json_body=json_body,
    )


def list_projects(auth_mode: str = "Use Space secret", credential: str = "") -> str:
    """List active Edge Impulse projects visible to the configured credential.

    Args:
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing the projects the configured Edge Impulse credential can read.
    """
    return _safe_call("/api/projects", auth_mode=auth_mode, credential=credential)


def get_project_info(
    project_id: int, auth_mode: str = "Use Space secret", credential: str = ""
) -> str:
    """Get read-only project information for one Edge Impulse project.

    Args:
        project_id: Numeric Edge Impulse project ID.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON describing the selected Edge Impulse project.
    """
    project_id = _positive_int(project_id, "project_id")
    return _safe_call(f"/api/{project_id}", auth_mode=auth_mode, credential=credential)


def list_deployment_targets(
    project_id: int, auth_mode: str = "Use Space secret", credential: str = ""
) -> str:
    """List deployment targets available for an Edge Impulse project.

    Args:
        project_id: Numeric Edge Impulse project ID.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing deployment targets for the selected project.
    """
    project_id = _positive_int(project_id, "project_id")
    return _safe_call(
        f"/api/{project_id}/deployment/targets",
        auth_mode=auth_mode,
        credential=credential,
    )


def list_active_jobs(
    project_id: int,
    only_root_jobs: bool = True,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """List active jobs for an Edge Impulse project.

    Args:
        project_id: Numeric Edge Impulse project ID.
        only_root_jobs: Whether to return only root jobs.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing currently active jobs.
    """
    project_id = _positive_int(project_id, "project_id")
    return _safe_call(
        f"/api/{project_id}/jobs",
        params={"rootOnly": "true" if only_root_jobs else "false"},
        auth_mode=auth_mode,
        credential=credential,
    )


def get_job_status(
    project_id: int, job_id: int, auth_mode: str = "Use Space secret", credential: str = ""
) -> str:
    """Get status information for an Edge Impulse job.

    Args:
        project_id: Numeric Edge Impulse project ID.
        job_id: Numeric Edge Impulse job ID.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON describing the selected job status.
    """
    project_id = _positive_int(project_id, "project_id")
    job_id = _positive_int(job_id, "job_id")
    return _safe_call(
        f"/api/{project_id}/jobs/{job_id}/status",
        auth_mode=auth_mode,
        credential=credential,
    )


def get_job_logs(
    project_id: int,
    job_id: int,
    limit: int = 80,
    log_level: str = "info",
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Fetch recent stdout logs for an Edge Impulse job.

    Args:
        project_id: Numeric Edge Impulse project ID.
        job_id: Numeric Edge Impulse job ID.
        limit: Maximum number of log rows to request.
        log_level: Log level filter, for example info, warn, error, or debug.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing recent log output for the selected job.
    """
    project_id = _positive_int(project_id, "project_id")
    job_id = _positive_int(job_id, "job_id")
    bounded_limit = max(1, min(int(limit), 500))
    return _safe_call(
        f"/api/{project_id}/jobs/{job_id}/stdout",
        params={"limit": bounded_limit, "logLevel": log_level},
        auth_mode=auth_mode,
        credential=credential,
    )


def start_retrain_job(
    project_id: int,
    confirm: bool,
    impulse_id: int | None = None,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Start a retrain job for the current/default impulse.

    Args:
        project_id: Numeric Edge Impulse project ID.
        confirm: Must be true. This starts a real Studio job.
        impulse_id: Optional impulse ID. Leave blank for the default impulse.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing the started job ID and response metadata.
    """
    project_id = _positive_int(project_id, "project_id")
    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/retrain",
        params=_optional_impulse_params(impulse_id),
        auth_mode=auth_mode,
        credential=credential,
    )


def start_evaluate_job(
    project_id: int,
    confirm: bool,
    impulse_id: int | None = None,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Start an evaluation job for every variant of the current/default impulse.

    Args:
        project_id: Numeric Edge Impulse project ID.
        confirm: Must be true. This starts a real Studio job.
        impulse_id: Optional impulse ID. Leave blank for the default impulse.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing the started job ID and response metadata.
    """
    project_id = _positive_int(project_id, "project_id")
    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/evaluate",
        params=_optional_impulse_params(impulse_id),
        auth_mode=auth_mode,
        credential=credential,
    )


def generate_features_job(
    project_id: int,
    dsp_id: int,
    confirm: bool,
    calculate_feature_importance: bool = False,
    skip_feature_explorer: bool = False,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Generate features for a DSP block.

    Args:
        project_id: Numeric Edge Impulse project ID.
        dsp_id: DSP block ID to generate features for.
        confirm: Must be true. This starts a real Studio job.
        calculate_feature_importance: Whether to generate feature importance when available.
        skip_feature_explorer: Skip feature explorer, mainly useful for tests.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing the started job ID and response metadata.
    """
    project_id = _positive_int(project_id, "project_id")
    dsp_id = _positive_int(dsp_id, "dsp_id")
    body = {
        "dspId": dsp_id,
        "calculateFeatureImportance": bool(calculate_feature_importance),
        "skipFeatureExplorer": bool(skip_feature_explorer),
    }
    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/generate-features",
        json_body=body,
        auth_mode=auth_mode,
        credential=credential,
    )


def train_keras_job(
    project_id: int,
    learn_id: int,
    keras_parameters_json: str,
    confirm: bool,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Train a Keras learn block using the provided parameter JSON.

    Args:
        project_id: Numeric Edge Impulse project ID.
        learn_id: Learn block ID.
        keras_parameters_json: JSON object with Keras training parameters, e.g. {"trainingCycles": 30, "learningRate": 0.005}.
        confirm: Must be true. This starts a real Studio job.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing the started job ID and response metadata.
    """
    project_id = _positive_int(project_id, "project_id")
    learn_id = _positive_int(learn_id, "learn_id")
    body = _parse_json_object(keras_parameters_json, "keras_parameters_json")
    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/train/keras/{learn_id}",
        json_body=body,
        auth_mode=auth_mode,
        credential=credential,
    )


def build_ondevice_model_job(
    project_id: int,
    deployment_type: str,
    engine: str,
    confirm: bool,
    impulse_id: int | None = None,
    model_type: str = "",
    parameters_json: str = "{}",
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Build an on-device deployment artifact for a project.

    Args:
        project_id: Numeric Edge Impulse project ID.
        deployment_type: Deployment target format from list_deployment_targets, e.g. arduino.
        engine: Deployment engine, e.g. tflite, tflite-eon, or other target-supported engine.
        confirm: Must be true. This starts a real Studio job.
        impulse_id: Optional impulse ID. Leave blank for the default impulse.
        model_type: Optional model variant, e.g. int8 or float32.
        parameters_json: Optional JSON object of deployment-specific parameters.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing deployment version/job metadata.
    """
    project_id = _positive_int(project_id, "project_id")
    params = {"type": deployment_type.strip()}
    if not params["type"]:
        return "Error: deployment_type is required."
    impulse_params = _optional_impulse_params(impulse_id)
    if impulse_params:
        params.update(impulse_params)

    body: dict[str, Any] = {
        "engine": engine.strip(),
    }
    if not body["engine"]:
        return "Error: engine is required."
    if model_type.strip():
        body["modelType"] = model_type.strip()
    deployment_params = _parse_json_object(parameters_json, "parameters_json")
    if deployment_params:
        body["parameters"] = deployment_params

    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/build-ondevice-model",
        params=params,
        json_body=body,
        auth_mode=auth_mode,
        credential=credential,
    )


def cancel_job(
    project_id: int,
    job_id: int,
    confirm: bool,
    force_cancel: bool = False,
    auth_mode: str = "Use Space secret",
    credential: str = "",
) -> str:
    """Cancel a running Edge Impulse job.

    Args:
        project_id: Numeric Edge Impulse project ID.
        job_id: Numeric Edge Impulse job ID.
        confirm: Must be true. This changes job state in Studio.
        force_cancel: Mark the job as finished without waiting for the job cluster to cancel it.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.

    Returns:
        JSON containing cancellation status.
    """
    project_id = _positive_int(project_id, "project_id")
    job_id = _positive_int(job_id, "job_id")
    params = {"forceCancel": "true"} if force_cancel else None
    return _safe_action(
        confirm,
        f"/api/{project_id}/jobs/{job_id}/cancel",
        params=params,
        auth_mode=auth_mode,
        credential=credential,
    )


def _first_int(patterns: list[str], text: str) -> int | None:
    for pattern in patterns:
        match = re.search(pattern, text, flags=re.IGNORECASE)
        if match:
            return int(match.group(1))
    return None


def _extract_project_id(message: str, project_id: int | float | None) -> int | None:
    if not _is_blank(project_id):
        return _positive_int(project_id, "project_id")
    return _first_int(
        [
            r"\bproject(?:\s+id)?\s*[:#]?\s*(\d+)\b",
            r"\bpid\s*[:#]?\s*(\d+)\b",
        ],
        message,
    )


def _extract_training_params(message: str) -> dict[str, Any]:
    params: dict[str, Any] = {}
    cycles = _first_int([r"\b(?:cycles|epochs|training cycles)\s*[:=]?\s*(\d+)\b"], message)
    batch_size = _first_int([r"\b(?:batch|batch size)\s*[:=]?\s*(\d+)\b"], message)
    learning_rate_match = re.search(
        r"\b(?:lr|learning rate)\s*[:=]?\s*(0?\.\d+|\d+(?:\.\d+)?)\b",
        message,
        flags=re.IGNORECASE,
    )

    if cycles:
        params["trainingCycles"] = cycles
    if batch_size:
        params["batchSize"] = batch_size
    if learning_rate_match:
        params["learningRate"] = float(learning_rate_match.group(1))
    return params


def _shorten_for_prompt(text: str, limit: int = 6000) -> str:
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n\n... truncated at {limit} characters."


def _local_qwen_complete(prompt: str, model: str = DEFAULT_CHAT_MODEL) -> str:
    global _LAST_QWEN_ERROR, _LOCAL_PIPELINE, _LOCAL_PIPELINE_MODEL

    if not ENABLE_LOCAL_QWEN:
        raise LlmError("Space-hosted Qwen is disabled. Set ENABLE_LOCAL_QWEN=true to enable it.")

    try:
        from transformers import pipeline
    except Exception as exc:
        raise LlmError(
            "Space-hosted Qwen requires transformers and torch. Install the Space requirements or choose an external API provider."
        ) from exc

    selected_model = model.strip() or DEFAULT_CHAT_MODEL
    try:
        if _LOCAL_PIPELINE is None or _LOCAL_PIPELINE_MODEL != selected_model:
            try:
                import torch

                pipeline_kwargs = (
                    {"device_map": "auto", "torch_dtype": "auto"}
                    if torch.cuda.is_available()
                    else {"device": -1}
                )
            except Exception:
                pipeline_kwargs = {"device": -1}

            _LOCAL_PIPELINE = pipeline(
                "text-generation",
                model=selected_model,
                **pipeline_kwargs,
            )
            _LOCAL_PIPELINE_MODEL = selected_model

        result = _LOCAL_PIPELINE(
            prompt,
            max_new_tokens=LOCAL_LLM_MAX_NEW_TOKENS,
            do_sample=False,
            return_full_text=False,
        )
        _LAST_QWEN_ERROR = ""
        return str(result[0]["generated_text"]).strip()
    except Exception as exc:
        _LAST_QWEN_ERROR = f"{type(exc).__name__}: {exc}"
        print("Space-hosted Qwen failed:\n" + traceback.format_exc(), flush=True)
        raise LlmError(
            f"Space-hosted Qwen failed to load or generate: {_LAST_QWEN_ERROR}"
        ) from exc


def qwen_status(load_model: bool = False, model: str = DEFAULT_CHAT_MODEL) -> str:
    """Check whether the Space-hosted Qwen runtime is configured and optionally load it.

    Args:
        load_model: When true, attempts a tiny generation to force model download/load.
        model: Hugging Face model ID to check.

    Returns:
        JSON status for the Space-hosted model runtime.
    """
    status: dict[str, Any] = {
        "enabled": ENABLE_LOCAL_QWEN,
        "configuredModel": model or DEFAULT_CHAT_MODEL,
        "loaded": _LOCAL_PIPELINE is not None,
        "loadedModel": _LOCAL_PIPELINE_MODEL,
        "lastError": _LAST_QWEN_ERROR,
    }

    try:
        import torch
        import transformers

        status["torchVersion"] = torch.__version__
        status["transformersVersion"] = transformers.__version__
        status["cudaAvailable"] = bool(torch.cuda.is_available())
    except Exception as exc:
        status["dependencyError"] = f"{type(exc).__name__}: {exc}"

    if load_model:
        try:
            sample = _local_qwen_complete("Reply with: ready", model)
            status["loaded"] = True
            status["loadedModel"] = _LOCAL_PIPELINE_MODEL
            status["sample"] = sample[:200]
            status["lastError"] = _LAST_QWEN_ERROR
        except LlmError as exc:
            status["loadError"] = str(exc)
            status["lastError"] = _LAST_QWEN_ERROR

    return _format_response(status)


def _preload_space_qwen_worker() -> None:
    print(f"Preloading Space-hosted Qwen model: {DEFAULT_CHAT_MODEL}", flush=True)
    try:
        status = qwen_status(load_model=True, model=DEFAULT_CHAT_MODEL)
        print(f"Space-hosted Qwen preload status:\n{status}", flush=True)
    except Exception as exc:
        print(f"Space-hosted Qwen preload failed: {exc}", flush=True)


def preload_space_qwen() -> None:
    if not PRELOAD_QWEN:
        print("Space-hosted Qwen preload disabled by PRELOAD_QWEN=false.", flush=True)
        return

    thread = threading.Thread(
        target=_preload_space_qwen_worker,
        name="space-qwen-preload",
        daemon=True,
    )
    thread.start()
    print("Space-hosted Qwen preload started in the background.", flush=True)


def _openai_compatible_complete(
    prompt: str,
    api_key: str,
    model: str,
    base_url: str,
) -> str:
    if not api_key.strip():
        raise LlmError("Missing LLM API key.")
    if not model.strip():
        raise LlmError("Missing LLM model.")

    endpoint = base_url.strip().rstrip("/") or "https://api.openai.com/v1"
    response = requests.post(
        f"{endpoint}/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key.strip()}",
            "Content-Type": "application/json",
        },
        json={
            "model": model.strip(),
            "messages": [
                {
                    "role": "system",
                    "content": "You are a concise Edge Impulse assistant. Summarize tool results and suggest safe next steps.",
                },
                {"role": "user", "content": prompt},
            ],
            "temperature": 0.2,
            "max_tokens": 500,
        },
        timeout=REQUEST_TIMEOUT_SECONDS,
    )
    if response.status_code >= 400:
        raise LlmError(f"LLM API returned HTTP {response.status_code}: {response.text[:1000]}")
    data = response.json()
    return data["choices"][0]["message"]["content"].strip()


def _summarize_with_llm(
    provider: str,
    prompt: str,
    api_key: str,
    model: str,
    base_url: str,
) -> str:
    if provider == PROVIDER_SPACE_QWEN:
        return _local_qwen_complete(prompt, model)
    if provider == PROVIDER_EXTERNAL:
        return _openai_compatible_complete(prompt, api_key, model, base_url)
    return ""


def _looks_like_edge_impulse_request(message: str, project_id: int | float | None) -> bool:
    if not _is_blank(project_id):
        return True

    lower = message.lower()
    edge_terms = [
        "edge impulse",
        "project",
        "projects",
        "deployment",
        "deploy",
        "target",
        "job",
        "jobs",
        "logs",
        "retrain",
        "evaluate",
        "features",
        "dsp",
        "keras",
        "learn block",
        "impulse",
        "cancel",
        "train",
        "build",
    ]
    return any(term in lower for term in edge_terms)


def _general_space_agent_response(
    message: str,
    llm_provider: str,
    llm_api_key: str,
    llm_model: str,
    llm_base_url: str,
) -> str:
    fallback = (
        "I can help you use this Edge Impulse MCP demo. Ask me general questions, or add an Edge Impulse "
        "API key/JWT and a project ID to inspect projects, check jobs, view logs, retrain, evaluate, generate "
        "features, train Keras blocks, build deployments, or cancel jobs."
    )

    if llm_provider == PROVIDER_NONE:
        return fallback

    prompt = (
        "You are the Space-hosted Qwen assistant for an Edge Impulse MCP demo. "
        "Answer conversationally and concisely. Explain that Edge Impulse API calls require the user to enter "
        "an API key or JWT, but general questions can be answered without credentials.\n\n"
        f"User message:\n{message}"
    )
    try:
        return _summarize_with_llm(
            llm_provider,
            prompt,
            llm_api_key,
            llm_model,
            llm_base_url,
        )
    except LlmError as exc:
        return f"{fallback}\n\nSpace-hosted model note: {exc}"


def _format_tool_chat_response(tool_name: str, tool_result: str, summary: str = "") -> str:
    missing_credentials = "Missing Edge Impulse credentials" in tool_result
    if missing_credentials:
        credential_guidance = (
            "This request needs Edge Impulse access. Paste an API key or JWT in the connection panel first. "
            "Use a project API key if you want to limit this demo to one project, or a JWT for broader account access."
        )
        summary = f"{summary}\n\n{credential_guidance}" if summary else credential_guidance

    response = f"Tool: `{tool_name}`\n\n"
    if summary:
        response += f"{summary}\n\n"
    response += f"Raw result:\n```json\n{tool_result}\n```"
    return response


def _skill_catalog() -> str:
    rows = [
        "Slash skills available in this demo:",
        "",
        "- `/edge-impulse` - general Edge Impulse Studio control-plane skill",
        "- `/project-info` - inspect project metadata and deployment targets",
        "- `/monitor-job` - check job status and logs",
        "- `/train-ei-model` - retrain, evaluate, generate features, and train Keras blocks",
        "- `/deploy-impulse` - build deployment artifacts",
        "- `/docs` - open the Edge Impulse skill documentation",
        "",
        "Examples:",
    ]
    for skill in SLASH_SKILLS.values():
        rows.extend(f"- `{example}`" for example in skill["examples"][:1])
    rows.extend(
        [
            "",
            f"Full tutorial: {SKILL_DOC_URL}",
            f"Documentation index for agents: {DOCS_INDEX_URL}",
        ]
    )
    return "\n".join(rows)


def _skill_detail(skill_name: str) -> str:
    skill = SLASH_SKILLS.get(skill_name)
    if not skill:
        return _skill_catalog()

    examples = "\n".join(f"- `{example}`" for example in skill["examples"])
    return (
        f"`/{skill_name}`\n\n"
        f"{skill['summary']}\n\n"
        f"Examples:\n{examples}\n\n"
        f"Learn how to create and install the full Edge Impulse agent skill here:\n{SKILL_DOC_URL}"
    )


def _skill_prefixed_request(skill_name: str, request: str) -> str:
    if skill_name == "project-info":
        if "project" not in request.lower():
            return f"project {request} project info"
        return request
    if skill_name == "monitor-job":
        return request if ("job" in request.lower() or "log" in request.lower()) else f"{request} job status"
    if skill_name == "train-ei-model":
        return request
    if skill_name == "deploy-impulse":
        return request if "build" in request.lower() else f"build deployment {request}"
    return request


def _handle_slash_command(
    message: str,
    project_id: int | None,
    allow_actions: bool,
    auth_mode: str,
    credential: str,
) -> str | None:
    stripped = message.strip()
    if not stripped.startswith("/"):
        return None

    command, _, rest = stripped[1:].partition(" ")
    command = command.strip().lower()
    rest = rest.strip()

    if command in {"help", "skills"}:
        return _skill_catalog()

    if command == "docs":
        return (
            "Edge Impulse agent-skill documentation:\n"
            f"{SKILL_DOC_URL}\n\n"
            "The tutorial explains how `/edge-impulse` gives agents persistent knowledge of the Studio "
            "and Ingestion APIs, including project metadata, sample management, training jobs, logs, "
            "deployment exports, and skill install paths for common coding agents.\n\n"
            f"Docs index for agents: {DOCS_INDEX_URL}"
        )

    if command in SLASH_SKILLS:
        if not rest:
            return _skill_detail(command)

        routed_request = _skill_prefixed_request(command, rest)
        tool_name, tool_result = _route_edge_impulse_message(
            routed_request,
            _extract_project_id(routed_request, project_id),
            allow_actions,
            auth_mode,
            credential,
        )
        summary = (
            f"Slash skill `/{command}` selected. "
            f"This demo implements the runnable subset through the MCP tool layer. "
            f"For the full installable Agent Skill pattern, see {SKILL_DOC_URL}."
        )
        return _format_tool_chat_response(tool_name, tool_result, summary)

    return (
        f"Unknown slash skill `/{command}`.\n\n"
        + _skill_catalog()
    )


def _route_edge_impulse_message(
    message: str,
    project_id: int | None,
    allow_actions: bool,
    auth_mode: str,
    credential: str,
) -> tuple[str, str]:
    lower = message.lower()
    pid = _extract_project_id(message, project_id)

    if any(term in lower for term in ["list projects", "show projects", "my projects"]):
        return "list_projects", list_projects(auth_mode=auth_mode, credential=credential)

    if not pid:
        return (
            "needs_project_id",
            "I need a project ID for that. Add it in the Project ID field or say something like `project 69300`.",
        )

    if any(term in lower for term in ["deployment target", "deploy target", "targets"]):
        return "list_deployment_targets", list_deployment_targets(pid, auth_mode, credential)

    if any(term in lower for term in ["active jobs", "running jobs", "list jobs"]):
        return "list_active_jobs", list_active_jobs(pid, True, auth_mode, credential)

    job_id = _first_int([r"\bjob(?:\s+id)?\s*[:#]?\s*(\d+)\b"], message)
    if "log" in lower and job_id:
        return "get_job_logs", get_job_logs(pid, job_id, auth_mode=auth_mode, credential=credential)
    if ("job status" in lower or "status" in lower) and job_id:
        return "get_job_status", get_job_status(pid, job_id, auth_mode, credential)

    impulse_id = _first_int([r"\bimpulse(?:\s+id)?\s*[:#]?\s*(\d+)\b"], message)

    if "retrain" in lower:
        return "start_retrain_job", start_retrain_job(pid, allow_actions, impulse_id, auth_mode, credential)

    if "evaluate" in lower or "evaluation" in lower:
        return "start_evaluate_job", start_evaluate_job(pid, allow_actions, impulse_id, auth_mode, credential)

    if "generate features" in lower or "feature generation" in lower:
        dsp_id = _first_int([r"\bdsp(?:\s+block)?(?:\s+id)?\s*[:#]?\s*(\d+)\b"], message)
        if not dsp_id:
            return "needs_dsp_id", "I need a DSP block ID to generate features."
        return (
            "generate_features_job",
            generate_features_job(pid, dsp_id, allow_actions, auth_mode=auth_mode, credential=credential),
        )

    if "train" in lower and ("keras" in lower or "model" in lower):
        learn_id = _first_int([r"\blearn(?:\s+block)?(?:\s+id)?\s*[:#]?\s*(\d+)\b"], message)
        if not learn_id:
            return "needs_learn_id", "I need a learn block ID to train a Keras model."
        params = _extract_training_params(message) or {"trainingCycles": 30}
        return (
            "train_keras_job",
            train_keras_job(
                pid,
                learn_id,
                json.dumps(params),
                allow_actions,
                auth_mode,
                credential,
            ),
        )

    if "build" in lower and ("deployment" in lower or "on-device" in lower or "ondevice" in lower):
        deployment_match = re.search(
            r"\b(?:target|format|deployment)\s*[:=]?\s*([a-zA-Z0-9_.-]+)\b",
            message,
            flags=re.IGNORECASE,
        )
        engine_match = re.search(
            r"\bengine\s*[:=]?\s*([a-zA-Z0-9_.-]+)\b",
            message,
            flags=re.IGNORECASE,
        )
        deployment_type = deployment_match.group(1) if deployment_match else "arduino"
        engine = engine_match.group(1) if engine_match else "tflite-eon"
        return (
            "build_ondevice_model_job",
            build_ondevice_model_job(
                pid,
                deployment_type,
                engine,
                allow_actions,
                impulse_id,
                auth_mode=auth_mode,
                credential=credential,
            ),
        )

    if "cancel" in lower and job_id:
        force = "force" in lower
        return "cancel_job", cancel_job(pid, job_id, allow_actions, force, auth_mode, credential)

    if any(term in lower for term in ["project info", "project details", "summarize project", "show project"]):
        return "get_project_info", get_project_info(pid, auth_mode, credential)

    return (
        "help",
        (
            "I can list projects, inspect project info, list deployment targets, check jobs/logs, "
            "start retrain/evaluate/feature-generation/training/deployment jobs, and cancel jobs. "
            "For actions, enable Allow actions."
        ),
    )


def chat_with_edge_impulse(
    message: str,
    project_id: int | None = None,
    allow_actions: bool = False,
    auth_mode: str = "Use Space secret",
    credential: str = "",
    llm_provider: str = PROVIDER_SPACE_QWEN,
    llm_api_key: str = "",
    llm_model: str = DEFAULT_CHAT_MODEL,
    llm_base_url: str = "https://api.openai.com/v1",
) -> str:
    """Chat with the Edge Impulse tool layer and optionally summarize using an LLM.

    Args:
        message: Natural-language request.
        project_id: Optional default Edge Impulse project ID.
        allow_actions: Enables tools that start, build, train, evaluate, or cancel jobs.
        auth_mode: Choose API key, JWT token, or Space secret fallback.
        credential: Optional Edge Impulse API key or JWT token. Not stored.
        llm_provider: Space-hosted Qwen, external OpenAI-compatible API, or No LLM.
        llm_api_key: Optional API key for external OpenAI-compatible providers.
        llm_model: Model name for local or API provider.
        llm_base_url: OpenAI-compatible base URL.

    Returns:
        A chat response with the selected tool call and result summary.
    """
    if not message.strip():
        return "Ask me about an Edge Impulse project, job, deployment target, or training workflow. Type `/skills` to see slash-callable skills."

    try:
        resolved_project_id = _extract_project_id(message, project_id)
    except EdgeImpulseError as exc:
        return f"Project ID issue: {exc} Leave the default project ID blank for general chat, or enter a numeric Edge Impulse project ID."

    slash_response = _handle_slash_command(
        message,
        resolved_project_id,
        allow_actions,
        auth_mode,
        credential,
    )
    if slash_response is not None:
        return slash_response

    if not _looks_like_edge_impulse_request(message, project_id):
        return _general_space_agent_response(
            message,
            llm_provider,
            llm_api_key,
            llm_model,
            llm_base_url,
        )

    tool_name, tool_result = _route_edge_impulse_message(
        message,
        resolved_project_id,
        allow_actions,
        auth_mode,
        credential,
    )

    summary = ""
    if llm_provider != PROVIDER_NONE:
        prompt = (
            f"User request:\n{message}\n\n"
            f"Tool selected: {tool_name}\n\n"
            f"Tool result:\n{_shorten_for_prompt(tool_result)}\n\n"
            "Explain what happened in plain English. If the tool returned an error, say exactly what the user should fix."
        )
        try:
            summary = _summarize_with_llm(
                llm_provider,
                prompt,
                llm_api_key,
                llm_model,
                llm_base_url,
            )
        except LlmError as exc:
            summary = f"LLM summary unavailable: {exc}"

    return _format_tool_chat_response(tool_name, tool_result, summary)


def chat_ui_response(
    message: str,
    history: list[dict[str, str]] | None,
    project_id: int | None = None,
    allow_actions: bool = False,
    auth_mode: str = "Use Space secret",
    credential: str = "",
    llm_provider: str = PROVIDER_SPACE_QWEN,
    llm_api_key: str = "",
    llm_model: str = DEFAULT_CHAT_MODEL,
    llm_base_url: str = "https://api.openai.com/v1",
) -> str:
    return chat_with_edge_impulse(
        message=message,
        project_id=project_id,
        allow_actions=allow_actions,
        auth_mode=auth_mode,
        credential=credential,
        llm_provider=llm_provider,
        llm_api_key=llm_api_key,
        llm_model=llm_model,
        llm_base_url=llm_base_url,
    )


def chat_ui_messages(
    message: str,
    history: list[dict[str, str]] | None,
    project_id: int | None,
    allow_actions: bool,
    auth_mode: str,
    credential: str,
    llm_provider: str,
    llm_api_key: str,
    llm_model: str,
    llm_base_url: str,
) -> tuple[list[dict[str, str]], str]:
    history = list(history or [])
    if not message.strip():
        return history, ""

    try:
        response = chat_with_edge_impulse(
            message=message,
            project_id=project_id,
            allow_actions=allow_actions,
            auth_mode=auth_mode,
            credential=credential,
            llm_provider=llm_provider,
            llm_api_key=llm_api_key,
            llm_model=llm_model,
            llm_base_url=llm_base_url,
        )
    except Exception as exc:
        response = (
            f"Error handled by the demo: {exc}\n\n"
            "Try `/skills`, `/docs`, or enter a numeric project ID before using project-specific tools."
        )
    history.extend(
        [
            {"role": "user", "content": message},
            {"role": "assistant", "content": response},
        ]
    )
    return history, ""


def run_tool_console(
    tool: str,
    project_id: int | None,
    job_id: int | None,
    dsp_id: int | None,
    learn_id: int | None,
    impulse_id: int | None,
    confirm: bool,
    only_root_jobs: bool,
    force_cancel: bool,
    deployment_type: str,
    engine: str,
    model_type: str,
    json_input: str,
    limit: int,
    log_level: str,
    auth_mode: str,
    credential: str,
) -> str:
    try:
        if tool == "List projects":
            return list_projects(auth_mode, credential)

        if _is_blank(project_id):
            return "Error: Project ID is required for this tool."
        pid = _positive_int(project_id, "project_id")

        if tool == "Project info":
            return get_project_info(pid, auth_mode, credential)
        if tool == "Deployment targets":
            return list_deployment_targets(pid, auth_mode, credential)
        if tool == "Active jobs":
            return list_active_jobs(pid, only_root_jobs, auth_mode, credential)
        if tool == "Job status":
            if _is_blank(job_id):
                return "Error: Job ID is required."
            return get_job_status(pid, _positive_int(job_id, "job_id"), auth_mode, credential)
        if tool == "Job logs":
            if _is_blank(job_id):
                return "Error: Job ID is required."
            return get_job_logs(
                pid,
                _positive_int(job_id, "job_id"),
                limit,
                log_level,
                auth_mode,
                credential,
            )
        if tool == "Start retrain":
            return start_retrain_job(pid, confirm, impulse_id, auth_mode, credential)
        if tool == "Start evaluate":
            return start_evaluate_job(pid, confirm, impulse_id, auth_mode, credential)
        if tool == "Generate features":
            if _is_blank(dsp_id):
                return "Error: DSP block ID is required."
            return generate_features_job(
                pid,
                _positive_int(dsp_id, "dsp_id"),
                confirm,
                auth_mode=auth_mode,
                credential=credential,
            )
        if tool == "Train Keras":
            if _is_blank(learn_id):
                return "Error: Learn block ID is required."
            return train_keras_job(
                pid,
                _positive_int(learn_id, "learn_id"),
                json_input or "{}",
                confirm,
                auth_mode,
                credential,
            )
        if tool == "Build on-device model":
            return build_ondevice_model_job(
                pid,
                deployment_type,
                engine,
                confirm,
                impulse_id,
                model_type,
                json_input or "{}",
                auth_mode,
                credential,
            )
        if tool == "Cancel job":
            if _is_blank(job_id):
                return "Error: Job ID is required."
            return cancel_job(
                pid,
                _positive_int(job_id, "job_id"),
                confirm,
                force_cancel,
                auth_mode,
                credential,
            )
        return f"Error: Unknown tool {tool!r}."
    except EdgeImpulseError as exc:
        return f"Error: {exc}"


def build_single_page_demo() -> gr.Blocks:
    with gr.Blocks(
        title="Edge Impulse MCP Agent",
        theme=gr.themes.Soft(),
        css=APP_CSS,
        fill_height=True,
    ) as blocks:
        gr.Markdown(
            "# Edge Impulse MCP Agent\n"
            "Use one credential setup for chat and tools. The built-in conversation runs inside this Space "
            "with Space-hosted Qwen by default; external agents can also connect to the MCP endpoint.\n\n"
            "Type `/skills` in chat to see slash-callable skills, or `/docs` for the Edge Impulse Agent Skill tutorial."
        )

        with gr.Accordion("Connection and model settings", open=True, elem_classes=["settings-panel"]):
            gr.Markdown(AUTH_HELP)
            with gr.Row():
                auth_mode = gr.Dropdown(
                    label="Edge Impulse credential type",
                    choices=["API key", "JWT token", "Use Space secret"],
                    value="API key",
                    scale=1,
                )
                credential = gr.Textbox(
                    label="Edge Impulse credential",
                    type="password",
                    placeholder="Paste ei_... API key or JWT token once for this session.",
                    scale=2,
                )
                project_id = gr.Textbox(
                    label="Default project ID",
                    placeholder="Optional numeric project ID",
                    scale=1,
                )
                allow_actions = gr.Checkbox(label="Allow actions", value=False, scale=1)

            with gr.Row():
                llm_provider = gr.Dropdown(
                    label="Conversation model",
                    choices=LLM_PROVIDER_CHOICES,
                    value=PROVIDER_SPACE_QWEN,
                )
                llm_model = gr.Textbox(label="LLM model", value=DEFAULT_CHAT_MODEL)
            with gr.Row():
                llm_api_key = gr.Textbox(
                    label="External LLM API key",
                    type="password",
                    placeholder="Only needed when using an external OpenAI-compatible API.",
                )
                llm_base_url = gr.Textbox(
                    label="External LLM base URL",
                    value="https://api.openai.com/v1",
                )
            with gr.Row():
                check_qwen = gr.Button("Check Space-hosted Qwen", variant="secondary")
                load_qwen = gr.Checkbox(label="Force model load", value=True)
            qwen_output = gr.Code(label="Qwen status", language="json", max_lines=12)

        with gr.Row():
            with gr.Column(scale=3):
                chatbot = gr.Chatbot(
                    label="Conversation",
                    type="messages",
                    height=620,
                    show_copy_button=True,
                    elem_id="chatbot",
                )
                message = gr.Textbox(
                    label="Message",
                    placeholder=(
                        "Ask: /skills, /docs, /edge-impulse list projects, "
                        "/monitor-job project 69300 job 123 logs, or /train-ei-model retrain project 69300."
                    ),
                    elem_id="message_box",
                    lines=3,
                    max_lines=8,
                )
                with gr.Row():
                    send = gr.Button("Send", variant="primary")
                    clear = gr.Button("Clear")
                gr.Examples(
                    examples=[
                        "/skills",
                        "/docs",
                        "what can you do?",
                        "/edge-impulse list projects",
                        "/monitor-job project 69300 job 123 logs",
                        "/train-ei-model retrain project 69300",
                    ],
                    inputs=message,
                    label="Try",
                )

            with gr.Column(scale=2):
                with gr.Accordion("Tool console", open=False, elem_classes=["tool-console"]):
                    tool = gr.Dropdown(
                        label="Tool",
                        choices=[
                            "List projects",
                            "Project info",
                            "Deployment targets",
                            "Active jobs",
                            "Job status",
                            "Job logs",
                            "Start retrain",
                            "Start evaluate",
                            "Generate features",
                            "Train Keras",
                            "Build on-device model",
                            "Cancel job",
                        ],
                        value="Project info",
                    )
                    with gr.Row():
                        job_id = gr.Number(label="Job ID", precision=0)
                        dsp_id = gr.Number(label="DSP block ID", precision=0)
                        learn_id = gr.Number(label="Learn block ID", precision=0)
                    with gr.Row():
                        impulse_id = gr.Number(label="Impulse ID", precision=0)
                        limit = gr.Slider(label="Log limit", minimum=1, maximum=500, value=80, step=1)
                        log_level = gr.Dropdown(
                            label="Log level",
                            choices=["debug", "info", "warn", "error"],
                            value="info",
                        )
                    with gr.Row():
                        confirm = gr.Checkbox(label="Confirm action", value=False)
                        only_root_jobs = gr.Checkbox(label="Only root jobs", value=True)
                        force_cancel = gr.Checkbox(label="Force cancel", value=False)
                    with gr.Row():
                        deployment_type = gr.Textbox(label="Deployment target format", value="arduino")
                        engine = gr.Textbox(label="Engine", value="tflite-eon")
                        model_type = gr.Textbox(label="Model type", value="")
                    json_input = gr.Code(
                        label="JSON input",
                        language="json",
                        value='{"trainingCycles": 30, "learningRate": 0.005}',
                    )
                    run_tool = gr.Button("Run tool", variant="secondary")
                    tool_output = gr.Code(label="Tool result", language="json", max_lines=24)

        chat_inputs = [
            message,
            chatbot,
            project_id,
            allow_actions,
            auth_mode,
            credential,
            llm_provider,
            llm_api_key,
            llm_model,
            llm_base_url,
        ]
        send.click(
            chat_ui_messages,
            inputs=chat_inputs,
            outputs=[chatbot, message],
            api_name=False,
        )
        message.submit(
            chat_ui_messages,
            inputs=chat_inputs,
            outputs=[chatbot, message],
            api_name=False,
        )
        clear.click(lambda: [], outputs=chatbot, api_name=False)
        check_qwen.click(
            qwen_status,
            inputs=[load_qwen, llm_model],
            outputs=qwen_output,
            api_name="qwen_status",
            api_description="Check Space-hosted Qwen dependency and model-load status.",
        )

        run_tool.click(
            run_tool_console,
            inputs=[
                tool,
                project_id,
                job_id,
                dsp_id,
                learn_id,
                impulse_id,
                confirm,
                only_root_jobs,
                force_cancel,
                deployment_type,
                engine,
                model_type,
                json_input,
                limit,
                log_level,
                auth_mode,
                credential,
            ],
            outputs=tool_output,
            api_name="run_tool_console",
            api_description="Run one Edge Impulse tool using the shared page credential settings.",
        )

        with gr.Group(visible=False):
            api_message = gr.Textbox(label="Message", value="")
            api_project_id = gr.Textbox(label="Project ID", value="")
            api_allow_actions = gr.Checkbox(label="Allow actions", value=False)
            api_auth_mode = gr.Dropdown(
                label="Credential type",
                choices=["API key", "JWT token", "Use Space secret"],
                value="API key",
            )
            api_credential = gr.Textbox(label="Edge Impulse credential", type="password", value="")
            api_llm_provider = gr.Dropdown(
                label="Conversation model",
                choices=LLM_PROVIDER_CHOICES,
                value=PROVIDER_NONE,
            )
            api_llm_key = gr.Textbox(label="External LLM API key", type="password", value="")
            api_llm_model = gr.Textbox(label="LLM model", value=DEFAULT_CHAT_MODEL)
            api_llm_base = gr.Textbox(label="External LLM base URL", value="https://api.openai.com/v1")
            api_output = gr.Markdown(label="Assistant")
            api_run = gr.Button("API chat")
            api_run.click(
                chat_with_edge_impulse,
                inputs=[
                    api_message,
                    api_project_id,
                    api_allow_actions,
                    api_auth_mode,
                    api_credential,
                    api_llm_provider,
                    api_llm_key,
                    api_llm_model,
                    api_llm_base,
                ],
                outputs=api_output,
                api_name="chat_with_edge_impulse",
                api_description="Single-turn agent control-plane chat for Edge Impulse MCP use.",
            )

    return blocks


demo = build_single_page_demo()


if __name__ == "__main__":
    preload_space_qwen()
    demo.launch(mcp_server=True, ssr_mode=False)