Chatbot with Twitter Direct Message API

Toru Furukawa
Dec 13, 2018 · 16 min read

pyspa Advent Calendar 2018 13日目なう。

Twitter の Directe Message API を使うと、Direct Message (以下 DM)を使った chatbot を実装できる。ときどき「どうやって作るんですか?」とか聞かれることがあって、ドキュメント読んで勝手に作れば、と思っていた。内緒だけど今でも思っている。それはそうなんだけど、この1年で DM API の beta が取れたり、機能が変わったり、手続きが増えたりとかで、分かりにくいとは思う。

というわけで、昨年からのアップデートをする。

免責事項

  • 個人的な調べ物の成果である。
  • この文書は概念実証を示すものであり、いかなる保証もしない。
  • 公開された情報だけを使っている。

使ったもの

特に最新版である必要はないはずだけれど、使ったバージョンを記載しておく。

手続き

Account Activity API というのを使うにあたって、以下の登録作業をする。

  1. 開発者登録 https://developer.twitter.com/en/apply/user
  2. アプリ作成 https://developer.twitter.com/en/apps
    Read, write, and direct messages を有効にする
    consumer key と consumer secret を後で使う
  3. Account Activity API 登録 https://developer.twitter.com/en/apply

すべての申請・登録・承認が完了すると https://developer.twitter.com/en/account/environments から dev environment label が見えるようになる。これも後でつかう。

OAuth

開発者登録したアカウントを chatbot にするなら、アプリの管理画面から手作業で Access Token と Access Token Secret を取得できるので、このセクションの作業は不要。

開発者登録アカウントとは別のアカウントを、chatbot にするのであれば以下の手順。

$ twurl authorize --consumer-key XXXXXXXX --consumer-secret XXXXXXXX
Go to https://api.twitter.com/oauth/authorize?oauth_consumer_key=XXXXXXXX&oauth_nonce=... and paste in the supplied PIN

https://api.twitter.com/oauth/authorize で始まる URL をブラウザで開く。すると Twitter OAuth の画面になるので、進めていく。

最後に番号が表示される。この番号を、ターミナルにコピペする。

1234567
Authorization successful

これで完了。Access Token/Secret は ~/.twurl に入っている。

$ cat ~/.twurlrc
---
profiles:
torufurukawa:
XXXXXXXX:
username: torufurukawa
consumer_key: XXXXXXXX
consumer_secret: XXXXXXXX
token: XXXXXXXX
secret: XXXXXXXX
configuration:
default_profile:
- torufurukawa
- XXXXXXXX

token と secret が、それぞれ Access Token と Access Token Secret 。後で使う。

サーバーのスケルトン

とりあえず hello を返すサーバーを作って、少しずつ chatbot サーバーにしていく。

Python のライブラリをインストール。

pip install Flask==1.0.2 requests==2.21.0 requests-oauthlib==1.0.0

サーバーアプリを作ります。main1.py より抜粋。

from flask import Flask
app = Flask(__name__)
@app.route("/")
def hello():
return "Hello World!"
if __name__ == '__main__':
app.run(host='127.0.0.1', port=8080, debug=True)

実行すると…

$ python main1.py
* Serving Flask app "main1" (lazy loading)
* Environment: production
WARNING: Do not use the development server in a production environment.
Use a production WSGI server instead.
* Debug mode: on
* Running on http://127.0.0.1:8080/ (Press CTRL+C to quit)
...

http://127.0.0.1:8080 を開いて「Hello World!」が表示されたらOK。

Webhook の登録

Account Activity API は、アカウントに関連するイベント、たとえばリツートとか、DMとか、いいねとかが起こったときに、Twitter のサーバーがWebhook URL にイベントを届ける。

手順としては、(1) webhook URL を register し、(2) 特定のアカウントを subscribe する。完了すると subscribe したアカウントのイベントが webhook URL に送信される。

register するとき、Twitter が webhook URL に対して GETリクエストする。そのレスポンスを確認して、問題なければ register が成功する。

何を確認するのかと。Twitter が crc_token なる文字列を送ってくるので、consumer secret をキーに、HMAC SHA256 でハッシュ化して返す。

環境変数 CONSUMER_SECRET を定義してから、以下のコードを書く。

main2.py より抜粋。

import os
import base64
import hashlib
import hmac

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['GET'])
def webhook_challenge():
'''Webhook URL の正当性を確認'''
key = os.getenv('CONSUMER_SECRET').encode()
msg = request.args.get('crc_token').encode()
hash = hmac.new(key, msg=msg, digestmod=hashlib.sha256).digest()
response = {'response_token': 'sha256=' + base64.b64encode(hash).decode()}
return jsonify(response)
...

で、サーバーを立ち上げておく。

$ python main2.py

このサーバーが外部からアクセスできるように、ngrok でトンネルする。

$ ngrok http 8080

すると下のような画面になるであろう。

ngrok by @inconshreveable                                       (Ctrl+C to quit)Session Status                online
Account Toru Furukawa (Plan: Free)
Version 2.2.8
Region United States (us)
Web Interface http://127.0.0.1:4040
Forwarding http://xxxxxxxx.ngrok.io -> localhost:8080
Forwarding https://xxxxxxxx.ngrok.io -> localhost:8080
Connections ttl opn rt1 rt5 p50 p90
0 0 0.00 0.00 0.00 0.00

ありがとう、inconreveable さん。このプロセスが動いている間、https://xxxxxxxx.ngrok.io へのアクセスは、ローカルの 8080 に渡される。サブドメインは起動ごとに変わる。有料プランにすると、固定できる。

以下の手順で register する

$ twurl -X POST --consumer-key XXXXXXXX --consumer-secret XXXXXXXX --access-token XXXXXXXX --token-secret XXXXXXXX "/1.1/account_activity/all/env-beta/webhooks.json?url=https://xxxxxxxx.ngrok.io/webhook"

すると、python main2.py を動かしているプロセスに、Twitter から ngrok 経由でアクセスがある。CRC トークンをつけて GETしてくるので、先程実装したチャレンジの計算をした結果を、JSONで返す。問題がなければ register が完了する。

続いて subscribe 。

$ twurl -X POST --consumer-key XXXXXXXX --consumer-secret XXXXXXXX --access-token XXXXXXXX --token-secret XXXXXXXX "/1.1/account_activity/all/env-beta/subscriptions.json"

成功すると、これ以降、アカウントに関するイベントがあると、webhook に POSTリクエストがくるようになる。

試しにこのアカウントに DM を送ったりすると、POSTリクエストがくる。だが、POSTリクエスト処理を実装していないので、サーバーはエラーを

エコー

話しかけられたらオウム返しする chatbot を作る。環境変数 CONSUMER_KEY, ACCESS_TOKEN, ACCESS_TOKEN_SECRET を定義しておく。

main3.py より。

...
SEND_ENDPOINT = 'https://api.twitter.com/1.1/direct_messages/events/new.json'

# Twitter API への OAuth1 接続
twitter = requests.Session()
twitter.auth = OAuth1(os.getenv('CONSUMER_KEY'),
os.getenv('CONSUMER_SECRET'),
os.getenv('ACCESS_TOKEN'),
os.getenv('ACCESS_TOKEN_SECRET'))

app = Flask(__name__)
...@app.route('/webhook', methods=['POST'])
def webhook():
data = request.json
print(data)
# Direct Message イベントを順に処理する
for event in data.get('direct_message_events', []):
# 送信者と受信者が同じだったら、何もしない。
receiver_id = event['message_create']['target']['recipient_id']
sender_id = event['message_create']['sender_id']
if receiver_id == sender_id:
print('same user ID: ', receiver_id)
continue

# テキストを抽出する
text = event['message_create']['message_data']['text']
print('Text:', text)

