Pythonで簡単なActivityPubリマインダーボットを構築する方法

AmaseCocoa @cocoa@hackers.pub

このチュートリアルでは、Pythonを使用して簡単なActivityPubボットを構築する方法を説明します。このボットはメンションを監視し、特定の形式のメッセージを受信すると、指定された遅延の後にユーザーにリマインダーを送信します。

例えば、ユーザーが「@reminder@your.host.com 10m check the oven」というメッセージでボットにメンションすると、ボットは10分後に「🔔 Reminder for @user: check the oven」というメッセージで返信します。

前提条件

このチュートリアルを進めるには、Python 3.10以上と以下のライブラリが必要です：

  • apkit[server]: PythonでActivityPubアプリケーションを構築するための強力なツールキット。FastAPIベースのコンポーネントを含むserverエクストラを使用します。
  • uvicorn: FastAPIアプリケーションを実行するためのASGIサーバー。
  • cryptography: ActivityPubに必要な暗号鍵の生成と管理に使用されます。
  • uv: オプションですが、推奨される高速パッケージマネージャー。

これらの依存関係はuvまたはpipを使用してインストールできます。

# uvで新しいプロジェクトを初期化
uv init

# 依存関係をインストール
uv add "apkit[server]" uvicorn cryptography

プロジェクト構造

プロジェクト構造はシンプルで、ボットのロジックのためのPythonファイル1つで構成されています。

.
├── main.py
└── private_key.pem
  • main.py: ボットのコードをすべて含みます。
  • private_key.pem: ボットのActorの秘密鍵。初回実行時に自動的に生成されます。

コードの解説

アプリケーションのロジックは以下のステップに分けることができます：

  1. インポートと設定: 必要なインポートと基本的な設定変数をセットアップします。
  2. 鍵の生成: アクティビティの署名に必要な暗号鍵を準備します。
  3. Actorの定義: Fediverseにおけるボットのアイデンティティを定義します。
  4. サーバーの初期化: apkit ActivityPubサーバーをセットアップします。
  5. データストレージ: 作成されたアクティビティのための簡単なインメモリストアを実装します。
  6. リマインダーロジック: リマインダーの解析と通知送信のためのコアロジックをコーディングします。
  7. エンドポイント定義: 必要なWebエンドポイント（/actor、/inboxなど）を作成します。
  8. アクティビティハンドラー: 他のサーバーからの着信アクティビティを処理します。
  9. アプリケーション起動: サーバーを実行します。

それでは、main.pyファイルの各セクションを詳しく見ていきましょう。

1. インポートと設定

まず、必要なモジュールをインポートし、ボットの基本設定を定義します。

# main.py

import asyncio
import logging
import re
import uuid
import os
from datetime import timedelta, datetime

# Imports from FastAPI, cryptography, and apkit
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization as crypto_serialization

from apkit.config import AppConfig
from apkit.server import ActivityPubServer
from apkit.server.types import Context, ActorKey
from apkit.server.responses import ActivityResponse
from apkit.models import (
    Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,
)
from apkit.client import WebfingerResource, WebfingerResult, WebfingerLink
from apkit.client.asyncio.client import ActivityPubClient

# --- Logging Setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Basic Configuration ---
HOST = "your.host.com"  # Replace with your domain
USER_ID = "reminder"      # The bot's username

your.host.comを、ボットがホストされる実際のドメインに置き換えてください。これらの値は、ボットの一意の識別子（例：@reminder@your.host.com）を決定します。

2. 鍵の生成と永続化

ActivityPubはサーバー間の通信を保護するためにHTTP署名を使用します。これには、各アクターが公開鍵/秘密鍵のペアを持つ必要があります。以下のコードは、秘密鍵が存在しない場合に生成してファイルに保存します。

# main.py (続き)

# --- Key Persistence ---
KEY_FILE = "private_key.pem"

# 秘密鍵が存在する場合はロード、存在しない場合は新しく生成
if os.path.exists(KEY_FILE):
    logger.info(f"Loading existing private key from {KEY_FILE}.")
    with open(KEY_FILE, "rb") as f:
        private_key = crypto_serialization.load_pem_private_key(f.read(), password=None)
else:
    logger.info(f"No key file found. Generating new private key and saving to {KEY_FILE}.")
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    with open(KEY_FILE, "wb") as f:
        f.write(private_key.private_bytes(
            encoding=crypto_serialization.Encoding.PEM,
            format=crypto_serialization.PrivateFormat.PKCS8,
            encryption_algorithm=crypto_serialization.NoEncryption()
        ))

# 秘密鍵から公開鍵を生成
public_key_pem = private_key.public_key().public_bytes(
    encoding=crypto_serialization.Encoding.PEM,
    format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')

3. Actorの定義

次に、ボットのActorを定義します。ActorはActivityPubネットワークにおけるボットのアイデンティティです。このエンティティは自動化されているため、Applicationタイプを使用します。

# main.py (続き)

# --- Actor Definition ---
actor = Application(
    id=f"https://{HOST}/actor",
    name="Reminder Bot",
    preferredUsername=USER_ID,
    summary="A bot that sends you reminders. Mention me like: @reminder 5m Check the oven",
    inbox=f"https://{HOST}/inbox",      # アクティビティを受信するエンドポイント
    outbox=f"https://{HOST}/outbox",    # アクティビティを送信するエンドポイント
    publicKey=CryptographicKey(
        id=f"https://{HOST}/actor#main-key",
        owner=f"https://{HOST}/actor",
        publicKeyPem=public_key_pem
    )
)

4. サーバーの初期化

apkitからActivityPubServerを初期化し、送信アクティビティに署名するためのActorの鍵を取得する関数を提供します。

# main.py (続き)

# --- Key Retrieval Function ---
async def get_keys_for_actor(identifier: str) -> list[ActorKey]:
    """指定されたActor IDの鍵を返します。"""
    if identifier == actor.id:
        return [ActorKey(key_id=actor.publicKey.id, private_key=private_key)]
    return []

# --- Server Initialization ---
app = ActivityPubServer(apkit_config=AppConfig(
    actor_keys=get_keys_for_actor  # 鍵取得関数を登録
))

5. インメモリストレージとキャッシュ

作成されたアクティビティを提供するには、それらをどこかに保存する必要があります。簡単にするために、この例では基本的なインメモリ辞書をストアとキャッシュとして使用します。本番アプリケーションでは、これを永続的なデータベース（SQLiteやPostgreSQLなど）と適切なキャッシュ（Redisなど）に置き換えるでしょう。

# main.py (続き)

# --- In-memory Store and Cache ---
ACTIVITY_STORE = {} # 作成されたアクティビティを保存するシンプルな辞書
CACHE = {}          # 最近アクセスされたアクティビティのキャッシュ
CACHE_TTL = timedelta(minutes=5) # キャッシュの有効期限（5分）

6. リマインダーの解析と送信ロジック

これはボットのコアロジックです。parse_reminder関数は正規表現を使用してメンションから遅延とメッセージを抽出し、send_reminderは通知をスケジュールします。

# main.py (続き)

# --- Reminder Parsing Logic ---
def parse_reminder(text: str) -> tuple[timedelta | None, str | None, str | None]:
    """'5m do something'のようなリマインダーテキストを解析します。"""
    # ... (実装は簡潔にするため省略)

# --- Reminder Sending Function ---
async def send_reminder(ctx: Context, delay: timedelta, message: str, target_actor: APKitActor, original_note: Note):
    """指定された遅延の後にリマインダーを送信します。"""
    logger.info(f"Scheduling reminder for {target_actor.id} in {delay}: '{message}'")
    await asyncio.sleep(delay.total_seconds()) # 非同期で待機
    
    logger.info(f"Sending reminder to {target_actor.id}")
    
    # リマインダーNoteを作成
    reminder_note = Note(...)
    # Createアクティビティでラップ
    reminder_create = Create(...)
    
    # 作成されたアクティビティを保存
    ACTIVITY_STORE[reminder_note.id] = reminder_note
    ACTIVITY_STORE[reminder_create.id] = reminder_create
    
    # アクティビティをターゲットアクターのインボックスに送信
    keys = await get_keys_for_actor(f"https://{HOST}/actor")
    await ctx.send(keys, target_actor, reminder_create)
    logger.info(f"Reminder sent to {target_actor.id}")

7. エンドポイント定義

必要なActivityPubエンドポイントを定義します。apkitはFastAPI上に構築されているため、標準のFastAPIデコレータを使用できます。主なエンドポイントは以下の通りです：

  • Webfinger: 他のサーバー上のユーザーが@user@hostのようなアドレスを使用してボットを発見できるようにします。これは連合のための重要な最初のステップです。
  • /actor: ボットのActorオブジェクトを提供します。これにはプロフィール情報と公開鍵が含まれています。
  • /inbox: ボットが他のサーバーからアクティビティを受信するエンドポイント。apkitはこのルートを自動的に処理し、次のステップで定義するハンドラーにアクティビティを転送します。
  • /outbox: ボットによって作成されたアクティビティのコレクション。ただし、これはプレースホルダーコレクションを返します。
  • /notes/{note_id}と/creates/{create_id}: ボットによって作成された特定のオブジェクトを提供するエンドポイント。他のサーバーが一意のIDでそれらを取得できるようにします。

これらのエンドポイントを定義するコードは次の通りです：

# main.py (続き)

# インボックスエンドポイントはapkitによって自動的に処理されます。
app.inbox("/inbox")

@app.webfinger()
async def webfinger_endpoint(request: Request, acct: WebfingerResource) -> Response:
    """ボットを発見可能にするためのWebfinger要求を処理します。"""
    if not acct.url:
        # acct:user@hostのようなリソースクエリを処理
        if acct.username == USER_ID and acct.host == HOST:
            link = WebfingerLink(rel="self", type="application/activity+json", href=actor.id)
            wf_result = WebfingerResult(subject=acct, links=[link])
            return JSONResponse(wf_result.to_json(), media_type="application/jrd+json")
    else:
        # URLを使用したリソースクエリを処理
        if acct.url == f"https://{HOST}/actor":
            link = WebfingerLink(rel="self", type="application/activity+json", href=actor.id)
            wf_result = WebfingerResult(subject=acct, links=[link])
            return JSONResponse(wf_result.to_json(), media_type="application/jrd+json")
    return JSONResponse({"message": "Not Found"}, status_code=404)

@app.get("/actor")
async def get_actor_endpoint():
    """ボットのActorオブジェクトを提供します。"""
    return ActivityResponse(actor)

@app.get("/outbox")
async def get_outbox_endpoint():
    """ボットが送信したアクティビティのコレクションを提供します。"""
    items = sorted(ACTIVITY_STORE.values(), key=lambda x: x.id, reverse=True)
    outbox_collection = OrderedCollection(
        id=actor.outbox,
        totalItems=len(items),
        orderedItems=items
    )
    return ActivityResponse(outbox_collection)

@app.get("/notes/{note_id}")
async def get_note_endpoint(note_id: uuid.UUID):
    """特定のNoteオブジェクトをキャッシングして提供します。"""
    note_uri = f"https://{HOST}/notes/{note_id}"
    
    # まずキャッシュをチェック
    if note_uri in CACHE and (datetime.now() - CACHE[note_uri]["timestamp"]) < CACHE_TTL:
        return ActivityResponse(CACHE[note_uri]["activity"])
        
    # キャッシュになければストアから取得
    if note_uri in ACTIVITY_STORE:
        activity = ACTIVITY_STORE[note_uri]
        # 返す前にキャッシュに追加
        CACHE[note_uri] = {"activity": activity, "timestamp": datetime.now()}
        return ActivityResponse(activity)
        
    return Response(status_code=404) # Not Found

@app.get("/creates/{create_id}")
async def get_create_endpoint(create_id: uuid.UUID):
    """特定のCreateアクティビティをキャッシングして提供します。"""
    create_uri = f"https://{HOST}/creates/{create_id}"
    
    if create_uri in CACHE and (datetime.now() - CACHE[create_uri]["timestamp"]) < CACHE_TTL:
        return ActivityResponse(CACHE[create_uri]["activity"])
        
    if create_uri in ACTIVITY_STORE:
        activity = ACTIVITY_STORE[create_uri]
        CACHE[create_uri] = {"activity": activity, "timestamp": datetime.now()}
        return ActivityResponse(activity)
        
    return Response(status_code=404)

8. アクティビティハンドラー

@app.on()デコレータを使用して、インボックスに投稿された特定のアクティビティタイプのハンドラーを定義します。

  • on_follow_activity: Followリクエストを自動的に受け入れます。
  • on_create_activity: 着信Createアクティビティ（特にNoteオブジェクト）を解析してリマインダーをスケジュールします。
# main.py (続き)

# Followアクティビティのハンドラー
@app.on(Follow)
async def on_follow_activity(ctx: Context):
    """フォローリクエストを自動的に受け入れます。"""
    # ... (実装は簡潔にするため省略)

# Createアクティビティのハンドラー
@app.on(Create)
async def on_create_activity(ctx: Context):
    """メンションを解析してリマインダーをスケジュールします。"""
    activity = ctx.activity
    # Noteでなければ無視
    if not (isinstance(activity, Create) and isinstance(activity.object, Note)):
        return Response(status_code=202)

    note = activity.object
    
    # ボットがメンションされているかチェック
    is_mentioned = any(
        isinstance(tag, Mention) and tag.href == actor.id for tag in (note.tag or [])
    )
    
    if not is_mentioned:
        return Response(status_code=202)

    # ... (リマインダーテキストを解析)
    delay, message, time_str = parse_reminder(command_text)

    # 解析が成功したら、リマインダーをバックグラウンドタスクとしてスケジュール
    if delay and message and sender_actor:
        asyncio.create_task(send_reminder(ctx, delay, message, sender_actor, note))
        reply_content = f"<p>✅ OK! I will remind you in {time_str}.</p>"
    else:
        # 解析に失敗した場合、使用方法の説明を送信
        reply_content = "<p>🤔 Sorry, I didn\'t understand. Please use the format: `@reminder [time] [message]`.</p><p>Example: `@reminder 10m Check the oven`</p>"

    # ... (返信Noteを作成して送信)

9. アプリケーションの実行

最後に、uvicornを使用してアプリケーションを実行します。

# main.py (続き)

if __name__ == "__main__":
    import uvicorn
    logger.info("Starting uvicorn server...")
    uvicorn.run(app, host="0.0.0.0", port=8000)

ボットの実行方法

  1. main.pyのHOSTとUSER_ID変数を環境に合わせて設定します。

  2. ターミナルからサーバーを実行します：

    uvicorn main:app --host 0.0.0.0 --port 8000

  3. ボットはhttp://0.0.0.0:8000で実行されます。

これで、Fediverse内のどこからでもボットにメンション（例：@reminder@your.host.com）してリマインダーを設定できます。

次のステップ

このチュートリアルでは、シンプルなActivityPubボットを作成する基本を説明しました。インメモリストレージのみを使用しているため、サーバーが再起動するとすべてのリマインダーが失われます。以下に潜在的な改善点をいくつか示します：

  • 永続的ストレージ: インメモリのACTIVITY_STOREをSQLiteやPostgreSQLなどのデータベースに置き換える。
  • 堅牢なタスクキューイング: CeleryとRedisやRabbitMQブローカーのような専用のタスクキューを使用して、サーバーが再起動してもリマインダーが失われないようにする。
  • 高度なコマンド: 定期的なリマインダーなど、より複雑なコマンドのサポートを追加する。

このガイドが、あなた自身のActivityPubアプリケーションを構築するための良い出発点となることを願っています！

https://fedi-libs.github.io/apkit/

https://github.com/fedi-libs/apkit

https://github.com/AmaseCocoa/activitypub-reminder-bot

15
1
0

1 comment

If you have a fediverse account, you can comment on this article from your own instance. Search https://hackers.pub/ap/articles/01994967-5928-7a5d-b7bd-9eac2425e8a5 on your instance and reply to it.

0
0
0