read_query
Execute SELECT queries on IoTDB databases to retrieve time-series data using SQL with TABLE dialect and ISO 8601 time formatting.
Instructions
Execute a SELECT query on the IoTDB. Please use table sql_dialect when generating SQL queries.
Args: query_sql: The SQL query to execute (using TABLE dialect, time using ISO 8601 format, e.g. 2017-11-01T00:08:00.000)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query_sql | Yes |
Implementation Reference
- src/iotdb_mcp_server/server.py:317-343 (handler)The handler function for the 'read_query' tool. It executes SELECT, DESCRIBE, or SHOW queries using IoTDB TableSessionPool in table SQL dialect and formats results using prepare_res.
async def read_query(query_sql: str) -> list[TextContent]: """Execute a SELECT query on the IoTDB. Please use table sql_dialect when generating SQL queries. Args: query_sql: The SQL query to execute (using TABLE dialect, time using ISO 8601 format, e.g. 2017-11-01T00:08:00.000) """ table_session = None try: table_session = session_pool.get_session() stmt = query_sql.strip().upper() # Regular SELECT queries if ( stmt.startswith("SELECT") or stmt.startswith("DESCRIBE") or stmt.startswith("SHOW") ): res = table_session.execute_query_statement(query_sql) return prepare_res(res, table_session) else: table_session.close() raise ValueError("Only SELECT queries are allowed for read_query") except Exception as e: if table_session: table_session.close() logger.error(f"Failed to execute query: {str(e)}") raise - Helper function used by read_query (and other table tools) to convert query results into a list of TextContent with CSV-formatted data.
def prepare_res( _res: SessionDataSet, _table_session: TableSession ) -> list[TextContent]: columns = _res.get_column_names() result = [] while _res.has_next(): row = _res.next().get_fields() result.append(",".join(map(str, row))) _table_session.close() return [ TextContent( type="text", text="\n".join([",".join(columns)] + result), ) ] - src/iotdb_mcp_server/server.py:317-317 (registration)The @mcp.tool() decorator registers the read_query function as an MCP tool.
async def read_query(query_sql: str) -> list[TextContent]: