IoTDB MCP Server

Instructions

Implementation Reference

• src/iotdb_mcp_server/server.py:364-379 (handler)

  Handler function for the 'describe_table' tool. Decorated with @mcp.tool() for registration. Executes 'DESC {table_name} details' query on IoTDB table session and returns formatted schema via prepare_res.

  @mcp.tool()
  async def describe_table(table_name: str) -> list[TextContent]:
      """Get the schema information for a specific table
      Args:
          table_name: name of the table to describe
      """
      table_session = None
      try:
          table_session = session_pool.get_session()
          res = table_session.execute_query_statement("DESC " + table_name + " details")
          return prepare_res(res, table_session)
      except Exception as e:
          if table_session:
              table_session.close()
          logger.error(f"Failed to describe table {table_name}: {str(e)}")
          raise

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

  Helper function prepare_res specific to table dialect, used by describe_table to format the query result into a list of TextContent objects.

  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),
          )
      ]