Skip to content

Announcing Hashtags from Mastodon

In this tutorial, we will build an application that listens on the Mastodon streaming API <https://docs.joinmastodon.org/methods/streaming/>_ for statuses as Mastodon calls them, and then announces them. This can be considered more of an ecosystem tutorial then one for bovine. We will spend more time on Mastodon and discussing how to address the Announce activity than actually using bovine.

Mastodon API

Our first task is to create a Mastodon API token. For this log into your Mastodon account in your browser. Then

  1. Open your profile
  2. Select development (bottom left)
  3. Select new application (top right)
  4. Give it any name e.g. “Bovine Tutorial”
  5. Give it the “push, receive your push notifications” permission. Also keep the read scope.
  6. Submit
  7. The application name should appear in the list, open the linked
  8. Copy the string under “Your access token”

This access token will be used in an Authorization header as follows

Authorization: Bearer <Your access token>

We are now ready to test this API. First, we can use the BearerAuthClient provided by bovine to check the credentials (this needs the read scope) with

>>> import aiohttp
>>> from bovine.clients.bearer import BearerAuthClient
>>> session = aiohttp.ClientSession()

>>> url = f"https://{your_mastodon_domain}/api/v1/accounts/verify_credentials"
>>> client = BearerAuthClient(session, YOUR_TOKEN)
>>> response = await client.get(url)
>>> await response.json()

This should now display a JSON containing a bunch of stuff about your account. You should at least recognize your account name.

Next step is streaming the public timeline via:

>>> url = f"https://{your_mastodon_domain}/api/v1/streaming/public"
>>> async for entry in client.event_source(url):
>>>     print(entry)

At least in my case using a populous Mastodon server this leads to rapidly scrolling by lines of json. By looking at the json with sharp eyes, one recognizes that these are not ActivityPub objects, but some Mastodon interna. The only property that interests us is uri, which represents the object id of the corresponding ActivityPub object. This means we could simplify our code to:

>>> import json
>>> url = f"https://{your_mastodon_domain}/api/v1/streaming/public"
>>> async for entry in client.event_source(url):
>>>     if entry.event == "update":
>>>          try:
>>>              print(json.loads(entry.data)["uri"])
>>>          except:
>>>              pass

We filter here for events of type “update” as these are the only ones we care about. The try block is necessary as not all messages currently get reassembled correctly.

These are the features of Mastodon, we will be using. If you are interested in exploring more, you can take a look at Mastodon's documentation <https://docs.joinmastodon.org/>_.

Implementing Announcing

We start with introducing the skeleton of the application. For this, we assume that you stored the credentials necessary to grant access to BovineClient in bot.toml. The skeleton of an application would then look like

import asyncio
import bovine
from bovine.clients.bearer import BearerAuthClient

async def loop_over_mastodon_events(client):
    mastodon = BearerAuthClient(client.session, YOUR_TOKEN)
    ...

async def main():
    async with bovine.BovineClient.from_file("bot.toml") as client:
        await loop_over_mastodon_events(client)

asyncio.run(main())

This script doesn’t do anything yet, except possibly looking up your actor, through initializing BovineClient. We also note that, we reused the aiohttp.ClientSession that was initialized with BovineClient.

With our knowledge of the previous section, we are ready to complete the function via

import json

async def send_announce(client, object_id):
    ...

async def loop_over_mastodon_events(client):
    mastodon = BearerAuthClient(client.session, YOUR_TOKEN)
    url = f"https://{your_mastodon_domain}/api/v1/streaming/hashtag?tag={interesting_tag}"
    async for entry in mastodon.event_source(url):
        if entry.event == "update":
            try:
                object_id = json.loads(entry.data)["uri"]
                await send_announce(client, object_id)
            except:
                pass

Except for using a different API endpoint and needing to choose an interesting tag to watch, nothing much has changed. While developing, I can recommend using “catsofmastodon”, as it sees a lot of activity.

Finally, we can build the announce and send it

async def send_announce(client, object_id):
    announce = client.activity_factory.announce(object_id).as_public().build()

    await client.send_to_outbox(announce)

This might not seem like a lot of code, but the magic is hidden. The choice of addressing this activity to both the public collection and the followers collection is so that the announce shows up on Mastodon. Without adding the public collection, the announce is not visible on Mastodon.