MisskeyボットをPythonで作る

MisskeyのボットをPythonで作る方法について、現行バージョン(2024.x〜)での開発方法を書きます。

検証環境

OS: Windows 11
Python: 3.12.3

開発する前の準備

ボットを開発する際にwebsockets misskey.py を使用します。

py -3 -m pip install websockets misskey.py

でインストールします。

アクセストークンを作成する

Misskey の設定から「API」→「アクセストークンの発行」をクリックし、必要な権限を選んでからチェックマークをクリックします。
名前は、あとから何のために作成したのかを把握するのに必要なので必ず付けます。
とりあえず分からない時は「全て有効にする」でもいいでしょうし、心配な場合は権限を絞ります。
いずれにせよアクセストークンは外部に漏らさないことが重要です。

万が一漏らしてしまったり、不正に利用されていることが分かったら、設定から「API」→「アクセストークンの管理」より、作成するときに付けた名前で探してゴミ箱マークをクリックして無効化します。

こうして作成したアクセストークンを、今後使用していきます。
「わかった」を押すと二度と見れなくなるので、注意してください。

WebSocketに繋ぐ

タイムラインの内容はWebSocketでリアルタイムに流れてきます。

Misskey.ioの場合、URLはwss://misskey.io/streamingです。

このままでも接続できますが、できることが限られるので、作成したアクセストークンを使って接続します。URLに加えるだけです。

wss://misskey.io/streaming?i=[アクセストークン]

接続するまでをプログラムで書いていくと、こうなります。

import asyncio
import json
import websockets
from misskey import Misskey

TOKEN='YourToken'
msk = Misskey('misskey.io', i=TOKEN)
MY_ID = msk.i()['id']
WS_URL='wss://misskey.io/streaming?i='+TOKEN

async def runner():
    async with websockets.connect(WS_URL) as ws:
        # ...

asyncio.get_event_loop().run_until_complete(runner())
  

タイムラインの受信を始める

このままでは何も流れてこないので、Misskeyに「これを見たいのですが？」というリクエストをWebSocketで送ります。

そのときに「チャンネル」を指定しますが、主なチャンネルは5つあります。

チャンネル名	内容
main	フォローの通知など
homeTimeline	ホームタイムライン
localTimeline	ローカルタイムライン
hybridTimeline	ソーシャルタイムライン
globalTimeline	グローバルタイムライン

いくつでも指定でき、同時に受信できます。
ここでは、ホームタイムラインを見ることにします。

WebSocketに送る時は

{
    "type": "connect",
    "body": {
        "channel": "homeTimeline",
        "id": "test"
    }
}

となります。idの部分は複数受信するときに識別するものなので、今は適当でも構いませんが、ホームタイムライン＋ローカルタイムラインといった組み合わせで使う場合は必要になります。

Pythonで送る時は

async def runner():
    async with websockets.connect(WS_URL) as ws:
        await ws.send(json.dumps({
            "type": "connect",
            "body": {
                "channel": "homeTimeline",
                "id": "test"
            }
        }))

となります。

受信したものを処理する

await ws.send の下に、受信する処理を記述します。

while True:
    data = json.loads(await ws.recv())
    print(data)

こうすると、自分のアカウントで試している場合はフォロワーのノートがずらりと流れてきます。

ノートが流れてくる時、データ構造はこうなっています。

{
    "type": "channel",
    "body": {
        "type": "note",
        "body": {
            ...
        }
    }
}

ルートのtypeがchannel で、かつbodyの中のtypeがnote であれば、ノートが流れてきた、と判断することができます。

別関数で処理する

ここまできて、受信する部分を分けないとインデントが増えて見づらくなってくるので、ノートの処理をする専用の関数に処理を引き渡すことにします。

async def runner():
    async with websockets.connect(WS_URL) as ws:
    await ws.send(json.dumps({
        "type": "connect",
        "body": {
            "channel": "homeTimeline",
            "id": "test"
        }
    }))

    while True:
        data = json.loads(await ws.recv())
        print(data)
        if data['type'] == 'channel':
            if data['body']['type'] == 'note':
                note = data['body']['body']
                await on_note(note)

async def on_note(note):
    # ...

ノートを受信したらon_noteに処理を引き継いでいます。

メンションされたら返信してみる

メンションされると、誰からされたのかがWebSocketから流れてくるデータにはあるので、利用します。

async def on_note(note):
    if note.get('mentions'):
        if MY_ID in note['mentions']:
            msk.notes_create(text='呼んだ？', replyId=note['id'])

MY_ID、mskは、プログラム冒頭で宣言しています。
自分にメンションすると、「呼んだ？」と返信するはずです。

フォロバする

ホームタイムラインを見ている時、フォロワーがいなければ何も流れてこないので、フォローされたらボットからもフォローバックしてあげる必要があります。

処理を引き継ぐ部分に、フォローされたときに関数を呼び出すように追記します。
そして、mainチャンネルに接続するのも忘れないでください。

while True:
    data = json.loads(await ws.recv())
    print(data)
    if data['type'] == 'channel':
        if data['body']['type'] == 'note':
            note = data['body']['body']
            await on_note(note)

        if data['body']['type'] == 'followed':
            user = data['body']['body']
            await on_follow(user)

async def on_follow(user):
    try:
        msk.following_create(user['id'])
    except:
        pass

これで、フォローされたときに自動でフォローバックすることができます。
try-exceptで囲んでいるのは、相手が再びフォローしてきたときに、自分がすでにフォロー済みだったときにエラーが発生するので、それを無視するためです。

さいごに

PythonにはMisskeyのAPIラッパーやWebSocketのモジュールがありますので、そこまで難易度を上げずに簡易的なボットプログラムを作成することができました。

この記事で使ったMisskeyのAPIラッパーは、ゆずりょーさん作のモジュールです。とても使いやすく感謝しています！