Getting Started with Minimal-ActivityPub

This guide will help you quickly get started with minimal-activitypub, a Python library for interacting with ActivityPub servers like Mastodon, Pleroma, and Takahe.

Installation

Using pip

pip install minimal-activitypub

Using uv

uv add minimal-activitypub

Using pipx (for CLI tools)

pipx install minimal-activitypub

Prerequisites

  • Python 3.10 or higher
  • An account on an ActivityPub instance (Mastodon, Pleroma, Takahe, etc.)
  • Basic understanding of async/await in Python

Quick Example

Here's a minimal example to get you started:

import asyncio
from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub

async def main():
    # Create an HTTP client
    async with AsyncClient() as client:
        # Initialize the ActivityPub client
        ap = ActivityPub(
            instance="https://mastodon.social",
            client=client,
            access_token="your_access_token_here"
        )

        # Post a simple status
        status = await ap.post_status("Hello from minimal-activitypub! 🐘")
        print(f"Posted: {status['url']}")

# Run the async function
asyncio.run(main())

Step-by-Step Guide

1. Obtain an Access Token

Before you can use the library, you need an access token from your instance. There are two ways to get one:

Method A: Using Username and Password (Simpler)

from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub

async def get_token_with_password(instance_url, username, password):
    async with AsyncClient() as client:
        # Create an app on the instance
        client_id, client_secret = await ActivityPub.create_app(
            instance_url=instance_url,
            client=client
        )

        # Get access token
        token = await ActivityPub.get_auth_token(
            instance_url=instance_url,
            username=username,
            password=password,
            client=client
        )
        return token

Method B: Using OAuth Flow (More Secure)

from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub
from minimal_activitypub import USER_AGENT

async def get_token_with_oauth(instance_url):
    async with AsyncClient() as client:
        # Create an app
        client_id, client_secret = await ActivityPub.create_app(
            instance_url=instance_url,
            client=client
        )

        # Get authorization URL
        auth_url = await ActivityPub.generate_authorization_url(
            instance_url=instance_url,
            client_id=client_id,
            user_agent=USER_AGENT
        )

        print(f"Visit this URL to authorize: {auth_url}")
        auth_code = input("Enter the authorization code: ")

        # Get access token
        token = await ActivityPub.validate_authorization_code(
            client=client,
            instance_url=instance_url,
            authorization_code=auth_code,
            client_id=client_id,
            client_secret=client_secret
        )
        return token

2. Initialize the Client

Once you have an access token, initialize the ActivityPub client:

from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub

async def create_client(instance_url, access_token):
    client = AsyncClient(http2=True)  # HTTP/2 for better performance
    ap = ActivityPub(
        instance=instance_url,
        client=client,
        access_token=access_token,
        timeout=120  # 2 minute timeout
    )

    # Verify credentials and get instance info
    await ap.determine_instance_type()
    user_info = await ap.verify_credentials()
    print(f"Authenticated as: {user_info['username']}")

    return ap

3. Basic Operations

Posting a Status

async def post_example(ap):
    # Simple text status
    status1 = await ap.post_status("Hello, Fediverse!")

    # Status with visibility setting
    from minimal_activitypub import Visibility
    status2 = await ap.post_status(
        "This is unlisted",
        visibility=Visibility.UNLISTED
    )

    # Status with content warning
    status3 = await ap.post_status(
        "Spoiler content here",
        spoiler_text="Content warning",
        sensitive=True
    )

    return [status1, status2, status3]

Reading Timelines

async def read_timelines(ap):
    # Get public timeline (20 most recent posts)
    public = await ap.get_public_timeline(limit=20)
    print(f"Public timeline: {len(public)} posts")

    # Get home timeline (posts from people you follow)
    home = await ap.get_home_timeline(limit=10)
    print(f"Home timeline: {len(home)} posts")

    # Get hashtag timeline
    python_posts = await ap.get_hashtag_timeline("python", limit=15)
    print(f"#python posts: {len(python_posts)}")

    return public, home, python_posts

Working with Media

import aiofiles
import magic

