ANSES Ciqual MCP Server

Instructions

Implementation Reference

• src/server.py:26-106 (handler)

  The primary implementation of the 'query' MCP tool. This async function, decorated with @mcp.tool(), handles SQL query execution on the read-only ANSES Ciqual database. Includes input validation (SELECT only), connection management, error handling, and returns results as list of dicts. The docstring provides detailed schema information and usage examples.
  @mcp.tool() async def query(sql: str) -> list[dict]: """Execute SQL query on ANSES Ciqual French food composition database. IMPORTANT: Get ALL nutrients in ONE query! Don't make multiple queries for the same food. EXAMPLE - Get complete nutrition for a food: SELECT f.alim_nom_eng, n.const_nom_eng, c.teneur, n.unit FROM foods f JOIN composition c ON f.alim_code = c.alim_code JOIN nutrients n ON c.const_code = n.const_code WHERE f.alim_code = 23000; -- Returns ALL 60+ nutrients in one query! SCHEMA: • foods: 3,185+ foods with French/English names - alim_code (PK), alim_nom_fr, alim_nom_eng, alim_grp_code • nutrients: ~60+ nutrients with units - const_code (PK), const_nom_fr, const_nom_eng, unit • composition: nutritional values per 100g - alim_code, const_code, teneur (value), code_confiance (A/B/C/D) • foods_fts: full-text search for fuzzy matching - Use: WHERE foods_fts MATCH 'search term' COMMON QUERIES: 1. Search foods: SELECT * FROM foods_fts WHERE foods_fts MATCH 'cake'; 2. Get ALL nutrients: JOIN all 3 tables, no WHERE clause on nutrients 3. Get specific nutrients: Use IN clause with multiple codes at once KEY NUTRIENT CODES: Energy: 327 (kJ), 328 (kcal) Macros: 25000 (protein g), 31000 (carbs g), 40000 (fat g), 34100 (fiber g), 32000 (sugars g) Minerals: 10110 (sodium mg), 10200 (calcium mg), 10260 (iron mg), 10190 (potassium mg) Vitamins: 55400 (vit C mg), 56400 (vit D µg), 51330 (vit B12 µg), 56310 (vit E mg) The database is read-only. Use SELECT queries only. """ # Ensure database exists if not DB_PATH.exists(): logger.warning("Database not found at %s", DB_PATH) return [{"error": "Database not initialized. Please run the server first to download data."}] # Validate SQL query (basic safety check) sql_lower = sql.strip().lower() if not sql_lower.startswith(('select', 'with')): return [{"error": "Only SELECT queries are allowed for safety."}] # Connect with read-only mode try: logger.debug("Executing query: %s", sql[:100] + '...' if len(sql) > 100 else sql) conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) conn.row_factory = sqlite3.Row # Execute query with timeout conn.execute("PRAGMA query_only = ON") conn.execute("PRAGMA temp_store = MEMORY") cursor = conn.execute(sql) results = [dict(row) for row in cursor.fetchall()] logger.debug("Query returned %d rows", len(results)) return results except sqlite3.OperationalError as e: logger.error("SQL operational error: %s", e) if "no such table" in str(e): return [{"error": f"Table not found. Available tables: foods, nutrients, composition, foods_fts, food_groups"}] elif "read-only" in str(e) or "readonly" in str(e): return [{"error": "Database is read-only. Only SELECT queries are allowed."}] else: return [{"error": f"SQL error: {str(e)}"}] except sqlite3.Error as e: logger.error("Database error: %s", e) return [{"error": f"Database error: {str(e)}"}] except Exception as e: logger.error("Unexpected error: %s", e) return [{"error": f"Unexpected error: {str(e)}"}] finally: if 'conn' in locals(): conn.close()
• src/server.py:26-27 (registration)

  The @mcp.tool() decorator registers the 'query' function as an MCP tool with FastMCP instance.
  @mcp.tool() async def query(sql: str) -> list[dict]:
• src/server.py:27-27 (schema)

  Type annotations define the input schema (sql: str) and output schema (list[dict]). The extensive docstring details database schema, tables, and query examples.
  async def query(sql: str) -> list[dict]: