Creating a Bluesky Bot with Python and AWS Lambda

(Updated 3/6/2024 to reflect a change to how blobs are handled as post embeds.)

I have already written a post detailing my process for building a Twitter bot and a Mastodon bot, but now that Twitter has killed my Twitter bot and instituted a temporary(?) rate limit for reading tweets I have seen a lot of interest in Bluesky. Just like Mastodon, it's a federated social network but is built on the AT Protocol instead of ActivityPub.

I won't be going over the screenshot capture process in this post since I've covered that previously, but instead focus on how I've built my Python script to send posts and images to Bluesky. First, here is the link to the full unmodified script I am running.

Breaking it down, here are the functions within the script:

def get_app_password():
    ssm = boto3.client("ssm")
    app_password = ssm.get_parameter(Name="bsky_koth_app_password", WithDecryption=True)
    app_password = app_password["Parameter"]["Value"]
    return app_password

When you open your account settings, there is a section called App Passwords. Click on this and click "Create App Password".

You'll be presented with the app password once, so make sure you save it. If you lose it you can always create another.

I've saved this string as a SecureString within AWS Systems Manager Parameter Store as a way to retrieve it securely within Lambda.

The function above is retrieving the app password from parameter store by name.

def get_did():
    http = urllib3.PoolManager()
    HANDLE = "kothscreens.bsky.social"
    DID_URL = "https://bsky.social/xrpc/com.atproto.identity.resolveHandle"
    did_resolve = http.request("GET", DID_URL, fields={"handle": HANDLE})
    did_resolve = json.loads(did_resolve.data)
    did = did_resolve["did"]
    return did

The AT Protocol handles user identification by assigning persistent "DID"s to users that enables them to migrate to another federated host on the AT Protocol while maintaining that link to their identity. This function obtains the DID for our account.

def get_api_key(did, app_password):
    http = urllib3.PoolManager()
    API_KEY_URL = "https://bsky.social/xrpc/com.atproto.server.createSession"
    post_data = {"identifier": did, "password": app_password}
    headers = {"Content-Type": "application/json"}
    api_key = http.request(
        "POST",
        API_KEY_URL,
        headers=headers,
        body=bytes(json.dumps(post_data), encoding="utf-8"),
    )
    api_key = json.loads(api_key.data)
    return api_key["accessJwt"]

In order to perform any action as an authenticated user, we need to use our public DID and our private app password to obtain an API key.

I won't go into details about obtaining the image from S3 since I've done so on my previous posts, but the code is also contained in the gist. Just as before, we are saving the image to /tmp/local.jpg within the Lambda container.

def upload_blob(download_path, key):
    http = urllib3.PoolManager()
    upload_blob_url = "https://bsky.social/xrpc/com.atproto.repo.uploadBlob"
    with open(download_path, "rb") as img:
        jpeg_bin = img.read()
    blob_request = http.request(
        "POST",
        upload_blob_url,
        body=jpeg_bin,
        headers={"Content-Type": "image/jpeg", "Authorization": f"Bearer {key}"},
    )
    blob_request = json.loads(blob_request.data)
    print(f"blob request response: {blob_request}")
    blob = blob_request["blob"]
    return blob

Before we can post our image on Bluesky, we have to first upload the image as a blob to be added in a later request. This will return the blog object we can reference in our post.

def post_skeet(did, blob, key, season, episode):
    http = urllib3.PoolManager()
    now = datetime.today()
    post_feed_url = "https://bsky.social/xrpc/com.atproto.repo.createRecord"
    post_record = {
        "collection": "app.bsky.feed.post",
        "repo": did,
        "record": {
            "text": "",
            "createdAt": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "embed": {
                "$type": "app.bsky.embed.images",
                "images": [
                    {
                        "image": blob,
                        "alt": f"Screenshot from King of the Hill Season {season}, Episode {episode}",
                    }
                ],
            },
        },
    }
    post_request = http.request(
        "POST",
        post_feed_url,
        body=json.dumps(post_record),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"},
    )
    post_request = json.loads(post_request.data)
    print(post_request)
    return post_request

The Bluesky community has affectionately termed posts as "skeets" so I'm following suit here out of respect. To post our image we need our DID, the image blob CID, our API key, and in my case I'm adding some text for the season and episode numbers.

Putting it all together:

def lambda_handler(event, context):
    app_password = get_app_password()
    did = get_did()
    key = get_api_key(did, app_password)
    download = download_frame()
    download_path = download["download_path"]
    season = download["randomSeason"]
    episode = download["randomEpisode"]
    blob = upload_blob(download_path, key)
    response = post_skeet(did, blob, key, season, episode)
    return response

You can follow my new Bluesky bot here and my personal account here. Let me know what you come up with!

Also, I want to thank Felicitas Pojtinger whose gist here got me jumpstarted!