LoginSignup
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@take_me

discordのボットをAWS Lambdaで(wslなしのWindows環境で)作る with Python

Last updated at Posted at 2024-05-23

概要

  • Windows (wslなし) で、discord のスラッシュコマンドのボットを AWS Lambda を用いて作成したよ

wsl なしの謎縛りプレイですが、Python 3.8 なら簡単に実現できることがわかったので共有します。
→ 注: Python 3.8は2024/10でサポート終了らしい...
→ 注2: 3.12でも簡単にできそう！

本記事は、下記リポジトリの部分的な解説も兼ねています。

環境

  • Windows 10
  • Python 3.8.0
    • 3.8じゃないと、windowsだけでは難しそうです
  • pipenv 2023.12.1
    • 別にPipenvでなくても、環境変数が読み込めて、かつPythonのバージョン管理ができれば何でもよいです
  • PowerShell 7.4.2

本記事の新規性

  • wsl なしで windows 環境のみでディスコボット作ったよ
  • event のボディの型を作ったよ

注意点

ここでは、ギルド (特定のサーバー) 限定のスラッシュコマンドを実装します。
Python の文法等は解説しません。
また、最初に述べたように、身内サーバーで運用しているBOTの部分的な解説を含んでいます。

作り方

ボットの作成と、使いたいサーバー (開発用サーバー等) への招待を済ませましょう。

また、AWSへの登録を済ませておいてください。

ボット作成の流れ

  1. トークン等の取得
  2. AWS Lambda のハンドラー作成
  3. コマンドの登録
  4. アップロード用の zip 作成
  5. API Gateway の追加とデプロイ
  6. API Gateway の URL をポータルへコピペ
  7. 完成！

まずはディレクトリとかの整理

# 3.8じゃないとだめです。それ以外のバージョンを使うにはwslを要します。
pipenv --python 3.8
# 実はインストールしなくても大丈夫です (後述)
pipenv install pynacl
# コマンドの登録に利用するスクリプトで使います
pipenv install --dev requests

mkdir app

トークンの設定

適当なディレクトリ (以下 bot_project_dir) で .env ファイルを作っておきます。

.env
APP_ID = "YOUR_APP_ID"
SERVER_ID = "YOUR_SERVER_ID"
BOT_TOKEN = "YOUR_BOT_TOKEN"

APP_ID は、dev portal で取得できるはず。
SERVER_ID は、ディスコの開発者モードをクライアントアプリで有効化して、サーバーで右クリックすると見れるはず。

AWS Lambda のハンドラー作成

AWS Lambda では、リクエストを処理するハンドラー関数を定義して指定する必要があります。
このセクションでは、下記のハンドラーを説明します。
詳細に興味がない場合は、下記のコードをapp/lambda_function.pyに貼り付けてください。
discord_types.py, lambda_types.py をダウンロード(またはコピペ)して、appフォルダにいれてください。

app/lambda_function.py
import json

from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError

from discord_types import (
    DiscordCommandRequestBody,
    DiscordCommandResponseBody,
    DiscordResponseInteractionType,
)
from lambda_types import (
    MyApiGatewayEvent,
    LambdaContext,
    MyApiGatewayResponse,
)

PUBLIC_KEY = "YOUR_PUBLIC_KEY"

def toResponse(
    body: DiscordCommandResponseBody, statusCode=200
) -> MyApiGatewayResponse:
    return {
        "statusCode": statusCode,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps(body),
    }

def discord_response(body: DiscordCommandRequestBody) -> MyApiGatewayResponse:
    name = body["data"]["name"]

    res: DiscordCommandResponseBody
    if name == "hello":
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "World!"},
        }
    else:
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "Command not found."},
        }
    return toResponse(res)

def verify_request(event: MyApiGatewayEvent):
    """
    Verify the request signature.
    """
    signature = event["headers"]["x-signature-ed25519"]
    timestamp = event["headers"]["x-signature-timestamp"]

    verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY))

    message = timestamp + event["body"]

    try:
        verify_key.verify(message.encode(), signature=bytes.fromhex(signature))
    except BadSignatureError:
        return False

    return True
    
def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    t = body["type"]
    if t == 1:
        # handle ping
        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps({"type": DiscordResponseInteractionType.PONG}),
        }
    elif t == 2:  # handle application command
        return discord_response(body)

    return {
        "statusCode": 400,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps("invalid request type"),
    }

以下、解説。

基本的には、ハンドラーは下記のような形になります。

def lambda_handler(event, context):
    return {
        "statusCode": 400,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps("invalid request type"),
    }

上記のように、event: リクエスト内容等を含むオブジェクトとcontext: AWS のサービスに関連する情報等を含むオブジェクトの二つを受け取って、http通信のレスポンスを返すものになっています。

context については、次の参考リンクで型情報がわかります。

event については、MyApiGatewayEvent のような辞書のようです。

# event の型
class MyApiGatewayEvent(tp.TypedDict):
    headers: tp.Dict[str, str]
    body: str

# レスポンスの型
class MyApiGatewayResponse(tp.TypedDict):
    statusCode: int
    headers: tp.Dict[str, str]  # "Content-Type: application/json" は必須っぽいです。
    body: str

json.loads(event["body"]) は、次のような型 (辞書) になっていることがわかりました (公式ドキュメントと睨めっこ)。

import typing as tp

class DiscordCommandRequestBodyDataOption(tp.TypedDict):
    type: DiscordApplicationCommandOptionType
    name: str
    value: tp.Any  # NOTE: type depends on `type`. e.g. str, int, float


