はじめに
LINE WORKS APIの新しいバージョン「API2.0」がリリースされ、従来のAPI (API1.0) は非推奨化および提供終了が予定されている (参考)。
今回は、PythonによるAPI2.0に対応したBot実装例を、シンプルなオウム返しBotをベースにまとめる。
- Node.js + Express編 はこちら
ソースコード
具体的な実装については以下を参照。FastAPI をサーバーとして使った。
以下、実装内容について一部抜粋して解説する。
実装解説
- 環境: Python 3.9
改ざんチェック
LINE WORKS APIのBotには、Callbackで受け取ったRequest Eventの改ざんチェックを行う機能がある。
参考 : https://developers.worksmobile.com/jp/reference/bot-callback?lang=ja
Request headerの X-WORKS-Signature によって渡される署名と、Developer ConsoleのBot画面で発行されている Bot Secret を用いて、Request bodyの改ざんチェックを行う。
以下、コード例。
import hashlib
import hmac
from base64 import b64encode, b64decode
def validate_request(body: bytes, signature: str, bot_secret: str) -> bool:
"""Validate request
:param body: request body
:param signature: value of X-WORKS-Signature header
:param bot_secret: Bot Secret
:return: is valid
"""
secretKey = bot_secret.encode()
payload = body
# Encode by HMAC-SHA256 algorithm
encoded_body = hmac.new(secretKey, payload, hashlib.sha256).digest()
# BASE64 encode
encoded_b64_body = b64encode(encoded_body).decode()
# Compare
return encoded_b64_body == signature
Access Token取得
トークに返答するために、まずはAccess Tokenを取得する。
取得の仕方としては、用意されている認証方法のうち「Service Account認証」というJWTを使った認可の仕組みを使ってAccess Tokenを取得する。
参考 : https://developers.worksmobile.com/jp/reference/authorization-sa?lang=ja
Access Token取得までの流れは、
- (事前準備) Developer ConsoleからAppを作成し、以下の各種認証情報を設定・取得する。
- Client ID
- Client Secret
- Service Account
- Private Key
- OAuth Scopeの設定
- JWTの生成。以下の情報を利用。
- Client ID
- Service Account
- Private Key
- Access Tokenを取得する。以下の情報を利用
- 生成したJWT
- Client ID
- Client Secret
- 必要なScope
- 今回はBotへの返答を行うのみであるため
botscopeを指定する。
- 今回はBotへの返答を行うのみであるため
以下、コード例。
import jwt
from datetime import datetime
import urllib
import json
import requests
BASE_AUTH_URL = "https://auth.worksmobile.com/oauth2/v2.0"
def __get_jwt(client_id: str, service_account: str, privatekey: str) -> str:
"""Generate JWT for access token
:param client_id: Client ID
:param service_account: Service Account
:param privatekey: Private Key
:return: JWT
"""
current_time = datetime.now().timestamp()
iss = client_id
sub = service_account
iat = current_time
exp = current_time + (60 * 60) # 1 hour
jws = jwt.encode(
{
"iss": iss,
"sub": sub,
"iat": iat,
"exp": exp
}, privatekey, algorithm="RS256")
return jws
def get_access_token(client_id: str, client_secret: str, service_account: str, privatekey: str, scope: str) -> dict:
"""Get Access Token
:param client_id: Client ID
:param client_secret: Client ID
:param service_account: Service Account
:param privatekey: Private Key
:param scope: OAuth Scope
:return: response
"""
# Get JWT
jwt = __get_jwt(client_id, service_account, privatekey)
# Get Access Token
url = '{}/token'.format(BASE_AUTH_URL)
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
params = {
"assertion": jwt,
"grant_type": urllib.parse.quote("urn:ietf:params:oauth:grant-type:jwt-bearer"),
"client_id": client_id,
"client_secret": client_secret,
"scope": scope,
}
form_data = params
r = requests.post(url=url, data=form_data, headers=headers)
body = json.loads(r.text)
return body
返答
トークに返答をする。返答には「メッセージ送信」のAPIを使う。
送信先ユーザーは、Request bodyの source.userId を指定する。
返答内容は、送られてきたテキストをそのまま返すため、以下のようになる。
{
"content": {
"type": "text",
"text": "{Reply text}"
}
}
参考 : https://developers.worksmobile.com/jp/reference/bot-send-text?lang=ja
API呼び出しの際は、先に取得したAccess Tokenを Authorization headerに指定する。
Authorization: Bearer {{Access Token}}
メッセージ送信のコード例は以下の通り。
import json
import requests
BASE_API_URL = "https://www.worksapis.com/v1.0"
def send_message_to_user(content: dict, bot_id: str, user_id: str, access_token: str):
"""Send message to a user
:param content: Message content
:param bot_id: Bot ID
:param user_id: User ID
:param access_token: Access Token
"""
url = "{}/bots/{}/users/{}/messages".format(BASE_API_URL, bot_id, user_id)
headers = {
'Content-Type' : 'application/json',
'Authorization' : "Bearer {}".format(access_token)
}
params = content
form_data = json.dumps(params)
r = requests.post(url=url, data=form_data, headers=headers)
r.raise_for_status()
また、APIにはRate Limitが設けられており、制限を超えた際は429のエラーが返る。その際は時間を置いて再送する必要がある。それを考慮した再送処理を実装する。
加えて、Access Tokenが期限切れの場合を考慮し、Tokenの再発行の処理も追加する。
以下、実装例。
for i in range(RETRY_COUNT_MAX):
try:
# Reply message
res = lineworks.send_message_to_user(res_content,
bot_id,
user_id,
global_data["access_token"])
except RequestException as e:
body = e.response.json()
status_code = e.response.status_code
if status_code == 401:
if body["code"] == "UNAUTHORIZED":
# Access Token has been expired.
# Update Access Token
logger.info("Update access token")
res = lineworks.get_access_token(client_id,
client_secret,
service_account_id,
privatekey,
SCOPE)
global_data["access_token"] = res["access_token"]
else:
logger.exception(e)
break
elif status_code == 429:
# Requests over rate limit.
logger.info("Over rate limit")
logger.info(body)
else:
logger.exception(e)
break
# wait and retry
time.sleep(2 ** i)
else:
break
まとめ
API2.0対応のLINE WORKS Botの実装例をまとめた。
API 2.0に対応した実装例ではあるが、改ざんチェックやAccess Token取得の部分など、API1.0のBotのコードから流用できたものが多い。
もしこれらサンプルに不具合がある場合は、この記事のコメントもしくはGithubリポジトリのIssuesでお知らせください。