export_table_query

Instructions

Implementation Reference

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

  The handler function for the export_table_query tool, registered via @mcp.tool(). It handles queries in TABLE dialect, exports results to CSV/Excel using pandas, generates preview, and returns file information.

  @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

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

  The @mcp.tool() decorator registers the export_table_query function as an MCP tool.

  @mcp.tool()

• src/iotdb_mcp_server/server.py:382-382 (schema)

  Type hints and docstring define the input schema (query_sql: str, format: str='csv', filename: str=None) and output (list[TextContent]).

  async def export_table_query(query_sql: str, format: str = "csv", filename: str = None) -> list[TextContent]: