This tutorial will guide you through building a simple ActivityPub bot using Python. The bot will listen for mentions and, when it receives a message in a specific format, it will schedule and send a reminder back to the user after a specified delay.

For example, if a user mentions the bot with a message like "@reminder@your.host.com 10m check the oven", the bot will reply 10 minutes later with a message like "🔔 Reminder for @user: check the oven".

Prerequisites

To follow this tutorial, you will need Python 3.10+ and the following libraries:

• apkit[server]: A powerful toolkit for building ActivityPub applications in Python. We use the server extra, which includes FastAPI-based components.
• uvicorn: An ASGI server to run our FastAPI application.
• cryptography: Used for generating and managing the cryptographic keys required for ActivityPub.
• uv: An optional but recommended fast package manager.

You can install these dependencies using uv or pip.

# Initialize a new project with uvuv init# Install dependenciesuv add "apkit[server]" uvicorn cryptography

Project Structure

The project structure is minimal, consisting of a single Python file for our bot's logic.

.├── [main.py](http://main.py)└── private_key.pem

• main.py: Contains all the code for the bot.
• private_key.pem: The private key for the bot's Actor. This will be generated automatically on the first run.

Code Walkthrough

Our application logic can be broken down into the following steps:

1. Imports and Configuration: Set up necessary imports and basic configuration variables.
2. Key Generation: Prepare the cryptographic keys needed for signing activities.
3. Actor Definition: Define the bot's identity on the Fediverse.
4. Server Initialization: Set up the apkit ActivityPub server.
5. Data Storage: Implement a simple in-memory store for created activities.
6. Reminder Logic: Code the core logic for parsing reminders and sending notifications.
7. Endpoint Definitions: Create the necessary web endpoints (/actor, /inbox, etc.).
8. Activity Handlers: Process incoming activities from other servers.
9. Application Startup: Run the server.

Let's dive into each section of the main.py file.

1. Imports and Configuration

First, we import the necessary modules and define the basic configuration for our bot.

# [main.py](http://main.py)import asyncioimport loggingimport reimport uuidimport osfrom datetime import timedelta, datetime# Imports from FastAPI, cryptography, and apkitfrom fastapi import Request, Responsefrom fastapi.responses import JSONResponsefrom cryptography.hazmat.primitives.asymmetric import rsafrom cryptography.hazmat.primitives import serialization as crypto_serializationfrom apkit.config import AppConfigfrom apkit.server import ActivityPubServerfrom apkit.server.types import Context, ActorKeyfrom apkit.server.responses import ActivityResponsefrom apkit.models import ( Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,)from apkit.client import WebfingerResource, WebfingerResult, WebfingerLinkfrom 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 domainUSER_ID = "reminder" # The bot's username

Make sure to replace your.host.com with the actual domain where your bot will be hosted. These values determine your bot's unique identifier (e.g., @reminder@your.host.com).

2. Key Generation and Persistence

ActivityPub uses HTTP Signatures to secure communication between servers. This requires each actor to have a public/private key pair. The following code generates a private key and saves it to a file if one doesn't already exist.

# [main.py](http://main.py) (continued)# --- Key Persistence ---KEY_FILE = "private_key.pem"# Load the private key if it exists, otherwise generate a new oneif 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() ))# Generate the public key from the private keypublic_key_pem = private_key.public_key().public_bytes( encoding=crypto_serialization.Encoding.PEM, format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo).decode('utf-8')

3. Actor Definition

Next, we define the bot's Actor. The Actor is the bot's identity in the ActivityPub network. We use the Application type, as this entity is automated.

# [main.py](http://main.py) (continued)# --- 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", # Endpoint for receiving activities outbox=f"https://{HOST}/outbox", # Endpoint for sending activities publicKey=CryptographicKey( id=f"https://{HOST}/actor#main-key", owner=f"https://{HOST}/actor", publicKeyPem=public_key_pem ))

4. Server Initialization

We initialize the ActivityPubServer from apkit, providing it with a function to retrieve our Actor's keys for signing outgoing activities.

# [main.py](http://main.py) (continued)# --- Key Retrieval Function ---async def get_keys_for_actor(identifier: str) -> list[ActorKey]: """Returns the key for a given 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 # Register the key retrieval function))

5. In-Memory Storage and Cache

To serve created activities, we need to store them somewhere. For simplicity, this example uses a basic in-memory dictionary as a store and a cache. In a production application, you would replace this with a persistent database (like SQLite or PostgreSQL) and a proper cache (like Redis).

# [main.py](http://main.py) (continued)# --- In-memory Store and Cache ---ACTIVITY_STORE = {} # A simple dict to store created activitiesCACHE = {} # A cache for recently accessed activitiesCACHE_TTL = timedelta(minutes=5) # Cache expiration time (5 minutes)

6. Reminder Parsing and Sending Logic

This is the core logic of our bot. The parse_reminder function uses a regular expression to extract the delay and message from a mention, and send_reminder schedules the notification.

# [main.py](http://main.py) (continued)# --- Reminder Parsing Logic ---def parse_reminder(text: str) -> tuple[timedelta | None, str | None, str | None]: """Parses reminder text like '5m do something'.""" # ... (implementation omitted for brevity)# --- Reminder Sending Function ---async def send_reminder(ctx: Context, delay: timedelta, message: str, target_actor: APKitActor, original_note: Note): """Waits for a specified delay and then sends a reminder.""" logger.info(f"Scheduling reminder for {target_actor.id} in {delay}: '{message}'") await asyncio.sleep(delay.total_seconds()) # Asynchronously wait logger.info(f"Sending reminder to {target_actor.id}") # Create the reminder Note reminder_note = Note(...) # Wrap it in a Create activity reminder_create = Create(...) # Store the created activities ACTIVITY_STORE[reminder_note.id] = reminder_note ACTIVITY_STORE[reminder_create.id] = reminder_create # Send the activity to the target actor's inbox 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. Endpoint Definitions

We define the required ActivityPub endpoints. Since apkit is built on FastAPI, we can use standard FastAPI decorators. The main endpoints are:

• Webfinger: Allows users on other servers to discover the bot using an address like @user@host. This is a crucial first step for federation.
• /actor: Serves the bot's Actor object, which contains its profile information and public key.
• /inbox: The endpoint where the bot receives activities from other servers. apkit handles this route automatically, directing activities to the handlers we'll define in the next step.
• /outbox: A collection of the activities created by the bot. but this returns placeholder collection.
• /notes/{note_id} and /creates/{create_id}: Endpoints to serve specific objects created by the bot, allowing other servers to fetch them by their unique ID.

Here is the code for defining these endpoints:

# [main.py](http://main.py) (continued)# The inbox endpoint is handled by apkit automatically.app.inbox("/inbox") @/span app.webfinger()async def webfinger_endpoint(request: Request, acct: WebfingerResource) -> Response: """Handles Webfinger requests to make the bot discoverable.""" if not acct.url: # Handle resource queries like 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: # Handle resource queries using a 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) @/span app.get("/actor")async def get_actor_endpoint(): """Serves the bot's Actor object.""" return ActivityResponse(actor) @/span app.get("/outbox")async def get_outbox_endpoint(): """Serves a collection of the bot's sent activities.""" 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) @/span app.get("/notes/{note_id}")async def get_note_endpoint(note_id: uuid.UUID): """Serves a specific Note object, with caching.""" note_uri = f"https://{HOST}/notes/{note_id}" # Check cache first if note_uri in CACHE and (datetime.now() - CACHE[note_uri]["timestamp"]) < CACHE_TTL: return ActivityResponse(CACHE[note_uri]["activity"]) # If not in cache, get from store if note_uri in ACTIVITY_STORE: activity = ACTIVITY_STORE[note_uri] # Add to cache before returning CACHE[note_uri] = {"activity": activity, "timestamp": datetime.now()} return ActivityResponse(activity) return Response(status_code=404) # Not Found @/span app.get("/creates/{create_id}")async def get_create_endpoint(create_id: uuid.UUID): """Serves a specific Create activity, with caching.""" 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. Activity Handlers

We use the @app.on() decorator to define handlers for specific activity types posted to our inbox.

• on_follow_activity: Automatically accepts Follow requests.
• on_create_activity: Parses incoming Create activities (specifically for Note objects) to schedule reminders.

# [main.py](http://main.py) (continued)# Handler for Follow activities @/span app.on(Follow)async def on_follow_activity(ctx: Context): """Automatically accepts follow requests.""" # ... (implementation omitted for brevity)# Handler for Create activities @/span app.on(Create)async def on_create_activity(ctx: Context): """Parses mentions to schedule reminders.""" activity = ctx.activity # Ignore if it's not a Note if not (isinstance(activity, Create) and isinstance(activity.object, Note)): return Response(status_code=202) note = activity.object # Check if the bot was mentioned 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) # ... (Parse reminder text) delay, message, time_str = parse_reminder(command_text) # If parsing is successful, schedule the reminder as a background task 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: # If parsing fails, send usage instructions reply_content = "<p>🤔 Sorry, I didn\'t understand. Please use the format: `@reminder [time] [message]`.</p><p>Example: `@reminder 10m Check the oven`</p>" # ... (Create and send the reply Note)

9. Running the Application

Finally, we run the application using uvicorn.

# [main.py](http://main.py) (continued)if __name__ == "__main__": import uvicorn logger.info("Starting uvicorn server...") uvicorn.run(app, host="0.0.0.0", port=8000)

How to Run the Bot

1. Set the HOST and USER_ID variables in main.py to match your environment.

2. Run the server from your terminal:

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

3. Your bot will be running at http://0.0.0.0:8000.

Now you can mention your bot from anywhere in the Fediverse (e.g., @reminder@your.host.com) to set a reminder.

Next Steps

This tutorial covers the basics of creating a simple ActivityPub bot. Since it only uses in-memory storage, all reminders will be lost on server restart. Here are some potential improvements:

• Persistent Storage: Replace the in-memory ACTIVITY_STORE with a database like SQLite or PostgreSQL.
• Robust Task Queuing: Use a dedicated task queue like Celery with a Redis or RabbitMQ broker to ensure reminders are not lost if the server restarts.
• Advanced Commands: Add support for more complex commands, such as recurring reminders.

We hope this guide serves as a good starting point for building your own ActivityPub applications!

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

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

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

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

例えば、ユーザーが「@reminder@your.host.com 10m check the oven」というメッセージでボットにメンションすると、ボットは10分後に「🔔 Reminder for @/span 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](http://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](http://main.py)import asyncioimport loggingimport reimport uuidimport osfrom datetime import timedelta, datetime# Imports from FastAPI, cryptography, and apkitfrom fastapi import Request, Responsefrom fastapi.responses import JSONResponsefrom cryptography.hazmat.primitives.asymmetric import rsafrom cryptography.hazmat.primitives import serialization as crypto_serializationfrom apkit.config import AppConfigfrom apkit.server import ActivityPubServerfrom apkit.server.types import Context, ActorKeyfrom apkit.server.responses import ActivityResponsefrom apkit.models import ( Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,)from apkit.client import WebfingerResource, WebfingerResult, WebfingerLinkfrom 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 domainUSER_ID = "reminder" # The bot's username

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

2. 鍵の生成と永続化

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

# [main.py](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](http://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](http://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](http://main.py) (続き)# --- In-memory Store and Cache ---ACTIVITY_STORE = {} # 作成されたアクティビティを保存するシンプルな辞書CACHE = {} # 最近アクセスされたアクティビティのキャッシュCACHE_TTL = timedelta(minutes=5) # キャッシュの有効期限（5分）

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

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

# [main.py](http://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](http://main.py) (続き)# インボックスエンドポイントはapkitによって自動的に処理されます。app.inbox("/inbox") @/span 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) @/span app.get("/actor")async def get_actor_endpoint(): """ボットのActorオブジェクトを提供します。""" return ActivityResponse(actor) @/span 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) @/span 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 @/span 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](http://main.py) (続き)# Followアクティビティのハンドラー @/span app.on(Follow)async def on_follow_activity(ctx: Context): """フォローリクエストを自動的に受け入れます。""" # ... (実装は簡潔にするため省略)# Createアクティビティのハンドラー @/span 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](http://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

This tutorial will guide you through building a simple ActivityPub bot using Python. The bot will listen for mentions and, when it receives a message in a specific format, it will schedule and send a reminder back to the user after a specified delay.

For example, if a user mentions the bot with a message like "@reminder@your.host.com 10m check the oven", the bot will reply 10 minutes later with a message like "🔔 Reminder for @user: check the oven".

Prerequisites

To follow this tutorial, you will need Python 3.10+ and the following libraries:

• apkit[server]: A powerful toolkit for building ActivityPub applications in Python. We use the server extra, which includes FastAPI-based components.
• uvicorn: An ASGI server to run our FastAPI application.
• cryptography: Used for generating and managing the cryptographic keys required for ActivityPub.
• uv: An optional but recommended fast package manager.

You can install these dependencies using uv or pip.

# Initialize a new project with uvuv init# Install dependenciesuv add "apkit[server]" uvicorn cryptography

Project Structure

The project structure is minimal, consisting of a single Python file for our bot's logic.

.├── [main.py](http://main.py)└── private_key.pem

• main.py: Contains all the code for the bot.
• private_key.pem: The private key for the bot's Actor. This will be generated automatically on the first run.

Code Walkthrough

Our application logic can be broken down into the following steps:

1. Imports and Configuration: Set up necessary imports and basic configuration variables.
2. Key Generation: Prepare the cryptographic keys needed for signing activities.
3. Actor Definition: Define the bot's identity on the Fediverse.
4. Server Initialization: Set up the apkit ActivityPub server.
5. Data Storage: Implement a simple in-memory store for created activities.
6. Reminder Logic: Code the core logic for parsing reminders and sending notifications.
7. Endpoint Definitions: Create the necessary web endpoints (/actor, /inbox, etc.).
8. Activity Handlers: Process incoming activities from other servers.
9. Application Startup: Run the server.

Let's dive into each section of the main.py file.

1. Imports and Configuration

First, we import the necessary modules and define the basic configuration for our bot.

# [main.py](http://main.py)import asyncioimport loggingimport reimport uuidimport osfrom datetime import timedelta, datetime# Imports from FastAPI, cryptography, and apkitfrom fastapi import Request, Responsefrom fastapi.responses import JSONResponsefrom cryptography.hazmat.primitives.asymmetric import rsafrom cryptography.hazmat.primitives import serialization as crypto_serializationfrom apkit.config import AppConfigfrom apkit.server import ActivityPubServerfrom apkit.server.types import Context, ActorKeyfrom apkit.server.responses import ActivityResponsefrom apkit.models import ( Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,)from apkit.client import WebfingerResource, WebfingerResult, WebfingerLinkfrom 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 domainUSER_ID = "reminder" # The bot's username

Make sure to replace your.host.com with the actual domain where your bot will be hosted. These values determine your bot's unique identifier (e.g., @reminder@your.host.com).

2. Key Generation and Persistence

ActivityPub uses HTTP Signatures to secure communication between servers. This requires each actor to have a public/private key pair. The following code generates a private key and saves it to a file if one doesn't already exist.

# [main.py](http://main.py) (continued)# --- Key Persistence ---KEY_FILE = "private_key.pem"# Load the private key if it exists, otherwise generate a new oneif 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() ))# Generate the public key from the private keypublic_key_pem = private_key.public_key().public_bytes( encoding=crypto_serialization.Encoding.PEM, format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo).decode('utf-8')

3. Actor Definition

Next, we define the bot's Actor. The Actor is the bot's identity in the ActivityPub network. We use the Application type, as this entity is automated.

# [main.py](http://main.py) (continued)# --- 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", # Endpoint for receiving activities outbox=f"https://{HOST}/outbox", # Endpoint for sending activities publicKey=CryptographicKey( id=f"https://{HOST}/actor#main-key", owner=f"https://{HOST}/actor", publicKeyPem=public_key_pem ))

4. Server Initialization

We initialize the ActivityPubServer from apkit, providing it with a function to retrieve our Actor's keys for signing outgoing activities.

# [main.py](http://main.py) (continued)# --- Key Retrieval Function ---async def get_keys_for_actor(identifier: str) -> list[ActorKey]: """Returns the key for a given 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 # Register the key retrieval function))

5. In-Memory Storage and Cache

To serve created activities, we need to store them somewhere. For simplicity, this example uses a basic in-memory dictionary as a store and a cache. In a production application, you would replace this with a persistent database (like SQLite or PostgreSQL) and a proper cache (like Redis).

# [main.py](http://main.py) (continued)# --- In-memory Store and Cache ---ACTIVITY_STORE = {} # A simple dict to store created activitiesCACHE = {} # A cache for recently accessed activitiesCACHE_TTL = timedelta(minutes=5) # Cache expiration time (5 minutes)

6. Reminder Parsing and Sending Logic

This is the core logic of our bot. The parse_reminder function uses a regular expression to extract the delay and message from a mention, and send_reminder schedules the notification.

# [main.py](http://main.py) (continued)# --- Reminder Parsing Logic ---def parse_reminder(text: str) -> tuple[timedelta | None, str | None, str | None]: """Parses reminder text like '5m do something'.""" # ... (implementation omitted for brevity)# --- Reminder Sending Function ---async def send_reminder(ctx: Context, delay: timedelta, message: str, target_actor: APKitActor, original_note: Note): """Waits for a specified delay and then sends a reminder.""" logger.info(f"Scheduling reminder for {target_actor.id} in {delay}: '{message}'") await asyncio.sleep(delay.total_seconds()) # Asynchronously wait logger.info(f"Sending reminder to {target_actor.id}") # Create the reminder Note reminder_note = Note(...) # Wrap it in a Create activity reminder_create = Create(...) # Store the created activities ACTIVITY_STORE[reminder_note.id] = reminder_note ACTIVITY_STORE[reminder_create.id] = reminder_create # Send the activity to the target actor's inbox 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. Endpoint Definitions

We define the required ActivityPub endpoints. Since apkit is built on FastAPI, we can use standard FastAPI decorators. The main endpoints are:

• Webfinger: Allows users on other servers to discover the bot using an address like @user@host. This is a crucial first step for federation.
• /actor: Serves the bot's Actor object, which contains its profile information and public key.
• /inbox: The endpoint where the bot receives activities from other servers. apkit handles this route automatically, directing activities to the handlers we'll define in the next step.
• /outbox: A collection of the activities created by the bot. but this returns placeholder collection.
• /notes/{note_id} and /creates/{create_id}: Endpoints to serve specific objects created by the bot, allowing other servers to fetch them by their unique ID.

Here is the code for defining these endpoints:

# [main.py](http://main.py) (continued)# The inbox endpoint is handled by apkit automatically.app.inbox("/inbox") @/span app.webfinger()async def webfinger_endpoint(request: Request, acct: WebfingerResource) -> Response: """Handles Webfinger requests to make the bot discoverable.""" if not acct.url: # Handle resource queries like 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: # Handle resource queries using a 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) @/span app.get("/actor")async def get_actor_endpoint(): """Serves the bot's Actor object.""" return ActivityResponse(actor) @/span app.get("/outbox")async def get_outbox_endpoint(): """Serves a collection of the bot's sent activities.""" 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) @/span app.get("/notes/{note_id}")async def get_note_endpoint(note_id: uuid.UUID): """Serves a specific Note object, with caching.""" note_uri = f"https://{HOST}/notes/{note_id}" # Check cache first if note_uri in CACHE and (datetime.now() - CACHE[note_uri]["timestamp"]) < CACHE_TTL: return ActivityResponse(CACHE[note_uri]["activity"]) # If not in cache, get from store if note_uri in ACTIVITY_STORE: activity = ACTIVITY_STORE[note_uri] # Add to cache before returning CACHE[note_uri] = {"activity": activity, "timestamp": datetime.now()} return ActivityResponse(activity) return Response(status_code=404) # Not Found @/span app.get("/creates/{create_id}")async def get_create_endpoint(create_id: uuid.UUID): """Serves a specific Create activity, with caching.""" 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. Activity Handlers

We use the @app.on() decorator to define handlers for specific activity types posted to our inbox.

• on_follow_activity: Automatically accepts Follow requests.
• on_create_activity: Parses incoming Create activities (specifically for Note objects) to schedule reminders.

# [main.py](http://main.py) (continued)# Handler for Follow activities @/span app.on(Follow)async def on_follow_activity(ctx: Context): """Automatically accepts follow requests.""" # ... (implementation omitted for brevity)# Handler for Create activities @/span app.on(Create)async def on_create_activity(ctx: Context): """Parses mentions to schedule reminders.""" activity = ctx.activity # Ignore if it's not a Note if not (isinstance(activity, Create) and isinstance(activity.object, Note)): return Response(status_code=202) note = activity.object # Check if the bot was mentioned 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) # ... (Parse reminder text) delay, message, time_str = parse_reminder(command_text) # If parsing is successful, schedule the reminder as a background task 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: # If parsing fails, send usage instructions reply_content = "<p>🤔 Sorry, I didn\'t understand. Please use the format: `@reminder [time] [message]`.</p><p>Example: `@reminder 10m Check the oven`</p>" # ... (Create and send the reply Note)

9. Running the Application

Finally, we run the application using uvicorn.

# [main.py](http://main.py) (continued)if __name__ == "__main__": import uvicorn logger.info("Starting uvicorn server...") uvicorn.run(app, host="0.0.0.0", port=8000)

How to Run the Bot

1. Set the HOST and USER_ID variables in main.py to match your environment.

2. Run the server from your terminal:

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

3. Your bot will be running at http://0.0.0.0:8000.

Now you can mention your bot from anywhere in the Fediverse (e.g., @reminder@your.host.com) to set a reminder.

Next Steps

This tutorial covers the basics of creating a simple ActivityPub bot. Since it only uses in-memory storage, all reminders will be lost on server restart. Here are some potential improvements:

• Persistent Storage: Replace the in-memory ACTIVITY_STORE with a database like SQLite or PostgreSQL.
• Robust Task Queuing: Use a dedicated task queue like Celery with a Redis or RabbitMQ broker to ensure reminders are not lost if the server restarts.
• Advanced Commands: Add support for more complex commands, such as recurring reminders.

We hope this guide serves as a good starting point for building your own ActivityPub applications!

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

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

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

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

例えば、ユーザーが「@reminder@your.host.com 10m check the oven」というメッセージでボットにメンションすると、ボットは10分後に「🔔 Reminder for @/span 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](http://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](http://main.py)import asyncioimport loggingimport reimport uuidimport osfrom datetime import timedelta, datetime# Imports from FastAPI, cryptography, and apkitfrom fastapi import Request, Responsefrom fastapi.responses import JSONResponsefrom cryptography.hazmat.primitives.asymmetric import rsafrom cryptography.hazmat.primitives import serialization as crypto_serializationfrom apkit.config import AppConfigfrom apkit.server import ActivityPubServerfrom apkit.server.types import Context, ActorKeyfrom apkit.server.responses import ActivityResponsefrom apkit.models import ( Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,)from apkit.client import WebfingerResource, WebfingerResult, WebfingerLinkfrom 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 domainUSER_ID = "reminder" # The bot's username

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

2. 鍵の生成と永続化

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

# [main.py](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](http://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](http://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](http://main.py) (続き)# --- In-memory Store and Cache ---ACTIVITY_STORE = {} # 作成されたアクティビティを保存するシンプルな辞書CACHE = {} # 最近アクセスされたアクティビティのキャッシュCACHE_TTL = timedelta(minutes=5) # キャッシュの有効期限（5分）

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

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

# [main.py](http://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](http://main.py) (続き)# インボックスエンドポイントはapkitによって自動的に処理されます。app.inbox("/inbox") @/span 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) @/span app.get("/actor")async def get_actor_endpoint(): """ボットのActorオブジェクトを提供します。""" return ActivityResponse(actor) @/span 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) @/span 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 @/span 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](http://main.py) (続き)# Followアクティビティのハンドラー @/span app.on(Follow)async def on_follow_activity(ctx: Context): """フォローリクエストを自動的に受け入れます。""" # ... (実装は簡潔にするため省略)# Createアクティビティのハンドラー @/span 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を作成して送信)