class DiscordCommandRequestBodyData(tp.TypedDict):
    id: tp.Any
    name: str
    options: tp.List[DiscordCommandRequestBodyDataOption]


class DiscordCommandRequestBody(tp.TypedDict):
    type: tp.Literal[
        DiscordRequestInteractionType.PING,
        DiscordRequestInteractionType.APPLICATION_COMMAND,
    ]

    # only for `APPLICATION_COMMAND`
    data: DiscordCommandRequestBodyData

なお、上記の型は最低限のキーしか書いていませんので、複雑なコマンドの際には他にも色々必要見ないです。下記の公式リファレンスをご覧ください。

上記の型情報を踏まえると、ハンドラーは次のよう定義すればよいことになります。

def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    ...

では、後はコマンドに応じた処理を行うだけ...ではなくて、discordから飛んでくる ping に応答できる必要があります。

どうやら nacl とやらで認証を行えるようです:

ここで windows 縛りの弊害なのですが、上記のチュートリアル通りに行うと動きません。下記のようにLambda のログで怒られます。

[ERROR] Runtime.ImportModuleError: Unable to import module 'lambda_function': No module named 'nacl._sodium'

これは nacl がプラットフォームに依存したビルドを行っているからであり、windows 環境でビルドしたものと、Lambda のランタイム上の OS (Linux系) と互換性がないからです。

最も簡単な解決策は WSL を利用することですが、ここでは Layers を利用することでこれを解決できます。
したがってここでは pip install しません。

ということで、nacl をインストールせずに、認証用の関数は、参考記事よりほぼコピペして、次のように定義しましょう。

def verify_request(event: MyApiGatewayEvent):
    """
    Verify the request signature.
    """
    signature = event["headers"]["x-signature-ed25519"]
    timestamp = event["headers"]["x-signature-timestamp"]

    verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY))

    message = timestamp + event["body"]

    try:
        verify_key.verify(message.encode(), signature=bytes.fromhex(signature))
    except BadSignatureError:
        return False

    return True

認証用関数を用いて、Discord が飛ばしてくる ping に対して

  • 認証ができていないなら401を返す
  • 認証ができているなら200を返す

ようなハンドラーを実装します。

def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    t = body["type"]  # 1 は ping, 2 はコマンド と決まっています。
    if t == 1:
        # handle ping
        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps({"type": DiscordResponseInteractionType.PONG}),
        }
    elif t == 2:  # handle application command
        ...

これで後は、コマンドに応じた処理を書くだけです。
ハンドラーの最後に、受け取ったボディを処理してレスポンスを返す関数を定義します。

def discord_response(body: DiscordCommandRequestBody) -> MyApiGatewayResponse:
    name = body["data"]["name"]

    res: DiscordCommandResponseBody
    # hello コマンドをとりあえず定義
    if name == "hello":
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "World!"},
        }
    else:
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "Command not found."},
        }
    return toResponse(res)

コマンドの登録

スラッシュコマンドの登録は、リクエストをディスコのAPIに投げることで行います。

register.py
import os

import requests


APP_ID = os.environ["APP_ID"]
SERVER_ID = os.environ["SERVER_ID"]
BOT_TOKEN = os.environ["BOT_TOKEN"]

url = (
    f"https://discord.com/api/v10/applications/{APP_ID}/guilds/{SERVER_ID}/commands"
)

payload = [
    {
        "name": "hello",
        "description": "return world",
        "options": []  # 引数の設定等。今回は引数なしなので空の配列で。 See https://discord.com/developers/docs/interactions/application-commands#application-command-object-application-command-option-structure
    }
]

response = requests.put(
    url, headers={"Authorization": f"Bot {BOT_TOKEN}"}, json=payload
)

print(response.json())

このスクリプトを実行。

pipenv run python -u register.py

成功だと200が返ってくる。
サーバーコマンドなので、ほぼ即時にコマンドが登録されます。

アップロード用の zip 作成

Lambda では、依存関係も全てまとめてアップロードする必要があります。
今回、pynaclに依存していますが、これは後ほどLambdaの Layer で解決するので、単に zip を作るだけです。

Compress-Archive ./app/*.py -DestinationPath app.zip -Force

出来上がったzipファイルを、AWS Lambda でアップロードしてください。

参考 (公式):

API Gateway の追加とデプロイ

トリガーとして、API Gateway を追加します。
設定はすべてデフォルト値でおっけー。

Lambda の Layers に arn 値を指定で、下記の値を入力・追加してください。

arn:aws:lambda:ap-northeast-1:770693421928:layer:Klayers-p38-PyNaCl:2

これを追加すると、PyNaCl がインストールされた環境が用意されます。

これが Python 3.8 しか使えない所以で、先人が作成した Layers は下記のリポジトリで見ることができます。

PyNaCl がインストールされた Python の最新バージョンが 3.8 だったのです。本当は 3.12 使いたかった...
→ 自分でLayerを簡単に作成できるので、Python 3.12にすんなりと移行できました！

最後に、urlの値をコピーして、discord dev portal の INTERACTIONS ENDPOINT URL の欄に貼り付けます。
登録の際に早速 ping が飛ぶようで、ここで蹴られると実装が間違っていることになります。
Cloud Watch等でログを確認してみてください。

まとめ

本記事では、wslを使わずにwindowsだけでAWSのサーバーレス構成のディスコボットを作成するtipsを紹介しました。
また、Python 用の型情報を考察・定義しました。
型ガチガチPythonで、wslなし縛りwindows環境下で、ディスコボットを開発する非常に稀な人種に役に立つことを願います。

3
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?