gradio_mcp_server.py

from mcp.server.fastmcp import FastMCP
import json
import sys
import io
import requests

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')

mcp = FastMCP("lexicon_api_server")

BASE_URL = "https://lexicon.osfarm.org"

def make_api_request(endpoint: str, params: dict = None) -> str:
    """Helper function to make API requests with consistent error handling"""
    url = f"{BASE_URL}{endpoint}"
    
    try:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        
        # Check if response is GeoJSON or regular JSON
        content_type = response.headers.get('content-type', '')
        if 'application/geo+json' in content_type or endpoint.endswith('.geojson'):
            data_type = "geojson"
        else:
            data_type = "json"
        
        data = response.json()
        
        return json.dumps({
            "type": "success",
            "data_type": data_type,
            "endpoint": endpoint,
            "data": data,
            "message": f"Successfully retrieved data from {endpoint}"
        }, indent=2)
        
    except requests.exceptions.ConnectionError:
        return json.dumps({
            "type": "error",
            "endpoint": endpoint,
            "message": f"Could not connect to API. Please ensure the service is running."
        })
    except requests.exceptions.Timeout:
        return json.dumps({
            "type": "error",
            "endpoint": endpoint,
            "message": f"Request timed out for {endpoint}"
        })
    except requests.exceptions.HTTPError as e:
        status_code = e.response.status_code if e.response else "unknown"
        return json.dumps({
            "type": "error",
            "endpoint": endpoint,
            "status_code": status_code,
            "message": f"HTTP error {status_code} for {endpoint}. Resource may not exist or API may be unavailable."
        })
    except json.JSONDecodeError:
        return json.dumps({
            "type": "error",
            "endpoint": endpoint,
            "message": f"Invalid JSON response from {url}"
        })
    except Exception as e:
        return json.dumps({
            "type": "error",
            "endpoint": endpoint,
            "message": f"Unexpected error: {str(e)}"
        })

@mcp.tool()
async def get_parcel_identifier_json(latitude: float, longitude: float) -> str:
    """
    Retrieve parcel identifier information in JSON format for a given geographic location. -- ST PORCHAIRE

    Args:
        latitude (float): Latitude of the point of interest.
        longitude (float): Longitude of the point of interest.

    Returns:
        str: JSON string containing parcel identifier data for the specified coordinates.

    This tool allows you to obtain parcel identification data by providing precise geographic coordinates.
    Useful for reverse-geocoding a location to its cadastral reference.
    """
    params = {"latitude": latitude, "longitude": longitude}
    return make_api_request("/tools/parcel-identifier.json", params)[:20000]

@mcp.tool()
async def get_cadastral_parcels(
    page: int = 1,
    code: str = None,
    prefix: str = None,
    section: str = None,
    number: str = None
) -> str:
    """
    Retrieve a paginated list of cadastral parcels, with optional filters.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        code (str, optional): Commune code to filter parcels.
        prefix (str, optional): Parcel prefix for more precise filtering.
        section (str, optional): Parcel section identifier.
        number (str, optional): Parcel number.

    Returns:
        str: JSON string containing a list of cadastral parcels matching the filters.

    This tool enables searching for cadastral parcels using various administrative and parcel-specific filters.
    Useful for exploring land registry data at different levels of granularity.
    Using a single postal_code, gives every cadastral parcels codes in that city, for example.
    """
    params = {"page": page}
    if code: params["code"] = code
    if prefix: params["prefix"] = prefix
    if section: params["section"] = section
    if number: params["number"] = number
    return make_api_request("/geographical-references/cadastral-parcels.json", params)

@mcp.tool()
async def get_cadastral_parcel(parcel_id: str) -> str:
    """
    Retrieve detailed information about a specific cadastral parcel.

    Args:
        parcel_id (str): Unique identifier of the cadastral parcel.

    Returns:
        str: JSON string with detailed information about the parcel.

    Use this tool to get all available data for a single cadastral parcel, including administrative and spatial attributes.
    """
    return make_api_request(f"/geographical-references/cadastral-parcels/{parcel_id}.json")


