Hello, I'm an open source software engineer in my late 30s living in #Seoul, #Korea, and an avid advocate of #FLOSS and the #fediverse.
I'm the creator of @fedify, an #ActivityPub server framework in #TypeScript, @hollo, an ActivityPub-enabled microblogging software for single users, and @botkit, a simple ActivityPub bot framework.
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└── 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:
Imports and Configuration: Set up necessary imports and basic configuration variables.
Key Generation: Prepare the cryptographic keys needed for signing activities.
Actor Definition: Define the bot's identity on the Fediverse.
Server Initialization: Set up the apkit ActivityPub server.
Data Storage: Implement a simple in-memory store for created activities.
Reminder Logic: Code the core logic for parsing reminders and sending notifications.
Endpoint Definitions: Create the necessary web endpoints (/actor, /inbox, etc.).
Activity Handlers: Process incoming activities from other servers.
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.
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 (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 (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 (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 (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 (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 (continued)# The inbox endpoint is handled by apkit automatically.app.inbox("/inbox")@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)@app.get("/actor")async def get_actor_endpoint(): """Serves the bot's Actor object.""" return ActivityResponse(actor)@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)@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@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_create_activity: Parses incoming Create activities (specifically for Note objects) to schedule reminders.
# main.py (continued)# Handler for Follow activities@app.on(Follow)async def on_follow_activity(ctx: Context): """Automatically accepts follow requests.""" # ... (implementation omitted for brevity)# Handler for Create activities@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)
Set the HOST and USER_ID variables in main.py to match your environment.
Run the server from your terminal:
uvicorn main:app --host 0.0.0.0 --port 8000
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!
The update was hard; the jsonld.js library was choking on the FEP's context document because Codeberg Pages, where FEP context documents are hosted, is unreliable.
I ended up adding a feature to the `activitystrea.ms` NodeJS library to let you pre-cache a context document, so it didn't have to go fetch it at run time. While I was under the hood, I also fixed a couple of bugs that made it hard to work with extra context documents.
The bot server is now working; I'm going to add more contexts.
I updated the #ActivityPubBot server software to use FEP-5711, the inverse properties FEP. It is a way to declare two-way relations; if X has an `inbox` property with value Y, then Y has an `inboxOf` property with value X. It helps with navigation when you're using the ActivityPub API, and it also helps with spoofing (since both objects agree on the relationship).
We've just published an experimental pre-release version 1.9.0-pr.431.1597 that adds CommonJS support to all npm packages in the Fedify ecosystem! 🧪
What's new
This experimental build addresses one of the most requested features—better compatibility with CommonJS-based Node.js applications, especially NestJS projects. The pre-release eliminates the need for Node.js's --experimental-require-module flag and resolves dual package hazard issues.
Note: While we now support CommonJS for legacy project compatibility, we still recommend using ESM or migrating to Deno for the best experience with Fedify.
Who should test this?
NestJS developers using Fedify who need CommonJS compatibility
Legacy CommonJS-based Node.js projects that had trouble integrating Fedify
Anyone who previously needed experimental Node.js flags to use Fedify
How to test
Install the experimental pre-release version:
npm install @fedify/fedify@1.9.0-pr.431.1597# or for specific integrationsnpm install @fedify/nestjs@1.9.0-pr.431.1597
You should now be able to use standard require() syntax without any experimental flags.
What we're looking for
Does CommonJS import work correctly in your legacy project?
Are you able to remove the --experimental-require-module flag?
Any issues or regressions compared to the current stable version?
Your feedback on this experimental build is invaluable for ensuring this major compatibility improvement works smoothly before the official 1.9.0 release!
Whoa, as of 10 days ago we have functions to go between a Uint8Array and a Base64 Normal/URL Padded/Unpadded string in JavaScript without having to go via a String of “bytes” and manually juggle the padding in every major browser.
@thisismissem@mariusor@oranadoz@hongminhee +1, an image and a descriptor are different things and should be treated as different things. content negotiation is not a solution here -- the same information should be returned for the same resource (modulo whichever representation you ask for or receive).
@evan@hongminhee more and more i am thinking that Link was a bad idea from a data modeling perspective. "assume bare href instead of bare id" is something that can never make sense. if we really want to maintain validity of Link then it should *always* be embedded as an anonymous object:
일본의 기술 블로깅 플랫폼인 Zenn 첫 페이지에 내가 Optique에 관해 썼던 글인 「TypeScriptの型推論でCLIバリデーションをなくせた話」(TypeScript의 타입 추론으로 CLI 유효성 검사를 필요 없게 만든 이야기)가 떴다!
ALT text details일본 기술 블로깅 플랫폼 Zenn의 Tech 섹션 스크린샷. 페이지에는 여러 기술 관련 글들이 나열되어 있으며, 그 중 빨간 원으로 강조된 「TypeScriptの型推論でCLIバリデーションをなくせた話」(TypeScript의 타입 추론으로 CLI 유효성 검사를 필요 없게 만든 이야기) 제목의 글이 20개의 좋아요와 12개의 댓글을 받으며 주목받고 있다. 해당 글은 TypeScript와 CLI 파서 라이브러리인 Optique에 관한 내용을 다루고 있다.
@hongminhee It's a place where our loosey goosey style goes into nondeterminism. We should tighten it up in the next version. My main answer would be: publishers, don't do that.
First half is us talking about @anewsocial, and the second half is Sean letting me rave about all the exciting work being done on the open social web by @hongminhee, @cheeaun, @gabboman, @thisismissem, and so many others.