LambdaでX APIを呼び出してみた

2024.03.08Update

並木

はじめに

こんにちは。ニフティ株式会社の並木です。
今回は「X API」を利用するための手順と、「AWS Lambda」から「X API」を呼び出す方法についてご紹介いたします。

背景

以前「Zapierを使ってRSSフィードの更新をトリガーにしたTwitterへの自動投稿機能を作ってみた」で紹介した方法で、RSSフィードの更新をトリガーにしたX（旧Twitter）への自動投稿機能を作成したのですが、上記記事の「④Twitterに文章を投稿したい」で紹介した機能が2023年9月に使用できなくなってしまいました。（詳細：https://help.zapier.com/hc/en-us/articles/18657531069965）
別の方法で自動投稿機能を復活できないかという話になり、前から使用していた「Zapier」を流用しつつ、新たに「Amazon API Gateway」「AWS Lambda」「X API」を使用して、自動投稿を実現することにしました。

構成図は以下の通りです。

構成図の【1】【2】については、「Zapierを使ってRSSフィードの更新をトリガーにしたTwitterへの自動投稿機能を作ってみた」の処理をそのまま流用します。
【4】の手順は「API GatewayとLambdaでX投稿するAPIを作ってみた」に記載しています。
【3】【6】については別途記事を執筆予定です。
「【5】LambdaからX APIを呼び出して投稿」が今回の説明範囲になります。

「X API」とは

外部のアプリケーションから、Xの機能やデータを利用するためのAPIです。
Freeプラン（無料プラン）では毎月1500件の投稿をすることができます。

「X API」を利用できるようにする

X Developer Platformにアクセスし、Freeプランの説明の下にある「Get started」ボタンを押下します。

先の画面へ進んでいくと、「X API」の使用用途を250文字以上で入力する画面が出てきます。（使用用途は英語で入力する必要があるので注意してください）
使用用途や規約同意の入力が完了すると、「Developer Portal」のページが表示されます。
「Projects & Apps」に、自動生成されたProjectとAppが表示されますので、App（下の画像の左メニューの赤枠部分）を押下します。
「Settings」が選択された状態で、そのまま下にスクロールしていくと「Set up」ボタンがあります。
こちらを押下すると「User authentication settings」画面に遷移します。

今回は「X API」を使った投稿（Write）がしたいので、「App permissions」で「Read and write」を選択します。

上記画面で「Save」を押下後、「Keys and tokens」を選択し、「API Key and Secret」と「Access Token and Secret」をそれぞれ生成します。（以下画像ではボタン名が「Regenerate」となっていますが、初めて生成する場合はボタン名が「Generate」となっています）
生成した「API Key and Secret」と「Access Token and Secret」は後で使用するためメモしておきます。

「AWS Lambda」から「X API」を呼び出す

「AWS Lambda」から「X API」を呼び出せるようにします。
こちらのAWSの公式記事の「4-2. AWS Lambda 関数のひな形を作成する」を参考にしながら以下を実施します。

ランタイム：Pythonの最新バージョンを選択
「AWS CloudShell」を使用してPythonモジュール「requests_oauthlib」をインストール
インストールしてきた「requests_oauthlib」をzipファイルにまとめ、Lambda関数にソースコードとしてアップロード

上記を実施した上で、Lambdaにコードを書いていきます。

import json

import os

from requests_oauthlib import OAuth1Session

import boto3

import ast

def lambda_handler(event, context):

#################################

# Secrets Managerからキーを取得

#################################

secret_name = "シークレットの名前"

region_name = "リージョン名"

# Create a Secrets Manager client

session = boto3.session.Session()

client = session.client(

service_name='secretsmanager',

region_name=region_name

)

get_secret_value_response = client.get_secret_value(

SecretId=secret_name

)

secret_data = get_secret_value_response['SecretString']

secret = ast.literal_eval(secret_data)

consumer_key = secret['API_KEY']

client_secret = secret['API_KEY_SECRET']

access_token = secret['ACCESS_TOKEN']

access_token_secret = secret['ACCESS_TOKEN_SECRET']

#################################

# Xにポストする

#################################

oauth = OAuth1Session(consumer_key, client_secret, access_token, access_token_secret)

text = event['text']

payload = {'text': text}

response = oauth.post(

"https://api.twitter.com/2/tweets",

json=payload,

)

if response.status_code == 201:

result = 'OK'

else:

result = 'NG'

return {

"statusCode": 200,

"headers": {

"Content-Type": "application/json"

"body": {

"result": result,

"api_response": response.text

}

Pythonモジュール「requests_oauthlib」の「OAuth1Session」を使用することで、認証周りの実装を簡単に行うことができます。
当初は、Zapierから直接「X API」を呼び出すことを検討していたのですが、ZapierにPythonの外部ライブラリを入れることができず、認証周りの実装が困難であることから、Lambdaで「OAuth1Session」を使用して実装することにしました。

OAuthセッションを作るのに、最初に生成した「API Key and Secret」と「Access Token and Secret」を使用します。
ハードコーディングを避けるため「AWS Secrets Manager」を使って管理・呼び出しをしています。
「AWS Secrets Manager」の使い方については、以前投稿した「LambdaでSecrets Managerを使ってみた」を参照してください。

X投稿が成功した場合は「OK」、失敗した場合は「NG」を返すようにしています。（SlackにX投稿結果を通知するのに使用します）

おわりに

LambdaでPythonモジュール「requests_oauthlib」を使用することで、簡単に「X API」を呼び出すことができました。
当初、Zapierだけで何とか実装できないかと試行錯誤し、認証周りの調査に時間をかけてしまったので、Lambdaで実装するという手段に早めに切り替えれば良かったと感じました。
実現したいことにあわせて、様々なサービスを組み合わせて柔軟に対応することの大切さを学びました。