async def post_with_image(ap, image_path, caption):
    # Determine MIME type
    mime_type = magic.from_file(image_path, mime=True)

    # Upload the image
    async with aiofiles.open(image_path, "rb") as f:
        media = await ap.post_media(
            file=f,
            mime_type=mime_type,
            description="A beautiful sunset"  # Alt text for accessibility
        )

    # Post status with the image
    status = await ap.post_status(
        status=caption,
        media_ids=[media["id"]]
    )

    return status

4. Complete Working Example

Here's a complete script that demonstrates the full workflow:

import asyncio
import os
from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub

async def main():
    # Configuration
    INSTANCE = "https://mastodon.social"
    ACCESS_TOKEN = os.getenv("MASTODON_ACCESS_TOKEN")

    if not ACCESS_TOKEN:
        print("Please set MASTODON_ACCESS_TOKEN environment variable")
        return

    async with AsyncClient(http2=True) as client:
        # Initialize client
        ap = ActivityPub(
            instance=INSTANCE,
            client=client,
            access_token=ACCESS_TOKEN
        )

        # Verify we're connected
        await ap.determine_instance_type()
        user = await ap.verify_credentials()
        print(f"Connected as: @{user['username']}@{INSTANCE.split('//')[1]}")

        # Post a test status
        status = await ap.post_status(
            "Testing minimal-activitypub library! 🚀"
        )
        print(f"Posted: {status['url']}")

        # Read some timelines
        timeline = await ap.get_public_timeline(limit=5)
        print(f"\nLatest from public timeline:")
        for post in timeline:
            account = post['account']['username']
            content_preview = post['content'][:50].replace('\n', ' ')
            print(f"  @{account}: {content_preview}...")

if __name__ == "__main__":
    asyncio.run(main())

Common Patterns

Error Handling

from minimal_activitypub import NetworkError, RatelimitError, ActivityPubError

async def safe_post(ap, text):
    try:
        return await ap.post_status(text)
    except NetworkError as e:
        print(f"Network error: {e.__cause__}")
    except RatelimitError as e:
        print(f"Rate limited at {e.endpoint}. Reset: {ap.ratelimit_reset}")
    except ActivityPubError as e:
        print(f"{e.method} {e.endpoint} failed: {e.status_code} — {e.message}")

See the Error Handling guide for the full exception hierarchy, all available attributes, and more detailed examples.

Pagination

async def get_all_statuses(ap, account_id):
    all_statuses = []
    max_id = None

    while True:
        batch = await ap.get_account_statuses(
            account_id=account_id,
            max_id=max_id,
            limit=40
        )

        if not batch:
            break

        all_statuses.extend(batch)
        max_id = batch[-1]['id']

        # Check if we have pagination info
        if not ap.pagination['next']['max_id']:
            break

    return all_statuses

Context Manager Usage

async with AsyncClient() as client:
    ap = ActivityPub(
        instance="https://mastodon.social",
        client=client,
        access_token=token
    )

    # All your API calls here
    # Client will be properly closed when block exits

Next Steps

  1. Explore the API: Check out the API Methods reference for all available methods.
  2. Learn Authentication: Read the Authentication guide for detailed OAuth flow information.
  3. Advanced Posting: See the Posting guide for media uploads, scheduling, and more.
  4. Check Examples: Look at the test files in the repository for more usage patterns.

Troubleshooting

Common Issues

  1. "Cannot resolve imported module" errors: Make sure you have all dependencies installed: bash pip install httpx[http2] pytz whenever

  2. Timeout errors: Increase the timeout when creating the ActivityPub instance: python ap = ActivityPub(..., timeout=300) # 5 minutes

  3. Rate limiting: The library automatically handles rate limits, but you can check current limits: python print(f"Remaining: {ap.ratelimit_remaining}/{ap.ratelimit_limit}")

  4. Instance compatibility: Some instances may have different API implementations. Use determine_instance_type() to detect the instance type.

Getting Help

  • Check the GitHub Issues for known problems
  • Review the test files for usage examples
  • Ensure you're using the latest version of the library

Best Practices

  1. Always use async/await: All API methods are asynchronous.
  2. Reuse HTTP clients: Create one AsyncClient and reuse it for multiple requests.
  3. Handle errors: Always wrap API calls in try/except blocks.
  4. Respect rate limits: The library helps, but monitor your usage.
  5. Store tokens securely: Never commit access tokens to version control.

Now you're ready to start building with minimal-activitypub! 🎉