Sample code to demo how to get threaded conversations using full-archive search

fas_conversation_lookup.py

import requests
import json

# Replace your own bearer token below from the academic research project in the Twitter developer portal
bearer_token = "REPLACE_ME"

# Replace with conversation ID below (Tweet ID of the root Tweet)
conversation_id = '1354143047324299264'

search_url = "https://api.twitter.com/2/tweets/search/all"

query_params = {
    'query': 'conversation_id:{}'.format(conversation_id),
    'tweet.fields': 'created_at',
    'start_time': '2021-01-25T00:00:00Z',
    'max_results': 500
}


def create_headers(bearer_token):
    headers = {"Authorization": "Bearer {}".format(bearer_token)}
    return headers


def connect_to_endpoint(url, headers, params):
    response = requests.request("GET",
                                search_url,
                                headers=headers,
                                params=params)
    print(response.status_code)
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    return response.json()


def main():
        headers = create_headers(bearer_token)
        json_response = connect_to_endpoint(search_url, headers, query_params)
        print(json.dumps(json_response, indent=4, sort_keys=True))


if __name__ == "__main__":
    main()