特定のXアカウントが投稿するたびに、Discordのチャンネル、Telegramのチャット、Slackにメッセージを届けたい。このガイドでは、アカウント監視と約60行のPythonで書いたリレーでこれを実現します。リレーは新しいツイートをWebhookで受け取り、当社から送られたものか確認して、チャットに投稿します。

アカウント監視は、プランに応じて各アカウントを少なくとも60秒、5秒、1秒ごとに確認するため、通知はツイートの数秒後に届きます。最小のプランは10アカウントで月$5です。

ステップ1：チャット側の送信先を作る

どれか1つ、または3つすべてを設定します。

  • Discord：チャンネルの編集 → 連携サービス → ウェブフック → 新しいウェブフック → ウェブフックURLをコピー。
  • Telegram：@BotFatherでボットを作成し、トークンをコピーします。ボットをグループかチャンネルに追加してチャットIDを調べます。たとえばメッセージを1つ送ってからhttps://api.telegram.org/bot<token>/getUpdatesを開くとわかります。
  • Slack：チャンネル用のIncoming Webhookを持つアプリを作成し、Webhook URLをコピーします。

リレーを動かすサーバーで、値を環境変数に設定します。

export TWITTRAPI_WEBHOOK_SECRET="whsec_..."   # from step 3
export DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/..."
export TELEGRAM_BOT_TOKEN="123456:ABC..."
export TELEGRAM_CHAT_ID="-1001234567890"
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."

使わないものは省いてかまいません。

ステップ2：リレーを動かす

リレーは生のボディで署名を検証し、すぐに応答してから、バックグラウンドのスレッドでツイートを転送します。チャット側のAPIが遅くても、Webhookがタイムアウトすることはありません。relay.pyとして保存します。

import hashlib
import hmac
import os
import threading
import time
 
import requests
from flask import Flask, abort, request
 
SECRET = os.environ["TWITTRAPI_WEBHOOK_SECRET"].encode()
app = Flask(__name__)
 
 
def verify(header: str, raw_body: bytes, tolerance: int = 300) -> bool:
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    t, sig = parts.get("t", ""), parts.get("v1", "")
    if not t.isdigit() or not sig or abs(time.time() - int(t)) > tolerance:
        return False
    expected = hmac.new(SECRET, t.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)
 
 
def forward(tweet: dict) -> None:
    text = f"@{tweet['author']['username']}: {tweet['text']}\n{tweet['url']}"
    if url := os.environ.get("DISCORD_WEBHOOK_URL"):
        requests.post(url, json={"content": text[:2000]}, timeout=10)
    if token := os.environ.get("TELEGRAM_BOT_TOKEN"):
        requests.post(
            f"https://api.telegram.org/bot{token}/sendMessage",
            json={"chat_id": os.environ["TELEGRAM_CHAT_ID"], "text": text[:4096]},
            timeout=10,
        )
    if url := os.environ.get("SLACK_WEBHOOK_URL"):
        requests.post(url, json={"text": text}, timeout=10)
 
 
@app.post("/hooks/twittrapi")
def hook():
    if not verify(request.headers.get("X-TwittrAPI-Signature", ""), request.get_data()):
        abort(401)
    event = request.get_json()
    if event["event"] == "tweet":
        threading.Thread(target=forward, args=(event["tweet"],), daemon=True).start()
    return "", 204
pip install flask requests
flask --app relay run --host 0.0.0.0 --port 8000

リレーは公開アドレスからhttpsでアクセスできる必要があります。既存のWebサーバーの背後に置くか、小さなホスティング環境にデプロイしてください。テスト中はngrokやCloudflare Tunnelなどのトンネルでもかまいません。

Discordは1メッセージ2,000文字、Telegramは4,096文字までなので、本文を切り詰めています。通常のツイートはこれより短いですが、Xの長文ポストはどちらの上限も超えることがあります。

ステップ3：Webhookの送信先をリレーにする

ダッシュボードのMonitorsでプランを申し込み、APIキーを使ってアカウントを追加し、Webhookを設定します。

import os
import requests
 
API = "https://api.twittrapi.com"
HEADERS = {"X-API-KEY": os.environ["TWITTRAPI_API_KEY"]}
 
requests.post(f"{API}/monitors/accounts", headers=HEADERS, json={"usernames": ["NASA", "SpaceX"]}, timeout=30)
r = requests.put(f"{API}/monitors/webhook", headers=HEADERS, json={"url": "https://relay.example.com/hooks/twittrapi"}, timeout=30)
print(r.json()["webhook"]["secret"])

表示されたwhsec_のシークレットをリレーのTWITTRAPI_WEBHOOK_SECRETに設定し、リレーを再起動します。

ステップ4：テストイベントを送る

print(requests.post(f"{API}/monitors/webhook/test", headers=HEADERS, timeout=30).json())

レスポンスで、リレーがイベントを受け付けたかどうかと、かかった時間がわかります。テストイベントは"event": "test"なので、上のリレーは受け付けるだけでチャットには投稿しません。実際の通知を見るには、監視中のアカウントが投稿するのを待ちます。

必要な通知だけを受け取る

監視中のアカウントのツイートは、リプライやリポストも含めてすべて送られます。送信前にforwardの中で絞り込みます。

def wanted(tweet: dict) -> bool:
    if tweet.get("replyTo") or tweet.get("retweetOf"):
        return False  # original posts only
    return "launch" in tweet["text"].lower()

forwardの先頭でif not wanted(tweet): returnを呼び出してください。tweet["author"]["username"]を見て、アカウントごとに別のチャンネルへ振り分けることもできます。

サーバーがない場合はn8n

n8n（英語のドキュメント）を使えば、Pythonは不要です。トリガーにWebhookノードを置いたワークフローを作り、その本番用URLを監視のWebhookに設定して、{{ $json.body.tweet.url }}を送るDiscord、Telegram、Slackのいずれかのノードにつなぎます。Webhookノードは当社の署名を検証しないので、URLは外部に漏らさないでください。

よくある質問

通知はどのくらいの速さで届きますか？

各アカウントは、プランに応じて少なくとも60秒、5秒、1秒ごとに確認されます。これに、ツイートがX上に表示されるまでのX側の遅れと、チャット側のAPIの処理時間が加わります。各イベントにはtweetCreatedAtとdetectedAtが含まれるので、実際の時間を測れます。

アカウントではなくキーワードで通知できますか？

アカウント監視が追うのはアカウントです。キーワードの場合は、since_idを付けて検索（英語）を定期的に呼び出し、結果を同じ方法で送ってください。

リレーが停止していたらどうなりますか？

配信は最大24時間リトライされ、イベントは7日間pull APIで読み取れます。手順はWebhookのガイドをご覧ください。

フォロワーの少ないアカウントでも使えますか？

使えます。フォロワー数の下限はありません。フォロワー50人のアカウントも、500万人のアカウントと同じ頻度で確認します。

出典

TwittrAPIはX Corp.とは提携していません。公開データのみを読み取ります。