IoTDB MCP Server

Instructions

Implementation Reference

• src/iotdb_mcp_server/server.py:381-460 (handler)

  Implementation of the export_table_query tool handler. Decorated with @mcp.tool() for registration. Executes SQL query on IoTDB table dialect using session pool, exports results to CSV or Excel file using pandas, provides a preview of first 10 rows.
  @mcp.tool() async def export_table_query(query_sql: str, format: str = "csv", filename: str = None) -> list[TextContent]: """Execute a query and export the results to a CSV or Excel file. Args: query_sql: The SQL query to execute (using TABLE dialect, time using ISO 8601 format, e.g. 2017-11-01T00:08:00.000) format: Export format, either "csv" or "excel" (default: "csv") filename: Optional filename for the exported file. If not provided, a unique filename will be generated. SQL Syntax: SELECT ⟨select_list⟩ FROM ⟨tables⟩ [WHERE ⟨condition⟩] [GROUP BY ⟨groups⟩] [HAVING ⟨group_filter⟩] [FILL ⟨fill_methods⟩] [ORDER BY ⟨order_expression⟩] [OFFSET ⟨n⟩] [LIMIT ⟨n⟩]; Returns: Information about the exported file and a preview of the data (max 10 rows) """ table_session = None try: table_session = session_pool.get_session() stmt = query_sql.strip().upper() if stmt.startswith("SELECT") or stmt.startswith("SHOW") or stmt.startswith("DESCRIBE") or stmt.startswith("DESC"): # Execute the query res = table_session.execute_query_statement(query_sql) # Create a pandas DataFrame df = res.todf() # Close the session table_session.close() # Generate unique filename with timestamp timestamp = int(datetime.datetime.now().timestamp()) if filename is None: filename = f"dump_{uuid.uuid4().hex[:4]}_{timestamp}" filepath = "" if format.lower() == "csv": if(filename.lower().endswith(".csv")): filename = filename[:-4] filepath = f"{config.export_path}/{filename}.csv" df.to_csv(filepath, index=False) elif format.lower() == "excel": if(filename.lower().endswith(".xlsx")): filename = filename[:-5] filepath = f"{config.export_path}/{filename}.xlsx" df.to_excel(filepath, index=False) else: raise ValueError("Format must be either 'csv' or 'excel'") # Generate preview (first 10 rows) preview_rows = min(10, len(df)) preview_data = [] preview_data.append(",".join(df.columns)) # Header for i in range(preview_rows): preview_data.append(",".join(map(str, df.iloc[i]))) # Return information return [ TextContent( type="text", text=f"Query results exported to {filepath}\n\nPreview (first {preview_rows} rows):\n" + "\n".join(preview_data) ) ] else: raise ValueError("Only SELECT, SHOW or DESCRIBE queries are allowed for export") except Exception as e: if table_session: table_session.close() logger.error(f"Failed to export table query: {str(e)}") raise