API Methods Reference

This document provides a comprehensive reference for all available API methods in minimal-activitypub.

ActivityPub Class

The main class for interacting with ActivityPub servers.

Constructor

ActivityPub(
    instance: str,
    client: AsyncClient,
    access_token: str | None = None,
    timeout: int | None = None
)

Parameters: - instance: Domain name or URL of the instance to connect to - client: httpx.AsyncClient to use for communication - access_token: Authentication token (optional) - timeout: Seconds to wait for response (default: 120)

Example:

from httpx import AsyncClient
from minimal_activitypub.client_2_server import ActivityPub

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

Authentication Methods

create_app

@staticmethod
async def create_app(
    instance_url: str,
    client: AsyncClient
) -> tuple[str, str]

Creates an OAuth application on the specified instance.

Parameters: - instance_url: URL of the instance - client: AsyncClient instance

Returns: Tuple of (client_id, client_secret)

get_auth_token

@staticmethod
async def get_auth_token(
    instance_url: str,
    username: str,
    password: str,
    client: AsyncClient
) -> str

Gets an access token using username and password authentication.

Parameters: - instance_url: URL of the instance - username: User's username or email - password: User's password - client: AsyncClient instance

Returns: Access token string

generate_authorization_url

@staticmethod
async def generate_authorization_url(
    instance_url: str,
    client_id: str,
    user_agent: str
) -> str

Generates an authorization URL for OAuth flow.

Parameters: - instance_url: URL of the instance - client_id: OAuth client ID - user_agent: User agent string

Returns: Authorization URL string

validate_authorization_code

@staticmethod
async def validate_authorization_code(
    client: AsyncClient,
    instance_url: str,
    authorization_code: str,
    client_id: str,
    client_secret: str
) -> str

Validates an authorization code and returns an access token.

Parameters: - client: AsyncClient instance - instance_url: URL of the instance - authorization_code: Authorization code from user - client_id: OAuth client ID - client_secret: OAuth client secret

Returns: Access token string

Instance Methods

determine_instance_type

async def determine_instance_type() -> None

Determines the type of instance (Mastodon, Pleroma, or Takahe) and sets instance-specific parameters.

verify_credentials

async def verify_credentials() -> dict

Verifies that the access token is valid and returns the authenticated user's information.

Returns: Dictionary with user account information

Timeline Methods

get_public_timeline

async def get_public_timeline(
    local: bool = False,
    remote: bool = False,
    only_media: bool = False,
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20
) -> list[dict]

Gets the public timeline.

Parameters: - local: Show local statuses only (default: False) - remote: Show remote statuses only (default: False) - only_media: Show only statuses with media (default: False) - max_id: Return results older than this ID - since_id: Return results newer than this ID - min_id: Return results immediately newer than this ID - limit: Maximum number of results (default: 20)

Returns: List of status dictionaries

get_home_timeline

async def get_home_timeline(
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20
) -> list[dict]

Gets the home timeline (statuses from followed accounts).

Parameters: - max_id: Return results older than this ID - since_id: Return results newer than this ID - min_id: Return results immediately newer than this ID - limit: Maximum number of results (default: 20)

Returns: List of status dictionaries

get_hashtag_timeline

async def get_hashtag_timeline(
    hashtag: str,
    local: bool = False,
    remote: bool = False,
    only_media: bool = False,
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20
) -> list[dict]

Gets the timeline for a specific hashtag.

