pubby 0.1.1

A general-purpose library to add ActivityPub federation support to your website

A general-purpose Python library to add ActivityPub federation support to your website.

What is ActivityPub?

ActivityPub is a W3C standard for decentralized social networking. Servers exchange JSON-LD activities (posts, likes, follows, boosts) over HTTP, enabling federation across platforms like Mastodon, Pleroma, Misskey, and others. It's the protocol that powers the Fediverse.

What is Pubby?

Pubby is a framework-agnostic library that handles the ActivityPub plumbing so you can focus on your app:

• Inbox processing — receive and dispatch Follow, Like, Announce, Create, Update, Delete activities
• Outbox delivery — concurrent fan-out to follower inboxes with retry and shared-inbox deduplication
• HTTP Signatures — sign outgoing requests and verify incoming ones (draft-cavage, using cryptography directly — no httpsig dependency)
• Discovery — WebFinger and NodeInfo 2.1 endpoints
• Interaction storage — followers, interactions, activities, actor cache
• Framework adapters — Flask, FastAPI, Tornado
• Storage adapters — SQLAlchemy (any supported database) and file-based JSON

Installation

Base install:

pip install pubby

With extras:

pip install "pubby[db,flask]"        # SQLAlchemy + Flask
pip install "pubby[db,fastapi]"      # SQLAlchemy + FastAPI
pip install "pubby[db,tornado]"      # SQLAlchemy + Tornado

Available extras: db, flask, fastapi, tornado.

Quick Start

Flask

pip install "pubby[db,flask]"

from flask import Flask
from pubby import ActivityPubHandler
from pubby.crypto import generate_rsa_keypair, export_private_key_pem
from pubby.storage.adapters.db import init_db_storage
from pubby.server.adapters.flask import bind_activitypub

app = Flask(__name__)
storage = init_db_storage("sqlite:////tmp/pubby.db")

# Generate a keypair (persist this — don't regenerate on restart!)
private_key, _ = generate_rsa_keypair()

handler = ActivityPubHandler(
    storage=storage,
    actor_config={
        "base_url": "https://example.com",
        "username": "blog",
        "name": "My Blog",
        "summary": "A blog with ActivityPub support",
    },
    private_key=private_key,
)

bind_activitypub(app, handler)
app.run()

FastAPI

pip install "pubby[db,fastapi]"

from fastapi import FastAPI
from pubby import ActivityPubHandler
from pubby.crypto import generate_rsa_keypair
from pubby.storage.adapters.db import init_db_storage
from pubby.server.adapters.fastapi import bind_activitypub

app = FastAPI()
storage = init_db_storage("sqlite:////tmp/pubby.db")
private_key, _ = generate_rsa_keypair()

handler = ActivityPubHandler(
    storage=storage,
    actor_config={
        "base_url": "https://example.com",
        "username": "blog",
        "name": "My Blog",
        "summary": "A blog with ActivityPub support",
    },
    private_key=private_key,
)

bind_activitypub(app, handler)

Tornado

pip install "pubby[db,tornado]"

from tornado.web import Application
from tornado.ioloop import IOLoop
from pubby import ActivityPubHandler
from pubby.crypto import generate_rsa_keypair
from pubby.storage.adapters.db import init_db_storage
from pubby.server.adapters.tornado import bind_activitypub

app = Application()
storage = init_db_storage("sqlite:////tmp/pubby.db")
private_key, _ = generate_rsa_keypair()

handler = ActivityPubHandler(
    storage=storage,
    actor_config={
        "base_url": "https://example.com",
        "username": "blog",
        "name": "My Blog",
        "summary": "A blog with ActivityPub support",
    },
    private_key=private_key,
)

bind_activitypub(app, handler)
app.listen(8000)
IOLoop.current().start()

Registered Routes

All adapters register the same endpoints:

Method	Path	Description
GET	/.well-known/webfinger	WebFinger discovery
GET	/.well-known/nodeinfo	NodeInfo discovery
GET	/nodeinfo/2.1	NodeInfo 2.1 document
GET	/ap/actor	Actor profile (JSON-LD)
POST	/ap/inbox	Receive activities
GET	/ap/outbox	Outbox collection
GET	/ap/followers	Followers collection
GET	/ap/following	Following collection

The /ap prefix is configurable via the prefix parameter on bind_activitypub.

Publishing Content

Publish an article to all followers:

from pubby import Object

article = Object(
    id="https://example.com/posts/hello-world",
    type="Article",
    name="Hello World",
    content="<p>My first federated post!</p>",
    url="https://example.com/posts/hello-world",
    attributed_to="https://example.com/ap/actor",
)

handler.publish_object(article)

To update or delete:

# Update
handler.publish_object(updated_article, activity_type="Update")

# Delete
handler.publish_object(deleted_article, activity_type="Delete")

Delivery is concurrent (configurable via max_delivery_workers, default 10) with automatic retry and exponential backoff on failure.

Key Management

Important: your RSA keypair is your server's identity. Persist it — if you regenerate it, other servers won't be able to verify your signatures.

from pubby.crypto import (
    generate_rsa_keypair,
    export_private_key_pem,
    load_private_key,
)

# Generate once and save
private_key, public_key = generate_rsa_keypair()
pem = export_private_key_pem(private_key)

with open("/path/to/private_key.pem", "w") as f:
    f.write(pem)

# Load on startup
handler = ActivityPubHandler(
    storage=storage,
    actor_config={...},
    private_key_path="/path/to/private_key.pem",
)

Custom Storage

If you don't want to use SQLAlchemy or the file-based adapter, extend ActivityPubStorage:

from pubby import ActivityPubStorage, Follower, Interaction

class MyStorage(ActivityPubStorage):
    def store_follower(self, follower: Follower):
        ...

    def remove_follower(self, actor_id: str):
        ...

    def get_followers(self) -> list[Follower]:
        ...

    def store_interaction(self, interaction: Interaction):
        ...

    def delete_interaction(self, source_actor_id: str, target_resource: str, interaction_type: str):
        ...

    def get_interactions(self, target_resource: str | None = None, interaction_type: str | None = None) -> list[Interaction]:
        ...

    def store_activity(self, activity_id: str, activity_data: dict):
        ...

    def get_activities(self, limit: int = 20, offset: int = 0) -> list[dict]:
        ...

    def cache_remote_actor(self, actor_id: str, actor_data: dict):
        ...

    def get_cached_actor(self, actor_id: str, max_age_seconds: int = 86400) -> dict | None:
        ...

handler = ActivityPubHandler(
    storage=MyStorage(),
    actor_config={...},
    private_key=private_key,
)

File-based Storage

For apps that don't need a database (e.g. static-site generators):

from pubby.storage.adapters.file import FileActivityPubStorage

storage = FileActivityPubStorage(data_dir="/var/lib/myapp/activitypub")

Data is stored as JSON files in a structured directory layout, with thread-safe access via RLock per resource.

Configuration Reference

ActivityPubHandler Parameters

Parameter	Type	Default	Description
storage	ActivityPubStorage	required	Storage backend
actor_config	dict	required	Actor configuration (see below)
private_key	key / str / bytes	—	RSA private key
private_key_path	str / Path	—	Path to PEM private key file
on_interaction_received	Callable	None	Callback on new interaction
webfinger_domain	str	from base_url	Domain for acct: URIs
user_agent	str	"pubby/0.0.1"	Outgoing User-Agent
http_timeout	float	15.0	HTTP request timeout (seconds)
max_retries	int	3	Delivery retry attempts
max_delivery_workers	int	10	Concurrent delivery threads
software_name	str	"pubby"	NodeInfo software name
software_version	str	"0.0.1"	NodeInfo software version

actor_config Keys

Key	Type	Default	Description
base_url	str	required	Public base URL of your site
username	str	"blog"	Actor username (WebFinger handle)
name	str	username	Display name
summary	str	""	Actor bio/description
icon_url	str	""	Avatar URL
actor_path	str	"/ap/actor"	Path to the actor endpoint
type	str	"Person"	ActivityPub actor type
manually_approves_followers	bool	False	Require follow approval

Rendering Interactions

Pubby includes a Jinja2-based renderer for displaying interactions (replies, likes, boosts) on your pages:

from pubby import InteractionType

interactions = handler.storage.get_interactions(
    target_resource="https://example.com/posts/hello-world"
)

html = handler.render_interactions(interactions)

Then in your template:

<article>
  <h1>Hello World</h1>
  <p>My first federated post!</p>
</article>

<section class="interactions">
  {{ interactions_html }}
</section>

render_interactions returns a safe Markup object with theme-aware styling. You can also pass a custom Jinja2 template.

Rate Limiting

Protect your inbox with the built-in per-IP sliding window rate limiter:

from pubby import RateLimiter
from pubby.server.adapters.flask import bind_activitypub

rate_limiter = RateLimiter(max_requests=100, window_seconds=60)
bind_activitypub(app, handler, rate_limiter=rate_limiter)

Interaction Callbacks

Get notified when interactions arrive:

from pubby import Interaction

def on_interaction(interaction: Interaction):
    print(f"New {interaction.interaction_type}: {interaction.source_actor_id}")

handler = ActivityPubHandler(
    storage=storage,
    actor_config={...},
    private_key=private_key,
    on_interaction_received=on_interaction,
)

Tests

pip install -e ".[test]"
pytest tests

Development

pip install -e ".[dev]"
pre-commit install
pre-commit run --all-files

License

AGPL-3.0-or-later