I want to write a program to extract a list of clickable links from a PDF page.

#pypdf can list the link positions/sizes and target URLs. But in a PDF document, links are annotations, which are separate data from the document text.

To get the display text of a clickable link in a PDF, is the easiest way to convert the full page to PNG, crop it to the link's bounding box, and run that through OCR? Or am I missing something more reasonable?

#programming #python #IfItWorksItWorks

You can still help sponsor the Brazilian national #Python conference!
You can also donate a ticket so that a student can attend!
#PythonBrasil2025

I am working on a binary tree class for one of my algorithms class. I have gone a little off the rails and implemented generators for the different tree traversal techniques.

One thing I don't like though is all the redundancy. All of these generators are essentially the same but just the order in which nodes and trees are evaluated is changed. Is there anyway to simplify the code or reduce the redundancy?

#python #programming

You can still help sponsor the Brazilian national #Python conference!
You can also donate a ticket so that a student can attend!
#PythonBrasil2025

Good morning, especially to the two new and two returning @ThePSF directors!

And _especially_ good morning to the @ThePSF's moderation team. Y'all do great work and we don't thank you enough (nor could we)!

#Python

Good morning, especially to the two new and two returning @ThePSF directors!

And _especially_ good morning to the @ThePSF's moderation team. Y'all do great work and we don't thank you enough (nor could we)!

#Python

🐍🏃‍➡️ We kicked off the first day of the #Python #CoreSprint at #Arm in #Cambridge with lots of talks and lots of discussion about talks!

Ken Jin - Building a JIT Community AND Demo [effect] of new C API
@antocuni - Tracing JITs on real code
@brettcannon - WASI update AND precompiled binaries from python.org
Hood Chatham - Upstreaming Pyodide FFI
@freakboy3742 - Managing cross-platform wheel builds
Steering Council - PEP 793 and abi3/abi3t/abi4
Matthew Parkinson - Designing Deep Immutability

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 uv
uv init

# Install dependencies
uv 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:

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

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

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 one
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()
        ))

# Generate the public key from the private key
public_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 activities
CACHE = {}          # A cache for recently accessed activities
CACHE_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_follow_activity: Automatically accepts Follow requests.
• 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)

9. Running the Application

Finally, we run the application using uvicorn.

# 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

🐍🏃‍➡️ We kicked off the first day of the #Python #CoreSprint at #Arm in #Cambridge with lots of talks and lots of discussion about talks!

Ken Jin - Building a JIT Community AND Demo [effect] of new C API
@antocuni - Tracing JITs on real code
@brettcannon - WASI update AND precompiled binaries from python.org
Hood Chatham - Upstreaming Pyodide FFI
@freakboy3742 - Managing cross-platform wheel builds
Steering Council - PEP 793 and abi3/abi3t/abi4
Matthew Parkinson - Designing Deep Immutability

As inscrições para os tutoriais da Python Brasil 2025 abrem dia 22 de setembro, às 8h da manhã 🐍 🔥 E serão realizadas pelo Eventbrite (mesmo link dos ingressos de palestras)

Tutoriais são atividades gratuitas, e você pode se inscrever em mais de um tutorial (desde que os horários não sejam conflitantes).

Confira a programação 📅 e planeje com calma em quais dessas atividades incríveis você quer participar ♥️

https://talks.python.org.br/pythonbrasil-2025/schedule/

#python #pythonbrasil #bolhadev

Although I've used python's matplotlib for years, it has always been a case of hacking together code snippets from the web.

Why? Because I don't think there is a unified, coherent, consistent design philosophy. If there is, I have never found it in all these years.

I keep reading that the Grammar of Graphics is the solution - an actually designed framework which has been implemented first and foremost in R, but python implementations exist too.

What do people think?

I'd welcome your thoughts before I invest time in learning some grammar of graphics and maybe trying a library like plotnine.

Are my expectations of Grammar of Graphics too high?

cc @nrennie

#python #R #ggplot2 #grammarofgraphics #visualisation #dataviz

Although I've used python's matplotlib for years, it has always been a case of hacking together code snippets from the web.

Why? Because I don't think there is a unified, coherent, consistent design philosophy. If there is, I have never found it in all these years.

I keep reading that the Grammar of Graphics is the solution - an actually designed framework which has been implemented first and foremost in R, but python implementations exist too.

What do people think?

I'd welcome your thoughts before I invest time in learning some grammar of graphics and maybe trying a library like plotnine.

Are my expectations of Grammar of Graphics too high?

cc @nrennie

#python #R #ggplot2 #grammarofgraphics #visualisation #dataviz

#introduction

Hi all, my name is Stuart but you can call me Stu.

I am a big fan of all sorts of #technology and also a very huge fan of #music of all types of different genres.

I love playing #golf and I love watching movies and TV shows.

#openweb #opensource #linux #python #golf #tvvseries #music

As inscrições para os tutoriais da Python Brasil 2025 abrem dia 22 de setembro, às 8h da manhã 🐍 🔥 E serão realizadas pelo Eventbrite (mesmo link dos ingressos de palestras)

Tutoriais são atividades gratuitas, e você pode se inscrever em mais de um tutorial (desde que os horários não sejam conflitantes).

Confira a programação 📅 e planeje com calma em quais dessas atividades incríveis você quer participar ♥️

https://talks.python.org.br/pythonbrasil-2025/schedule/

#python #pythonbrasil #bolhadev

Days between today and any date with #Python #69

#!/usr/bin/python3
import sys; from datetime import datetime,date
formats=["%m/%d/%Y","%m/%d/%y","%Y-%m-%d","%d-%m-%Y","%d %b %Y","%d %B %Y"]
for f in formats:
try: d=datetime.strptime(sys.argv[1],f).date(); break
except: pass
else: exit("bad format")
print(abs((date.today()-d).days))

Days between today and any date with #Python #69

#!/usr/bin/python3
import sys; from datetime import datetime,date
formats=["%m/%d/%Y","%m/%d/%y","%Y-%m-%d","%d-%m-%Y","%d %b %Y","%d %B %Y"]
for f in formats:
try: d=datetime.strptime(sys.argv[1],f).date(); break
except: pass
else: exit("bad format")
print(abs((date.today()-d).days))

Buenas pessoal!

Estão sabendo que eu serei Keynote da PySul esse ano?

Fiquei super feliz com esse convite, gostaria de convidar a todes para participar desse evento que será lindo como sempre.

Quem eu encontro lá ein (no evento ou no bar)??

Pega teu mate e te aprochega!
📅 21 a 23 de Novembro de 2025
📍 Porto Alegre - RS

https://sul.python.org.br/keynote-bruna-ferreira-bug-esleif-python-sul-2025/

#python #tech #development #coding

Buenas pessoal!

Estão sabendo que eu serei Keynote da PySul esse ano?

Fiquei super feliz com esse convite, gostaria de convidar a todes para participar desse evento que será lindo como sempre.

Quem eu encontro lá ein (no evento ou no bar)??

Pega teu mate e te aprochega!
📅 21 a 23 de Novembro de 2025
📍 Porto Alegre - RS

https://sul.python.org.br/keynote-bruna-ferreira-bug-esleif-python-sul-2025/

#python #tech #development #coding

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 uv
uv init

# Install dependencies
uv 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:

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

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

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 one
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()
        ))

# Generate the public key from the private key
public_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 activities
CACHE = {}          # A cache for recently accessed activities
CACHE_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_follow_activity: Automatically accepts Follow requests.
• 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)

9. Running the Application

Finally, we run the application using uvicorn.

# 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

Should I make more T-shirts?

https://villares-shop.fourthwall.com/

https://umapenca.com/villares/

#Python #Processing #py5 #CreativeCoding #FLOSS #numpy #shapely #trimesh

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 uv
uv init

# Install dependencies
uv 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:

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

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

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 one
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()
        ))

# Generate the public key from the private key
public_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 activities
CACHE = {}          # A cache for recently accessed activities
CACHE_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_follow_activity: Automatically accepts Follow requests.
• 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)

9. Running the Application

Finally, we run the application using uvicorn.

# 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

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 uv
uv init

# Install dependencies
uv 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:

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

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

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 one
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()
        ))

# Generate the public key from the private key
public_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 activities
CACHE = {}          # A cache for recently accessed activities
CACHE_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_follow_activity: Automatically accepts Follow requests.
• 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)

9. Running the Application

Finally, we run the application using uvicorn.

# 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

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 uv
uv init

# Install dependencies
uv 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:

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

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

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 one
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()
        ))

# Generate the public key from the private key
public_key_pem = private_key.public_key().public_bytes(
    encoding=crypto_serialization.Encoding.PEM,
    format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')