@mcp.tool()
async def get_cadastral_parcel_prices(postal_code: str = None, city: str = None, department: str = None) -> str:
    """
    Retrieve cadastral parcel price information, filtered by postal code, city, or department,

    Args:
        postal_code (str, optional): Postal code to filter results.
        city (str, optional): City name for filtering.
        department (str, optional): Department code or name.

    Returns:
        str: JSON string with price information for cadastral parcels.

    This tool provides access to price data for cadastral parcels, supporting multiple administrative filters.
    It is possible, with a postal code and a city name, to get the prices of every cadastral parcel sold in that city and when, 
    for example. So it is possible to know which one is the most expensive or the cheapest.
    """
    params = {}
    if postal_code: params["postal_code"] = postal_code
    if city: params["city"] = city
    if department: params["department"] = department
    ## FOR DEMO PURPOSES:
    params["page"] = 2
    return make_api_request("/geographical-references/cadastral-parcel-prices.json", params)
    
# geographical-references/cadastral-parcel-prices?postal_code=&city=Saint-Porchaire&department=&page=2

@mcp.tool()
async def get_cap_parcels(page: int = 1, city: str = None) -> str:
    """
    Retrieve a paginated list of CAP (Common Agricultural Policy) parcels, 
    allowing to access the crops available in the filtered city.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        city (str, optional): City name to filter CAP parcels.

    Returns:
        str: JSON string containing CAP parcels matching the filters.

    This tool allows you to explore CAP parcels, which are relevant for agricultural policy and subsidy management.
    """
    return make_api_request(f"/geographical-references/cap-parcels.json?city={city}")

@mcp.tool()
async def get_municipalities(page: int = 1, country: str = None, city: str = None) -> str:
    """
    Retrieve a paginated list of municipalities, with optional filters for country and city.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        country (str, optional): Country name to filter municipalities.
        city (str, optional): City name for more precise filtering.

    Returns:
        str: JSON string containing municipalities matching the filters.

    This tool is useful for exploring administrative boundaries and locating municipalities by name or country.
    """
    params = {"page": page}
    if country: params["country"] = country
    if city: params["city"] = city
    return make_api_request("/geographical-references/municipalities.json", params)

@mcp.tool()
async def get_municipality(municipality_id: str) -> str:
    """
    Retrieve detailed information about a specific municipality.

    Args:
        municipality_id (str): Unique identifier of the municipality.

    Returns:
        str: JSON string with detailed information about the municipality.

    Use this tool to access all available data for a single municipality, including administrative and spatial attributes.
    """
    return make_api_request(f"/geographical-references/municipalities/{municipality_id}.json")

@mcp.tool()
async def get_productions(page: int = 1, family: str = None, usage: str = None) -> str:
    """
    Retrieve a paginated list of production data, with optional filters for family and usage.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        family (str, optional): Production family (e.g., crop type).
        usage (str, optional): Usage type (e.g., food, feed).

    Returns:
        str: JSON string containing production data matching the filters.

    This tool is useful for analyzing agricultural production by type and intended use.
    """
    params = {"page": page}
    if family: params["family"] = family
    if usage: params["usage"] = usage
    return make_api_request("/production/productions.json", params)

@mcp.tool()
async def get_cropsets(page: int = 1) -> str:
    """
    Retrieve a paginated list of phytosanitary cropsets.

    Args:
        page (int, optional): Page number for pagination (default: 1).

    Returns:
        str: JSON string containing cropset data.

    This tool provides access to phytosanitary cropsets, which are important for plant protection and regulatory compliance.
    """
    params = {"page": page}
    return make_api_request("/phytosanitary/cropsets.json", params)

@mcp.tool()
async def get_phytosanitary_products(page: int = 1) -> str:
    """
    Retrieve a paginated list of phytosanitary products, and returns:
     Name
     firm
     Type
     Active compounds
     Usage state

    Args:
        page (int, optional): Page number for pagination (default: 1).
        type (str, optional): Product type (e.g., herbicide, fungicide).
        state (str, optional): Product state (e.g., approved, withdrawn).

    Returns:
        str: JSON string containing phytosanitary products matching the filters.

    This tool is useful for exploring available plant protection products and their regulatory status.
    """
    return make_api_request("/phytosanitary/products.json")

@mcp.tool()
async def get_phytosanitary_symbols(page: int = 1) -> str:
    """
    Retrieve a paginated list of phytosanitary symbols.

    Args:
        page (int, optional): Page number for pagination (default: 1).

    Returns:
        str: JSON string containing phytosanitary symbols.

    This tool provides access to symbols used in plant protection and regulatory documentation.
    """
    params = {"page": page}
    return make_api_request("/phytosanitary/symbols.json", params)

