app.py

"""Gradio MCP Server for Indicateurs Territoriaux de Transition Écologique.

This application exposes 4 MCP tools for querying French territorial
ecological indicators via the Cube.js API.

Tools:
    - list_indicators: List all indicators with optional filters
    - get_indicator_details: Get detailed info about a specific indicator
    - query_indicator_data: Query data values for a territory
    - search_indicators: Search indicators by keywords

Usage:
    Run locally:
        python app.py
    
    Deploy on HuggingFace Spaces:
        Push to a Space with Gradio SDK configured.
    
    Connect as MCP Server:
        URL: http://your-server:7860/gradio_api/mcp/
"""

import os
import logging
import gradio as gr
from dotenv import load_dotenv


# =============================================================================
# Logging Configuration - Filter out non-critical ASGI errors
# =============================================================================
class ASGIErrorFilter(logging.Filter):
    """Filter to suppress known non-critical ASGI/MCP errors.
    
    These errors occur when external clients send malformed requests
    or when health checks hit MCP endpoints. They don't affect functionality.
    """
    
    SUPPRESSED_MESSAGES = [
        "Exception in ASGI application",
        "'NoneType' object is not callable",
        "Exception Group Traceback",
    ]
    
    def filter(self, record: logging.LogRecord) -> bool:
        """Return False to suppress the log record."""
        message = record.getMessage()
        for suppressed in self.SUPPRESSED_MESSAGES:
            if suppressed in message:
                return False
        return True


# Apply filter to uvicorn error logger
uvicorn_error_logger = logging.getLogger("uvicorn.error")
uvicorn_error_logger.addFilter(ASGIErrorFilter())

# Also filter the root logger for starlette errors
logging.getLogger("starlette").addFilter(ASGIErrorFilter())

# Load environment variables
load_dotenv()

# Import tools
from src.tools import (
    list_indicators,
    get_indicator_details,
    query_indicator_data,
    search_indicators,
)
from src.models import GEOGRAPHIC_LEVELS

# Check if token is configured
if not os.getenv("INDICATEURS_TE_TOKEN"):
    print("WARNING: INDICATEURS_TE_TOKEN not set. API calls will fail.")
    print("Set the token in .env file or as environment variable.")


# Create individual interfaces for each tool
list_interface = gr.Interface(
    fn=list_indicators,
    inputs=[
        gr.Textbox(
            label="Thématique FNV",
            placeholder="Ex: mieux se déplacer, mieux se loger...",
            info="Filtre par thématique France Nation Verte (recherche partielle)",
        ),
        gr.Dropdown(
            choices=[""] + GEOGRAPHIC_LEVELS,
            label="Maille géographique",
            info="Filtre par niveau géographique disponible",
        ),
    ],
    outputs=gr.JSON(label="Indicateurs"),
    title="Lister les indicateurs",
    description="Liste tous les indicateurs disponibles avec filtres optionnels.",
    api_name="list_indicators",
)

details_interface = gr.Interface(
    fn=get_indicator_details,
    inputs=[
        gr.Textbox(
            label="ID de l'indicateur",
            placeholder="Ex: 611",
            info="Identifiant numérique de l'indicateur",
        ),
    ],
    outputs=gr.JSON(label="Détails"),
    title="Détails d'un indicateur",
    description="Retourne les métadonnées complètes et les sources d'un indicateur.",
    api_name="get_indicator_details",
)

query_interface = gr.Interface(
    fn=query_indicator_data,
    inputs=[
        gr.Textbox(
            label="ID de l'indicateur",
            placeholder="Ex: 611",
            info="Identifiant numérique de l'indicateur",
        ),
        gr.Dropdown(
            choices=GEOGRAPHIC_LEVELS,
            label="Niveau géographique",
            value="region",
            info="Maille territoriale à interroger",
        ),
        gr.Textbox(
            label="Code INSEE",
            placeholder="Ex: 93 (PACA), 13 (Bouches-du-Rhône)...",
            info="Code du territoire (optionnel)",
        ),
        gr.Textbox(
            label="Année",
            placeholder="Ex: 2020",
            info="Année des données (optionnel)",
        ),
    ],
    outputs=gr.JSON(label="Données"),
    title="Interroger les données",
    description="Récupère les valeurs d'un indicateur pour un territoire donné.",
    api_name="query_indicator_data",
)

search_interface = gr.Interface(
    fn=search_indicators,
    inputs=[
        gr.Textbox(
            label="Recherche",
            placeholder="Ex: consommation espace, surface bio, émissions CO2...",
            info="Mots-clés à rechercher dans le nom et la description",
        ),
    ],
    outputs=gr.JSON(label="Résultats"),
    title="Rechercher des indicateurs",
    description="Recherche des indicateurs par mots-clés.",
    api_name="search_indicators",
)

# Combine all interfaces into a tabbed interface
demo = gr.TabbedInterface(
    interface_list=[
        list_interface,
        search_interface,
        details_interface,
        query_interface,
    ],
    tab_names=[
        "Lister",
        "Rechercher",
        "Détails",
        "Données",
    ],
    title="MCP Server - Indicateurs Territoriaux de Transition Écologique",
)

# Add a description block
with demo:
    gr.Markdown(
        """
        ---
        ### Connexion MCP
        
        Pour utiliser ce serveur comme outil MCP dans Claude Desktop, Cursor ou autre client MCP :
        
        ```json
        {
          "mcpServers": {
            "indicateurs-te": {
              "url": "https://YOUR-SPACE.hf.space/gradio_api/mcp/"
            }
          }
        }
        ```
        
        ### Structure des données
        
        Les cubes de données suivent le format `{thematique}_{maille}` :
        - `conso_enaf_com` → Consommation ENAF, maille commune
        - `surface_bio_dpt` → Surface bio, maille département
        
        Les measures contiennent l'ID de l'indicateur : `{cube}.id_{indicator_id}`
        
        ### API Cube.js
        
        Ce serveur interroge l'API du Hub d'Indicateurs Territoriaux du Ministère de la Transition Écologique.
        
        - Documentation : [ecologie.data.gouv.fr/indicators](https://ecologie.data.gouv.fr/indicators)
        - API : `https://api.indicateurs.ecologie.gouv.fr`
        """
    )


if __name__ == "__main__":
    demo.launch(
        mcp_server=True,
        server_name="0.0.0.0",
        server_port=7860,
    )