Merge pull request #212 from xdevplatform/subscriptions

missing scripts for subscriptions, webhooks, posts, follows

Author: VA-24

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()

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()

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()

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()

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()

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()

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()

python/requirements.txt

@@ -1 +1,3 @@
 xdk>=0.4.5
+flask>=3.0.0
+waitress>=3.0.0

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()