@mcp.tool()
async def get_seed_varieties(page: int = 1, species: str = None) -> str:
    """
    Retrieve a paginated list of seed varieties, optionally filtered by species.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        species (str, optional): Species name to filter seed varieties, ALWAYS IN CAPITAL LETTERS, e.g for "avoine", use "AVOINE",
        to find the right species, for example "AVOINE". Then varieties will be filtered.

    Returns:
        str: JSON string containing seed varieties matching the filters.

    This tool is useful for exploring available seed varieties for different crops.
    """
    params = {"page": page}
    if species: params["species"] = species
    return make_api_request("/seeds/varieties.json", params)


@mcp.tool()
async def get_vine_varieties(page: int = 1, category: str = None, color: str = None) -> str:
    """
    Retrieve a paginated list of vine varieties, with optional filters for category and color.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        category (str, optional): Vine category (e.g., table, wine).
        color (str, optional): Grape color (e.g., red, white).

    Returns:
        str: JSON string containing vine varieties matching the filters.

    This tool is useful for exploring grapevine diversity and selecting varieties for viticulture.
    """
    params = {"page": page}
    if category: params["category"] = category
    if color: params["color"] = color
    return make_api_request("/viticulture/vine-varieties.json", params)

@mcp.tool()
async def get_weather_stations(page: int = 1, country: str = None, name: str = None) -> str:
    """
    Retrieve a paginated list of weather stations, with optional filters for country and station name.

    Args:
        page (int, optional): Page number for pagination (default: 1).
        country (str, optional): Country name to filter stations.
        name (str, optional): Station name for more precise filtering.

    Returns:
        str: JSON string containing weather stations matching the filters.

    This tool is useful for discovering available weather stations and narrowing down by location or name.
    """
    params = {"page": page}
    if country: params["country"] = country
    if name: params["name"] = name
    return make_api_request("/weather/stations.json", params)

@mcp.tool()
async def get_weather_station(station_code: str) -> str:
    """
    Retrieve detailed information about a specific weather station.

    Args:
        station_code (str): Unique code identifying the weather station.

    Returns:
        str: JSON string with detailed information about the weather station.

    Use this tool to access metadata and attributes for a single weather station.
    """
    return make_api_request(f"/weather/stations/{station_code}.json")

@mcp.tool()
async def get_weather_data(station_code: str, start: str = None, end: str = None) -> str:
    """
    Retrieve hourly weather reports for a specific station, optionally filtered by date range.

    Args:
        station_code (str): Unique code identifying the weather station.
        start (str, optional): Start date/time in ISO format (e.g., '2024-01-01T00:00:00Z').
        end (str, optional): End date/time in ISO format (e.g., '2024-01-31T23:59:59Z').

    Returns:
        str: JSON string containing hourly weather reports for the specified station and date range.

    This tool is useful for analyzing weather data over time for a given location.
    """
    params = {}
    if start: params["start"] = start
    if end: params["end"] = end
    return make_api_request(f"/weather/stations/{station_code}/hourly-reports.json", params)


@mcp.tool()
async def get_cadastral_parcel_geolocation(parcel_id: str) -> str:
    """
    Retrieve geolocation data for a specific cadastral parcel in GeoJSON format.

    Args:
        parcel_id (str): Unique identifier of the cadastral parcel.

    Returns:
        str: GeoJSON string with spatial data for the parcel.

    Use this tool to obtain the geometry of a cadastral parcel for mapping or spatial analysis.
    """
    return make_api_request(f"/geographical-references/cadastral-parcels/{parcel_id}/geolocation.geojson")

@mcp.tool()
async def get_cap_parcel_geolocation(cap_id: str) -> str:
    """
    Retrieve geolocation data for a specific CAP parcel in GeoJSON format.

    Args:
        cap_id (str): Unique identifier of the CAP parcel.

    Returns:
        str: GeoJSON string with spatial data for the CAP parcel.

    Use this tool to obtain the geometry of a CAP parcel for mapping or spatial analysis.
    """
    return make_api_request(f"/geographical-references/cap-parcels/{cap_id}/geolocation.geojson")

@mcp.tool()
async def get_municipality_cadastre(municipality_id: str) -> str:
    """
    Retrieve cadastre data for a municipality in GeoJSON format.

    Args:
        municipality_id (str): Unique identifier of the municipality.

    Returns:
        str: GeoJSON string with cadastre data for the municipality.

    Use this tool to obtain the spatial extent of a municipality's cadastre for mapping or GIS analysis.
    """
    return make_api_request(f"/geographical-references/municipalities/{municipality_id}/cadastre.geojson")

if __name__ == "__main__":
    mcp.run(transport='stdio')