Tools for Building Agents with Dune

The Dune Agent provides a set of tools to access and analyze crypto market data using the DUNE API. These tools allow users to query, execute, and retrieve insights on various crypto indexes and narratives, offering valuable data for tracking trends.

execute_query(execute)

Description: Initiates the execution of a Dune query using the DUNE API.

Parameters:

• execute (str): The ID of the query to execute.

Returns:

• Response object: Response from the Dune API with query execution details.

execute_query(execute)

def execute_query(execute):
    headers = {"X-DUNE-API-KEY": dunekey}
    url = "https://api.dune.com/api/v1/query/{execute}/execute"
    try:
        response = requests.post(url.format(execute=execute), headers=headers)
        logging.info(response.text)
        return response
    except requests.exceptions.RequestException as e:
        logging.error(f"Error fetching data: {e}")

wait_for_execution(execution_id, max_attempts=60, delay=5)

Description: Monitors the status of the query execution until completion.

Parameters:

• execution_id (str): The execution ID of the query.

• max_attempts (int): Maximum number of status checks (default is 60).

• delay (int): Delay between checks in seconds (default is 5).

Returns:

• Response object: The final status of the query execution.

wait_for_execution(execution_id, max_attempts=60, delay=5)

def wait_for_execution(execution_id, max_attempts=60, delay=5):
    url = "https://api.dune.com/api/v1/execution/{execution_id}/status"
    headers = {"X-DUNE-API-KEY": dunekey}
    attempts = 0
    while attempts < max_attempts:
        try:
            response = requests.get(url.format(execution_id=execution_id), headers=headers)
            logging.info(response.text)
        except requests.exceptions.RequestException as e:
            logging.error(f"Error fetching data: {e}")
        if response.json().get("is_execution_finished", False):
            logging.info("Execution finished!")
            return response
        attempts += 1
        time.sleep(delay)
    logging.info(f"Execution did not finish after {max_attempts} attempts.")
    return None

get_results(query_id)

Description: Fetches the results of a Dune query after its execution.

Parameters:

• query_id (str): The ID of the query to retrieve results from.

Returns:

• Response object: The results of the query.

get_results(query_id)

def get_results(query_id):
    headers = {"X-DUNE-API-KEY": dunekey}
    url = "https://api.dune.com/api/v1/query/{query_id}/results"
    try:
        response = requests.get(url.format(query_id=query_id), headers=headers)
        logging.info(response.text)
        return response
    except requests.exceptions.RequestException as e:
        logging.error(f"Error fetching data: {e}")

run_dune_query(query_id)

Description: Runs a Dune query and retrieves the results by managing execution and result retrieval.

Parameters:

• query_id (str): The ID of the query to execute.

Returns:

• dict: Results of the query in dictionary format.

run_dune_query(query_id)

def run_dune_query(query_id):
    try:
        execution = execute_query(query_id)
        execution_id = execution.json().get("execution_id")
        executed_query_id = wait_for_execution(execution_id)
        results = get_results(executed_query_id.json().get("query_id"))
        return results.json()
    except Exception as e:
        logging.error(f"An error occurred: {e}")
        return None

---

Index Functions

get_alpha_index()

Description: Fetches the data for the Alpha Index from the Dune API.

Parameters: None.

Returns:

• dict: Alpha Index data.

get_alpha_index()

def get_alpha_index():
    return run_dune_query(query_ids["Alpha"])

get_beta_index()

Description: Fetches the data for the Beta Index from the Dune API.

Parameters: None.

Returns:

• dict: Beta Index data.

get_beta_index()

def get_beta_index():
    return run_dune_query(query_ids["Beta"])

get_gamma_index()

Description: Fetches the data for the Gamma Index from the Dune API.

Parameters: None.

Returns:

• dict: Gamma Index data.

get_gamma_index()

def get_gamma_index():
    return run_dune_query(query_ids["Gamma"])

---

Narrative Index Functions

get_daily_narrative_index()

Description: Fetches the Daily Crypto Narrative Index (24h).

Parameters: None.

Returns:

• dict: Daily Crypto Narrative data.

get_daily_narrative_index()

def get_daily_narrative_index():
    return run_dune_query(query_ids["24h"])

get_weekly_narrative_index()

Description: Fetches the Weekly Crypto Narrative Index (7d).

Parameters: None.

Returns:

• dict: Weekly Crypto Narrative data.

get_weekly_narrative_index()

def get_monthly_narrative_index():
    return run_dune_query(query_ids["30d"])

get_monthly_narrative_index()

Description: Fetches the Monthly Crypto Narrative Index (30d).

Parameters: None.

Returns:

• dict: Monthly Crypto Narrative data.

get_monthly_narrative_index()

def get_monthly_narrative_index():
    return run_dune_query(query_ids["30d"])

get_quarterly_narrative_index()

Description: Fetches the Quarterly Crypto Narrative Index (90d).

Parameters: None.

Returns:

• dict: Quarterly Crypto Narrative data.

get_quarterly_narrative_index()

def get_quarterly_narrative_index():
    return run_dune_query(query_ids["90d"])