From 5471b09a530f60cbaaa059a27ef932be0ab326be Mon Sep 17 00:00:00 2001 From: Vardhan Agnihotri Date: Thu, 19 Mar 2026 15:08:52 -0700 Subject: [PATCH] missing scripts for subscriptions, webhooks, posts, follows --- python/activity/create_subscription.py | 63 +++++++++++++++ python/activity/delete_subscription.py | 34 ++++++++ python/activity/list_subscriptions.py | 33 ++++++++ python/activity/stream_events.py | 39 ++++++++++ python/activity/update_subscription.py | 43 +++++++++++ python/posts/get_post_by_id.py | 37 +++++++++ python/posts/hide_reply.py | 69 +++++++++++++++++ python/requirements.txt | 2 + python/users/follow/follow_user.py | 70 +++++++++++++++++ python/users/follow/unfollow_user.py | 69 +++++++++++++++++ python/users/repost/get_reposts_of_me.py | 78 +++++++++++++++++++ python/webhooks/delete_webhook.py | 33 ++++++++ python/webhooks/list_webhooks.py | 28 +++++++ python/webhooks/register_webhook.py | 36 +++++++++ python/webhooks/validate_webhook.py | 35 +++++++++ python/webhooks/webhook_server.py | 98 ++++++++++++++++++++++++ 16 files changed, 767 insertions(+) create mode 100644 python/activity/create_subscription.py create mode 100644 python/activity/delete_subscription.py create mode 100644 python/activity/list_subscriptions.py create mode 100644 python/activity/stream_events.py create mode 100644 python/activity/update_subscription.py create mode 100644 python/posts/get_post_by_id.py create mode 100644 python/posts/hide_reply.py create mode 100644 python/users/follow/follow_user.py create mode 100644 python/users/follow/unfollow_user.py create mode 100644 python/users/repost/get_reposts_of_me.py create mode 100644 python/webhooks/delete_webhook.py create mode 100644 python/webhooks/list_webhooks.py create mode 100644 python/webhooks/register_webhook.py create mode 100644 python/webhooks/validate_webhook.py create mode 100644 python/webhooks/webhook_server.py diff --git a/python/activity/create_subscription.py b/python/activity/create_subscription.py new file mode 100644 index 0000000..84ac379 --- /dev/null +++ b/python/activity/create_subscription.py @@ -0,0 +1,63 @@ +""" +Create Activity Subscription - X API v2 +======================================== +Endpoint: POST https://api.x.com/2/activity/subscriptions +Docs: https://docs.x.com/x-api/activity/introduction + +Creates a subscription to receive real-time activity events for a specified +event type and filter. Once created, matching events will be delivered to the +activity stream (see stream_events.py) and optionally to a registered webhook. + +Supported public event types include: + - profile.update.bio + - profile.update.picture + - profile.update.banner + - profile.update.location + - profile.update.url + - profile.update.username + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the user ID you want to monitor for activity events +user_id = "2244994945" + +# Replace with the event type you want to subscribe to. +# See the supported event types listed in the docstring above. +event_type = "profile.update.bio" + +# Optional: replace with a registered webhook ID to also receive events via webhook delivery. +# If omitted, events are only available on the activity stream. +webhook_id = None + +def main(): + payload = { + "event_type": event_type, + "filter": { + "user_id": user_id + } + } + + # Attach a label to help identify this subscription in the stream + payload["tag"] = f"{event_type} for user {user_id}" + + # Optionally route events to a registered webhook in addition to the stream + if webhook_id: + payload["webhook_id"] = webhook_id + + response = client.activity.create_subscription(body=payload) + + print("Response code: 201") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/activity/delete_subscription.py b/python/activity/delete_subscription.py new file mode 100644 index 0000000..53159e1 --- /dev/null +++ b/python/activity/delete_subscription.py @@ -0,0 +1,34 @@ +""" +Delete Activity Subscription - X API v2 +======================================== +Endpoint: DELETE https://api.x.com/2/activity/subscriptions/:id +Docs: https://docs.x.com/x-api/activity/introduction + +Deletes an activity subscription. Once deleted, events matching that subscription +will no longer be delivered to the stream or associated webhook. Use +list_subscriptions.py to find the subscription_id you wish to remove. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the subscription ID you wish to delete. +# You can find subscription IDs by running list_subscriptions.py +subscription_id = "your-subscription-id" + +def main(): + response = client.activity.delete_subscription(subscription_id) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/activity/list_subscriptions.py b/python/activity/list_subscriptions.py new file mode 100644 index 0000000..c7e0f44 --- /dev/null +++ b/python/activity/list_subscriptions.py @@ -0,0 +1,33 @@ +""" +List Activity Subscriptions - X API v2 +======================================= +Endpoint: GET https://api.x.com/2/activity/subscriptions +Docs: https://docs.x.com/x-api/activity/introduction + +Returns all active activity subscriptions for your app. Use the subscription_id +from the response to update or delete individual subscriptions. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +def main(): + response = client.activity.get_subscriptions() + + # Access data attribute safely + response_data = getattr(response, 'data', None) + if response_data: + print(json.dumps(response_data, indent=4, sort_keys=True)) + else: + print(json.dumps(response, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/activity/stream_events.py b/python/activity/stream_events.py new file mode 100644 index 0000000..10ae6c9 --- /dev/null +++ b/python/activity/stream_events.py @@ -0,0 +1,39 @@ +""" +Activity Stream - X API v2 +========================== +Endpoint: GET https://api.x.com/2/activity/stream +Docs: https://docs.x.com/x-api/activity/introduction + +Opens a persistent HTTP connection and streams real-time activity events +matching your active subscriptions. Events are delivered as they occur on +the platform — no polling required. + +You must create at least one subscription (see create_subscription.py) before +events will be delivered to this stream. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +def main(): + print("Connecting to activity stream... (press Ctrl+C to stop)") + + # The stream() method returns a generator that yields events as they arrive. + # The SDK manages reconnection with exponential backoff automatically. + for event in client.activity.stream(): + # Access data attribute (model uses extra='allow' so data should be available) + # Use getattr with fallback in case data field is missing from response + event_data = getattr(event, 'data', None) + if event_data: + print(json.dumps(event_data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/activity/update_subscription.py b/python/activity/update_subscription.py new file mode 100644 index 0000000..025431a --- /dev/null +++ b/python/activity/update_subscription.py @@ -0,0 +1,43 @@ +""" +Update Activity Subscription - X API v2 +======================================== +Endpoint: PUT https://api.x.com/2/activity/subscriptions/:id +Docs: https://docs.x.com/x-api/activity/introduction + +Updates an existing activity subscription. You can change the filter (e.g. target +a different user ID), the tag, or the associated webhook. Use list_subscriptions.py +to find the subscription_id you wish to update. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the subscription ID you wish to update. +# You can find subscription IDs by running list_subscriptions.py +subscription_id = "your-subscription-id" + +# Replace with the updated user ID you want to monitor +updated_user_id = "2244994945" + +def main(): + payload = { + "filter": { + "user_id": updated_user_id + } + } + + response = client.activity.update_subscription(subscription_id, body=payload) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/posts/get_post_by_id.py b/python/posts/get_post_by_id.py new file mode 100644 index 0000000..7acf07a --- /dev/null +++ b/python/posts/get_post_by_id.py @@ -0,0 +1,37 @@ +""" +Single Post Lookup - X API v2 +============================= +Endpoint: GET https://api.x.com/2/tweets/:id +Docs: https://developer.x.com/en/docs/twitter-api/tweets/lookup/api-reference/get-tweets-id + +Authentication: Bearer Token (App-only) or OAuth (User Context) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the Post ID you want to look up +post_id = "post-id" + +def main(): + # Post fields are adjustable. Options include: + # attachments, author_id, context_annotations, conversation_id, + # created_at, entities, geo, id, in_reply_to_user_id, lang, + # non_public_metrics, organic_metrics, possibly_sensitive, + # promoted_metrics, public_metrics, referenced_tweets, + # source, text, and withheld + response = client.posts.get_by_id( + post_id, + tweet_fields=["created_at", "author_id", "lang", "source", "public_metrics", "entities"] + ) + + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/posts/hide_reply.py b/python/posts/hide_reply.py new file mode 100644 index 0000000..a5f923d --- /dev/null +++ b/python/posts/hide_reply.py @@ -0,0 +1,69 @@ +""" +Hide Reply - X API v2 +===================== +Endpoint: PUT https://api.x.com/2/tweets/:id/hidden +Docs: https://developer.x.com/en/docs/twitter-api/tweets/hide-replies/api-reference/put-tweets-id-hidden + +Authentication: OAuth 2.0 (User Context) +Required env vars: CLIENT_ID, CLIENT_SECRET + +Note: You can only hide or unhide replies to conversations you authored. +Pass hidden=True to hide a reply, or hidden=False to unhide one. +""" + +import os +import json +from xdk import Client +from xdk.oauth2_auth import OAuth2PKCEAuth + +# The code below sets the client ID and client secret from your environment variables +# To set environment variables on macOS or Linux, run the export commands below from the terminal: +# export CLIENT_ID='YOUR-CLIENT-ID' +# export CLIENT_SECRET='YOUR-CLIENT-SECRET' +client_id = os.environ.get("CLIENT_ID") +client_secret = os.environ.get("CLIENT_SECRET") + +# Replace the following URL with your callback URL, which can be obtained from your App's auth settings. +redirect_uri = "https://example.com" + +# Set the scopes +scopes = ["tweet.read", "tweet.write", "users.read", "offline.access"] + +# Replace with the ID of the reply you wish to hide. +# You can only hide replies to conversations you authored. +tweet_id = "reply-tweet-id-to-hide" + +def main(): + # Step 1: Create PKCE instance + auth = OAuth2PKCEAuth( + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + scope=scopes + ) + + # Step 2: Get authorization URL + auth_url = auth.get_authorization_url() + print("Visit the following URL to authorize your App on behalf of your X handle in a browser:") + print(auth_url) + + # Step 3: Handle callback + callback_url = input("Paste the full callback URL here: ") + + # Step 4: Exchange code for tokens + tokens = auth.fetch_token(authorization_response=callback_url) + access_token = tokens["access_token"] + + # Step 5: Create client + client = Client(access_token=access_token) + + # Step 6: Hide the reply + # Set hidden=False to unhide a previously hidden reply + response = client.posts.hide_reply(tweet_id, hidden=True) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/requirements.txt b/python/requirements.txt index ebc8553..2f878f3 100644 --- a/python/requirements.txt +++ b/python/requirements.txt @@ -1 +1,3 @@ xdk>=0.4.5 +flask>=3.0.0 +waitress>=3.0.0 diff --git a/python/users/follow/follow_user.py b/python/users/follow/follow_user.py new file mode 100644 index 0000000..3731c8d --- /dev/null +++ b/python/users/follow/follow_user.py @@ -0,0 +1,70 @@ +""" +Follow User - X API v2 +====================== +Endpoint: POST https://api.x.com/2/users/:id/following +Docs: https://developer.x.com/en/docs/twitter-api/users/follows/api-reference/post-users-id-following + +Authentication: OAuth 2.0 (User Context) +Required env vars: CLIENT_ID, CLIENT_SECRET +""" + +import os +import json +from xdk import Client +from xdk.oauth2_auth import OAuth2PKCEAuth + +# The code below sets the client ID and client secret from your environment variables +# To set environment variables on macOS or Linux, run the export commands below from the terminal: +# export CLIENT_ID='YOUR-CLIENT-ID' +# export CLIENT_SECRET='YOUR-CLIENT-SECRET' +client_id = os.environ.get("CLIENT_ID") +client_secret = os.environ.get("CLIENT_SECRET") + +# Replace the following URL with your callback URL, which can be obtained from your App's auth settings. +redirect_uri = "https://example.com" + +# Set the scopes +scopes = ["tweet.read", "users.read", "follows.write", "offline.access"] + +# Be sure to replace user-id-to-follow with the user id you wish to follow. +# You can find a user ID by using the user lookup endpoint +target_user_id = "user-id-to-follow" + +def main(): + # Step 1: Create PKCE instance + auth = OAuth2PKCEAuth( + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + scope=scopes + ) + + # Step 2: Get authorization URL + auth_url = auth.get_authorization_url() + print("Visit the following URL to authorize your App on behalf of your X handle in a browser:") + print(auth_url) + + # Step 3: Handle callback + callback_url = input("Paste the full callback URL here: ") + + # Step 4: Exchange code for tokens + tokens = auth.fetch_token(authorization_response=callback_url) + access_token = tokens["access_token"] + + # Step 5: Create client + client = Client(access_token=access_token) + + # Step 6: Get the authenticated user's ID + me_response = client.users.get_me() + user_id = me_response.data["id"] + + # Step 7: Follow the user + payload = {"target_user_id": target_user_id} + response = client.users.follow_user(user_id, body=payload) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/users/follow/unfollow_user.py b/python/users/follow/unfollow_user.py new file mode 100644 index 0000000..e640879 --- /dev/null +++ b/python/users/follow/unfollow_user.py @@ -0,0 +1,69 @@ +""" +Unfollow User - X API v2 +======================== +Endpoint: DELETE https://api.x.com/2/users/:source_user_id/following/:target_user_id +Docs: https://developer.x.com/en/docs/twitter-api/users/follows/api-reference/delete-users-source_user_id-following + +Authentication: OAuth 2.0 (User Context) +Required env vars: CLIENT_ID, CLIENT_SECRET +""" + +import os +import json +from xdk import Client +from xdk.oauth2_auth import OAuth2PKCEAuth + +# The code below sets the client ID and client secret from your environment variables +# To set environment variables on macOS or Linux, run the export commands below from the terminal: +# export CLIENT_ID='YOUR-CLIENT-ID' +# export CLIENT_SECRET='YOUR-CLIENT-SECRET' +client_id = os.environ.get("CLIENT_ID") +client_secret = os.environ.get("CLIENT_SECRET") + +# Replace the following URL with your callback URL, which can be obtained from your App's auth settings. +redirect_uri = "https://example.com" + +# Set the scopes +scopes = ["tweet.read", "users.read", "follows.write", "offline.access"] + +# Be sure to replace user-id-to-unfollow with the id of the user you wish to unfollow. +# You can find a user ID by using the user lookup endpoint +target_user_id = "user-id-to-unfollow" + +def main(): + # Step 1: Create PKCE instance + auth = OAuth2PKCEAuth( + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + scope=scopes + ) + + # Step 2: Get authorization URL + auth_url = auth.get_authorization_url() + print("Visit the following URL to authorize your App on behalf of your X handle in a browser:") + print(auth_url) + + # Step 3: Handle callback + callback_url = input("Paste the full callback URL here: ") + + # Step 4: Exchange code for tokens + tokens = auth.fetch_token(authorization_response=callback_url) + access_token = tokens["access_token"] + + # Step 5: Create client + client = Client(access_token=access_token) + + # Step 6: Get the authenticated user's ID + me_response = client.users.get_me() + user_id = me_response.data["id"] + + # Step 7: Unfollow the user + response = client.users.unfollow_user(user_id, target_user_id) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/users/repost/get_reposts_of_me.py b/python/users/repost/get_reposts_of_me.py new file mode 100644 index 0000000..977780f --- /dev/null +++ b/python/users/repost/get_reposts_of_me.py @@ -0,0 +1,78 @@ +""" +Reposts of Me - X API v2 +========================= +Endpoint: GET https://api.x.com/2/users/reposts_of_me +Docs: https://docs.x.com/x-api/users/get-reposts-of-me + +Authentication: OAuth 2.0 (User Context) +Required env vars: CLIENT_ID, CLIENT_SECRET + +Note: Returns posts from the authenticated user's timeline that have been reposted. +""" + +import os +import json +from xdk import Client +from xdk.oauth2_auth import OAuth2PKCEAuth + +# The code below sets the client ID and client secret from your environment variables +# To set environment variables on macOS or Linux, run the export commands below from the terminal: +# export CLIENT_ID='YOUR-CLIENT-ID' +# export CLIENT_SECRET='YOUR-CLIENT-SECRET' +client_id = os.environ.get("CLIENT_ID") +client_secret = os.environ.get("CLIENT_SECRET") + +# Replace the following URL with your callback URL, which can be obtained from your App's auth settings. +redirect_uri = "https://example.com" + +# Set the scopes +scopes = ["tweet.read", "users.read", "offline.access"] + +def main(): + # Step 1: Create PKCE instance + auth = OAuth2PKCEAuth( + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + scope=scopes + ) + + # Step 2: Get authorization URL + auth_url = auth.get_authorization_url() + print("Visit the following URL to authorize your App on behalf of your X handle in a browser:") + print(auth_url) + + # Step 3: Handle callback + callback_url = input("Paste the full callback URL here: ") + + # Step 4: Exchange code for tokens + tokens = auth.fetch_token(authorization_response=callback_url) + access_token = tokens["access_token"] + + # Step 5: Create client + client = Client(access_token=access_token) + + # Step 6: Get the authenticated user's posts that have been reposted + # Post fields are adjustable. Options include: + # attachments, author_id, context_annotations, conversation_id, + # created_at, entities, geo, id, in_reply_to_user_id, lang, + # non_public_metrics, organic_metrics, possibly_sensitive, + # promoted_metrics, public_metrics, referenced_tweets, + # source, text, and withheld + all_posts = [] + for page in client.users.get_reposts_of_me( + max_results=100, + tweet_fields=["created_at", "public_metrics"] + ): + # Access data attribute (model uses extra='allow' so data should be available) + # Use getattr with fallback in case data field is missing from response + page_data = getattr(page, 'data', []) or [] + all_posts.extend(page_data) + print(f"Fetched {len(page_data)} posts (total: {len(all_posts)})") + + print(f"\nTotal Reposted Posts: {len(all_posts)}") + print(json.dumps({"data": all_posts[:5]}, indent=4, sort_keys=True)) # Print first 5 as example + + +if __name__ == "__main__": + main() diff --git a/python/webhooks/delete_webhook.py b/python/webhooks/delete_webhook.py new file mode 100644 index 0000000..341bd05 --- /dev/null +++ b/python/webhooks/delete_webhook.py @@ -0,0 +1,33 @@ +""" +Delete Webhook - X API v2 +========================= +Endpoint: DELETE https://api.x.com/2/webhooks/:webhook_id +Docs: https://docs.x.com/x-api/webhooks/introduction + +Deletes a registered webhook. After deletion, X will stop delivering events +to the associated URL. Use list_webhooks.py to find your webhook_id. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the webhook ID you wish to delete. +# You can find your webhook IDs by running list_webhooks.py +webhook_id = "your-webhook-id" + +def main(): + response = client.webhooks.delete(webhook_id) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/webhooks/list_webhooks.py b/python/webhooks/list_webhooks.py new file mode 100644 index 0000000..de1a19f --- /dev/null +++ b/python/webhooks/list_webhooks.py @@ -0,0 +1,28 @@ +""" +List Webhooks - X API v2 +======================== +Endpoint: GET https://api.x.com/2/webhooks +Docs: https://docs.x.com/x-api/webhooks/introduction + +Returns all registered webhooks for your app. Use the webhook_id from the +response when managing subscriptions or deleting a webhook. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +def main(): + response = client.webhooks.get() + + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/webhooks/register_webhook.py b/python/webhooks/register_webhook.py new file mode 100644 index 0000000..b1d0462 --- /dev/null +++ b/python/webhooks/register_webhook.py @@ -0,0 +1,36 @@ +""" +Register Webhook - X API v2 +=========================== +Endpoint: POST https://api.x.com/2/webhooks +Docs: https://docs.x.com/x-api/webhooks/introduction + +Registers a new webhook URL with X. When you make this request, X immediately +sends a CRC challenge GET request to your URL to verify ownership. Your server +must respond correctly before the webhook is saved — see webhook_server.py. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with your publicly accessible HTTPS webhook URL. +# The URL must be reachable by X at the time of registration so the CRC check can complete. +# For local development you can use a tool like ngrok to expose a local server. +webhook_url = "https://your-domain.com/webhooks" + +def main(): + payload = {"url": webhook_url} + response = client.webhooks.create(body=payload) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/webhooks/validate_webhook.py b/python/webhooks/validate_webhook.py new file mode 100644 index 0000000..05269aa --- /dev/null +++ b/python/webhooks/validate_webhook.py @@ -0,0 +1,35 @@ +""" +Validate Webhook (Trigger CRC) - X API v2 +========================================== +Endpoint: PUT https://api.x.com/2/webhooks/:webhook_id +Docs: https://docs.x.com/x-api/webhooks/introduction + +Manually triggers a CRC (Challenge-Response Check) for the specified webhook. +Use this to re-validate your server's ownership or to re-enable a webhook that +was disabled due to failed CRC checks. Your webhook server must be running and +able to respond to the challenge before calling this endpoint. + +Authentication: Bearer Token (App-only) +Required env vars: BEARER_TOKEN +""" + +import os +import json +from xdk import Client + +bearer_token = os.environ.get("BEARER_TOKEN") +client = Client(bearer_token=bearer_token) + +# Replace with the webhook ID you wish to validate. +# You can find your webhook IDs by running list_webhooks.py +webhook_id = "your-webhook-id" + +def main(): + response = client.webhooks.validate(webhook_id) + + print("Response code: 200") + print(json.dumps(response.data, indent=4, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/python/webhooks/webhook_server.py b/python/webhooks/webhook_server.py new file mode 100644 index 0000000..b39fe77 --- /dev/null +++ b/python/webhooks/webhook_server.py @@ -0,0 +1,98 @@ +""" +Webhook Server - X API v2 +========================= +Docs: https://docs.x.com/x-api/webhooks/introduction + +This is a minimal webhook consumer server that handles two responsibilities: + 1. CRC (Challenge-Response Check) validation via GET — X sends a crc_token + and expects back an HMAC-SHA256 hash signed with your Consumer Secret. + This is required when registering a webhook and periodically thereafter + to confirm your server is still alive. + 2. Event delivery via POST — X sends account activity or filtered stream + events as JSON payloads to this endpoint in real time. + +To receive events you must: + 1. Run this server at a publicly accessible HTTPS URL (e.g. via ngrok). + 2. Register the URL with X: see register_webhook.py + 3. Subscribe user accounts or set stream rules to start receiving events. + +Authentication: Consumer Secret (HMAC-SHA256 for CRC validation) +Required env vars: CONSUMER_SECRET +Dependencies: flask, waitress (pip install flask waitress) +""" + +import base64 +import hashlib +import hmac +import json +import os +import sys + +from flask import Flask, jsonify, request +from waitress import serve + +app = Flask(__name__) + +# Your app's Consumer Secret — used to sign the CRC response. +# To set environment variables on macOS or Linux, run the export commands below from the terminal: +# export CONSUMER_SECRET='YOUR-CONSUMER-SECRET' +CONSUMER_SECRET = os.environ.get("CONSUMER_SECRET") +if CONSUMER_SECRET is None: + print("Missing consumer secret. Ensure CONSUMER_SECRET env var is set.") + sys.exit(1) + +HOST = "0.0.0.0" +PORT = 8080 + + +@app.route('/webhooks', methods=['GET', 'POST']) +def webhook_request(): + # Handle GET request (CRC challenge) + if request.method == 'GET': + crc_token = request.args.get('crc_token') + print(f"CRC Token received: {crc_token}") + + if crc_token is None: + print("Error: No crc_token found in the request.") + return jsonify({'error': 'No crc_token'}), 400 + + # Creates HMAC SHA-256 hash from the incoming token and your Consumer Secret + sha256_hash_digest = hmac.new( + CONSUMER_SECRET.encode('utf-8'), + msg=crc_token.encode('utf-8'), + digestmod=hashlib.sha256 + ).digest() + + # Construct response data with base64 encoded hash + response = { + 'response_token': 'sha256=' + base64.b64encode(sha256_hash_digest).decode('utf-8') + } + + # Returns properly formatted json response + return jsonify(response) + + # Handle POST request (webhook event delivery) + elif request.method == 'POST': + event_data = request.get_json() + if event_data: + print(json.dumps(event_data, indent=2)) + else: + # Log if the request body wasn't JSON or was empty + print(f"Body: {request.data.decode('utf-8')}") + + # Return 200 OK immediately to acknowledge receipt. + # X will retry delivery if it does not receive a 2xx response promptly. + return '', 200 + + return 'Method Not Allowed', 405 + + +def main(): + print("--- Starting Webhook Server ---") + print(f"Listening on {HOST}:{PORT}") + print("Expose this server publicly (e.g. via ngrok) then register the URL with register_webhook.py") + serve(app, host=HOST, port=PORT) + + +if __name__ == '__main__': + main()