# テキストをそのまま返す
e = {
'event': {
'type': 'message_create',
'message_create': {
'target': {'recipient_id': sender_id},
'message_data': {'text': text}
}
}
}
resp = twitter.post(SEND_ENDPOINT, json=e)
print(resp.status_code)

# 200 OK 的なレスポンス
return 'OK'
...

送信者と受信者が同じだったら、応答しないようにしている。これで独り言を延々と繰り返すことを防いでいる。

メッセージの文字列を取り出して、再送信する。それだけ。

Quick Reply

DM っぽいインタフェースでは、直接文字を入力するだけでなく、選択肢を提示することがある。Facebook Messenger しかり、LINE しかり。そして Twitter DM しかり。Twitter では quick reply とよぶ。

ユーザーに対して、選択肢を提供してタップ・クリックで入力させるためのデータ構造である。各選択肢には metadata なるパラメータがある。これについては、次のセクションで。

main4.py より

...@app.route('/webhook', methods=['POST'])
def webhook():
...
data = request.json
# Direct Message イベントを順に処理する
for event in data.get('direct_message_events', []):
...
# Quick Reply と Button をつけて返す
e = {
'event': {
'type': 'message_create',
'message_create': {
'target': {'recipient_id': sender_id},
'message_data': {
'text': text + 'って言った?',
'quick_reply': {
'type': 'options',
'options': [
{
'label': 'はい',
'description': '言ってたらこっちを選んでね',
'metadata': 'yes'
},
{
'label': 'いいえ',
'description': '言ってなかったらこっちを',
'metadata': text
}
]
}
}
}
}
}
resp = twitter.post(SEND_ENDPOINT, json=e)
print(resp.status_code)
......

Quick Reply からの返信を受け取る

選択肢をタップすると text に加えて、 metadata も一緒に webhook に届く。上述の例では、直前のユーザーの発言が metadata に格納している。(他の値を設定してもよい)

{
"direct_message_events": {
[
{
"message_create": {
"sender_id": "3945351",
"target": {"recipient_id": "919753461918416897"},
"message_data": {
"text": "いいえ",
"quick_reply_response": {
"metadata": "ちんこ"
}
}
}
}
]
}
}

「いいえ」を受け取ったら、metadata を使って何か発言するように変更する。

main5.py

...@app.route('/webhook', methods=['POST'])
def webhook():
...
text = event['message_create']['message_data']['text']

# もし「いいえ」なら言い返す
if text == 'いいえ':
message_data = event['message_create']['message_data']
prev_text = message_data['quick_reply_response']['metadata']
message = '「' + prev_text + '」って言いましたよね。'
e = {
'event': {
'type': 'message_create',
'message_create': {
'target': {'recipient_id': sender_id},
'message_data': {'text': message},
}
}
}
resp = twitter.post(SEND_ENDPOINT, json=e)
print(resp.status_code)
continue
... ......

本番投入

実践投入するには、少なくとも以下を考慮する必要がある。

  • Account Activity API の有料プランが必要かも。
  • DM 送信の per app の rate limit が小さいので、別途申請が必要。
  • 上のコードはエラーチェックとか、rate の制御とかなんにもしていない。そういうのが必要。
  • rate limit は常に存在するので、送信ペースの制御。

まとめ

Twitter DM APIを使って、プライベートな chatbot を作る考え方を紹介した。事実上 Account Activity API は必要であろう。

明日は、初めてソフトウェア開発の仕事をし始めたときに trac を教えてくれた @feiz です。メリークリスマス、そして、ハッピーニューイヤー。

Welcome to a place where words matter. On Medium, smart voices and original ideas take center stage - with no ads in sight. Watch
Follow all the topics you care about, and we’ll deliver the best stories for you to your homepage and inbox. Explore
Get unlimited access to the best stories on Medium — and support writers while you’re at it. Just $5/month. Upgrade