Parameters: - hashtag: Hashtag name (without #) - local: Show local statuses only (default: False) - remote: Show remote statuses only (default: False) - only_media: Show only statuses with media (default: False) - max_id: Return results older than this ID - since_id: Return results newer than this ID - min_id: Return results immediately newer than this ID - limit: Maximum number of results (default: 20)

Returns: List of status dictionaries

Status Methods

get_account_statuses

async def get_account_statuses(
    account_id: str,
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20,
    exclude_replies: bool = False,
    exclude_reblogs: bool = False,
    pinned: bool = False,
    tagged: str | None = None
) -> list[dict]

Gets statuses from a specific account.

Parameters: - account_id: Account ID to fetch statuses from - max_id: Return results older than this ID - since_id: Return results newer than this ID - min_id: Return results immediately newer than this ID - limit: Maximum number of results (default: 20) - exclude_replies: Exclude reply statuses (default: False) - exclude_reblogs: Exclude reblog statuses (default: False) - pinned: Include only pinned statuses (default: False) - tagged: Filter by hashtag (without #)

Returns: List of status dictionaries

post_status

async def post_status(
    status: str,
    visibility: Visibility = Visibility.PUBLIC,
    media_ids: list[str] | None = None,
    sensitive: bool = False,
    spoiler_text: str | None = None,
    scheduled_at: datetime | None = None
) -> dict

Posts a status to the fediverse.

Parameters: - status: The text content to post - visibility: Visibility level (default: Visibility.PUBLIC) - media_ids: List of media IDs to attach - sensitive: Mark as sensitive content (default: False) - spoiler_text: Content warning text - scheduled_at: Schedule post for future publication

Returns: Dictionary of the posted status

delete_status

async def delete_status(status_id: str) -> dict

Deletes a status.

Parameters: - status_id: ID of the status to delete

Returns: Dictionary of the deleted status

reblog

async def reblog(status_id: str) -> dict

Reblogs (boosts) a status.

Parameters: - status_id: ID of the status to reblog

Returns: Dictionary of the reblogged status

undo_reblog

async def undo_reblog(status_id: str) -> dict

Undoes a reblog (unboosts a status).

Parameters: - status_id: ID of the status to unboost

Returns: Dictionary of the status

undo_favourite

async def undo_favourite(status_id: str) -> dict

Removes a status from favorites.

Parameters: - status_id: ID of the status to unfavorite

Returns: Dictionary of the status

Media Methods

post_media

async def post_media(
    file: IO[bytes],
    mime_type: str,
    description: str | None = None,
    focus: tuple[float, float] | None = None
) -> dict

Uploads a media file to the instance.

Parameters: - file: File object opened in binary mode - mime_type: MIME type of the file - description: Alt text for accessibility - focus: Focal point coordinates as (x, y) tuple

Returns: Dictionary with media information including id

Search Methods

search

async def search(
    query: str,
    search_type: SearchType = SearchType.STATUSES,
    resolve: bool = False,
    following: bool = False,
    account_id: str | None = None,
    exclude_unreviewed: bool = False,
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20,
    offset: int = 0
) -> dict

Searches for content on the instance.

Parameters: - query: Search query string - search_type: Type of search (accounts, hashtags, or statuses) - resolve: Attempt to resolve non-local accounts (default: False) - following: Only include accounts the user is following (default: False) - account_id: Search within a specific account's statuses - exclude_unreviewed: Exclude unreviewed tags (default: False) - max_id: Return results older than this ID - since_id: Return results newer than this ID - min_id: Return results immediately newer than this ID - limit: Maximum number of results (default: 20) - offset: Skip the first N results (default: 0)

Returns: Dictionary with search results

Utility Methods

_pre_call_checks

async def _pre_call_checks() -> None

Internal method that checks rate limits before making API calls.

_parse_next_prev

def _parse_next_prev(links: str | None) -> None

Internal method that parses pagination links from response headers.

_update_ratelimit

def _update_ratelimit(headers: Any) -> None

Internal method that updates rate limit information from response headers.

Enumerations

Visibility

class Visibility(str, Enum):
    PUBLIC = "public"
    UNLISTED = "unlisted"
    PRIVATE = "private"
    DIRECT = "direct"

Visibility levels for statuses.

SearchType

class SearchType(str, Enum):
    ACCOUNTS = "accounts"
    HASHTAGS = "hashtags"
    STATUSES = "statuses"

Search types for the search method.

Error Classes

The library defines the following exception classes (all inherit from ActivityPubError):

Exception	HTTP Status	When raised
NetworkError	—	Transport/connection failure
ApiError	—	Generic API error
ClientError	4xx	Unmapped 4xx status code
UnauthorizedError	401	Bad or missing access token
ForbiddenError	403	Token lacks required scope
NotFoundError	404	Resource does not exist
ConflictError	409	Action conflicts with current state
GoneError	410	Resource permanently removed
UnprocessedError	422	Server rejected request parameters
RatelimitError	429	Rate limit exceeded
ServerError	5xx	Server-side error

All exceptions (except NetworkError) carry the following attributes set automatically from the HTTP response: status_code, reason_phrase, message, endpoint, method, request_id, response_text, occurred_at.

See the Error Handling guide for full details and examples.

Example Usage

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

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

        # Get public timeline
        timeline = await ap.get_public_timeline(limit=10)
        print(f"Got {len(timeline)} statuses")

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

asyncio.run(main())

Rate Limiting

The library automatically handles rate limiting by: 1. Parsing rate limit headers from responses 2. Tracking remaining requests and reset times 3. Raising RatelimitError when limits are exceeded 4. Providing rate limit information via instance attributes: - ratelimit_limit: Total requests allowed - ratelimit_remaining: Remaining requests - ratelimit_reset: When limits reset

Pagination

Methods that return lists of items support pagination using: - max_id: Get items older than this ID - since_id: Get items newer than this ID - min_id: Get items immediately newer than this ID

Pagination information is automatically parsed from Link headers and stored in the pagination attribute.

Notes

• All methods are asynchronous and must be awaited
• The library requires Python 3.10 or higher
• HTTP/2 is recommended for better performance
• Always handle exceptions appropriately in production code