TogoMCP

from fastmcp import FastMCP
import csv
from typing import Dict
import os
import httpx
import logging

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def toolcall_log(funname: str) -> None:
    """
    toolcall_log
    
    :param funname: The name of the tool being called.
    :type funname: str
    """
    logger.info(f"TogoMCP_tool: {funname}")
    return None


# The MIE files are used to define the shape expressions for SPARQL queries. 
CWD = os.getenv("TOGOMCP_DIR", ".")
MIE_DIR = CWD + "/mie"
MIE_PROMPT= CWD + "/resources/MIE_prompt.md"
RDF_PORTAL_GUIDE= CWD + "/resources/rdf_portal_guide.md"
SPARQL_EXAMPLES= CWD + "/sparql-examples"
RDF_CONFIG_TEMPLATE= CWD + "/rdf-config/template.yaml"
ENDPOINTS_CSV = CWD + "/resources/endpoints.csv"

def load_sparql_endpoints(path: str) -> Dict[str, str]:
    """Load SPARQL endpoints from a CSV file."""
    endpoints = {}
    with open(path, mode='r', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile)
        next(reader)  # Skip header
        for row in reader:
            db_name, endpoint_url = row
            key = db_name.lower().replace(' ', '_').replace('-', '')
            endpoints[key] = endpoint_url
    return endpoints

# The SPARQL endpoints for various RDF databases, loaded from a CSV file.
SPARQL_ENDPOINT = load_sparql_endpoints(ENDPOINTS_CSV)
DBNAME_DESCRIPTION = f"Database name: One of {", ".join(SPARQL_ENDPOINT.keys())}"

# Making this a @mcp.tool() becomes an error, so we keep it as a function.
async def execute_sparql(sparql_query: str, dbname: str) -> str:
    """ Execute a SPARQL query on RDF Portal. 
    Args:
        sparql_query (str): The SPARQL query to execute.
        dbname (str): The name of the database to query. To find the supported databases, use the `get_sparql_endpoints` tool.
    Returns:
        dict: The results of the SPARQL query in CSV.
    """

    if dbname not in SPARQL_ENDPOINT:
        raise ValueError(f"Unknown database: {dbname}")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            SPARQL_ENDPOINT[dbname], data={"query": sparql_query}, headers={"Accept": "text/csv"}
        )
    response.raise_for_status()
    return response.text

# The Primary MCP server
mcp = FastMCP("TogoMCP: RDF Portal MCP Server")