IoTDB MCP Server

Instructions

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

• src/iotdb_mcp_server/server.py:461-475 (helper)

  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]: