Simple client example #11
Description
It would be nice to have a simple example that uses apkit's client. Maybe even a collection of examples like for the (abandoned) pyfed library: https://dev.funkwhale.audio/funkwhale/pyfed/-/tree/main/examples
I tried to extend the client example from the guide to send a note to my Mastodon account. The server accepts the create activity and seems to store the note. But it does not appear in my notifications. So I guess something is still missing in the Note or the Create object.
import asyncio
import logging
import os
import uuid
from apkit.client.asyncio import ActivityPubClient
from apkit.models import Person, Note, CryptographicKey, Create
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization as crypto_serialization
from datetime import datetime, timezone
HOST="example.com"
USER_ID="demo"
TARGET_ID="https://example.net/users/alice"
# --- Logging Setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- Key Persistence ---
KEY_FILE = "private_key.pem"
if os.path.exists(KEY_FILE):
logger.info(f"Loading existing private key from {KEY_FILE}.")
with open(KEY_FILE, "rb") as f:
private_key = crypto_serialization.load_pem_private_key(f.read(), password=None)
else:
logger.info(f"No key file found. Generating new private key and saving to {KEY_FILE}.")
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
with open(KEY_FILE, "wb") as f:
f.write(private_key.private_bytes(
encoding=crypto_serialization.Encoding.PEM,
format=crypto_serialization.PrivateFormat.PKCS8,
encryption_algorithm=crypto_serialization.NoEncryption()
))
public_key_pem = private_key.public_key().public_bytes(
encoding=crypto_serialization.Encoding.PEM,
format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')
async def main():
async with ActivityPubClient() as client:
# Fetch a remote Actor
target_actor = await client.actor.fetch(TARGET_ID)
print(f"Fetched actor: {target_actor.name}")
# Get the inbox URL from the actor's profile
inbox_url = target_actor.inbox
if not inbox_url:
raise Exception("Could not find actor's inbox URL")
logger.info(f"Found actor's inbox: {inbox_url}")
# Create actor
actor = Person(
id=f"https://{HOST}/users/{USER_ID}",
name="apkit Demo",
preferredUsername="demo",
summary="This is a demo actor powered by apkit!",
inbox=f"https://{HOST}/users/{USER_ID}/inbox",
outbox=f"https://{HOST}/users/{USER_ID}/outbox",
publicKey=CryptographicKey(
id=f"https://{HOST}/users/{USER_ID}#main-key",
owner=f"https://{HOST}/users/{USER_ID}",
publicKeyPem=public_key_pem
)
)
# Create note
note = Note(
id=f"https://{HOST}/notes/{uuid.uuid4()}",
attributedTo=actor.id,
content=f"<p>Hello from apkit</p>",
published=datetime.utcnow().isoformat() + "Z",
to=[target_actor.id],
cc=["https://www.w3.org/ns/activitystreams#Public"],
)
# Create activity
create = Create(
id=f"https://{HOST}/creates/{uuid.uuid4()}",
actor=actor.id,
object=note.to_json(),
published=datetime.utcnow().isoformat() + "Z",
to=note.to,
cc=note.cc
)
print(create.to_json())
# Deliver the activity
logger.info("Delivering activity...")
resp = await client.post(
inbox_url,
key_id=actor.publicKey.id,
signature=private_key,
json=create
)
logger.info(f"Delivery result: {resp.status}")
if __name__ == "__main__":
asyncio.run(main())While sending the activity the server from the tutorial must be running to return the public key of the sending actor when the receiver wants to verify the signature.