Python wrapper for the Mastodon API providing comprehensive access to social media functionality
npx @tessl/cli install tessl/pypi-mastodon-py@2.1.0A comprehensive Python wrapper for the Mastodon social media platform API, enabling developers to create applications that interact with Mastodon instances. Supports the complete Mastodon API (v4.4.3) including authentication, posting, media handling, real-time streaming, and administrative functionality.
pip install Mastodon.pypip install Mastodon.py[webpush,blurhash]from mastodon import MastodonFor streaming functionality:
from mastodon import Mastodon, StreamListener, CallbackStreamListenerFor error handling:
from mastodon import (
Mastodon, MastodonError, MastodonAPIError,
MastodonNetworkError, MastodonRatelimitError,
MastodonNotFoundError, MastodonUnauthorizedError,
MastodonVersionError, MastodonIllegalArgumentError
)from mastodon import Mastodon
# Create API client
mastodon = Mastodon(
client_id='your_clientcred.secret',
client_secret='your_clientcred.secret',
access_token='your_usercred.secret',
api_base_url='https://mastodon.social'
)
# Post a status
status = mastodon.status_post("Hello, Mastodon!")
print(f"Posted status: {status['id']}")
# Get home timeline
timeline = mastodon.timeline_home()
for toot in timeline:
print(f"{toot['account']['acct']}: {toot['content']}")
# Follow an account
account = mastodon.account_lookup("gargron@mastodon.social")
mastodon.account_follow(account['id'])The library uses a modular architecture with the main Mastodon class inheriting functionality from specialized mixins:
This design provides comprehensive API coverage while maintaining code organization and allowing fine-grained access control.
Complete OAuth authentication flows, application registration, and API client configuration. Handles credential persistence and token management for seamless integration.
def create_app(
client_name: str,
scopes: list = None,
redirect_uris: str = "urn:ietf:wg:oauth:2.0:oob",
website: str = None,
to_file: str = None,
api_base_url: str = None,
request_timeout: int = 300,
session: requests.Session = None
) -> tuple: ...
def __init__(
self,
client_id: str = None,
client_secret: str = None,
access_token: str = None,
api_base_url: str = None,
client_credential_file: str = None,
access_token_file: str = None,
debug_requests: bool = False,
ratelimit_method: str = "wait",
ratelimit_pacefactor: float = 1.1,
request_timeout: int = 300,
mastodon_version: str = None,
version_check_mode: str = "created",
session: requests.Session = None,
feature_set: str = "mainline",
user_agent: str = None,
lang: str = None
): ...
def log_in(
self,
username: str = None,
password: str = None,
code: str = None,
redirect_uri: str = "urn:ietf:wg:oauth:2.0:oob",
refresh_token: str = None,
scopes: list = None,
to_file: str = None
) -> str: ...User account operations including profile management, following/blocking, and relationship handling. Supports both personal account actions and account discovery features.
def account(self, id: int) -> dict: ...
def account_verify_credentials(self) -> dict: ...
def account_update_credentials(
self,
display_name: str = None,
note: str = None,
avatar: str = None,
header: str = None,
locked: bool = None,
bot: bool = None,
discoverable: bool = None,
fields_attributes: list = None,
source: dict = None
) -> dict: ...
def account_follow(self, id: int, reblogs: bool = True, notify: bool = False) -> dict: ...
def account_unfollow(self, id: int) -> dict: ...
def account_block(self, id: int) -> dict: ...
def account_mute(self, id: int, notifications: bool = True, duration: int = None) -> dict: ...Creating, editing, and managing posts (statuses/toots) including media attachments, polls, and scheduling. Handles all status-related actions like favoriting, boosting, and bookmarking.
def status_post(
self,
status: str,
in_reply_to_id: int = None,
media_ids: list = None,
sensitive: bool = False,
visibility: str = None,
spoiler_text: str = None,
language: str = None,
idempotency_key: str = None,
content_type: str = None,
scheduled_at: datetime = None,
poll: dict = None,
quote_id: int = None
) -> dict: ...
def status_delete(self, id: int) -> dict: ...
def status_favourite(self, id: int) -> dict: ...
def status_reblog(self, id: int, visibility: str = None) -> dict: ...
def status_bookmark(self, id: int) -> dict: ...Accessing various timeline feeds including home, public, hashtag, and list timelines. Provides paginated access to status streams with filtering options.
def timeline_home(
self,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None,
local: bool = False,
remote: bool = False
) -> list: ...
def timeline_public(
self,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None,
only_media: bool = False,
local: bool = False,
remote: bool = False
) -> list: ...
def timeline_hashtag(
self,
hashtag: str,
local: bool = False,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None,
only_media: bool = False,
remote: bool = False
) -> list: ...Upload and manage media attachments including images, videos, and audio files. Supports metadata editing, focus points, and thumbnails for enhanced media presentation.
def media_post(
self,
media_file: str,
mime_type: str = None,
description: str = None,
focus: tuple = None,
file_name: str = None,
thumbnail: str = None,
thumbnail_mime_type: str = None,
synchronous: bool = False
) -> dict: ...
def media_update(
self,
id: int,
description: str = None,
focus: tuple = None,
thumbnail: str = None,
thumbnail_mime_type: str = None
) -> dict: ...WebSocket-based streaming for real-time updates from timelines, notifications, and user events. Implements event handlers for different types of streaming data.
def stream_user(
self,
listener: StreamListener,
run_async: bool = False,
timeout: int = 300,
reconnect_async: bool = False,
reconnect_async_wait_sec: int = 5
): ...
def stream_public(
self,
listener: StreamListener,
run_async: bool = False,
timeout: int = 300,
reconnect_async: bool = False,
reconnect_async_wait_sec: int = 5
): ...
class StreamListener:
def on_update(self, status: dict): ...
def on_notification(self, notification: dict): ...
def on_delete(self, status_id: int): ...Comprehensive search functionality across accounts, statuses, and hashtags with support for both v1 and v2 search APIs.
def search(
self,
q: str,
resolve: bool = True,
result_type: str = None,
account_id: int = None,
max_id: int = None,
min_id: int = None,
limit: int = None,
offset: int = 0,
following: bool = False
) -> dict: ...
def search_v2(
self,
q: str,
resolve: bool = True,
result_type: str = None,
account_id: int = None,
max_id: int = None,
min_id: int = None,
limit: int = None,
offset: int = 0,
following: bool = False
) -> dict: ...Create and manage content filters using both v1 and v2 API versions for automated content moderation.
def filters(self) -> list: ...
def filter_create(
self,
phrase: str,
context: str,
irreversible: bool = False,
whole_word: bool = True,
expires_in: int = None
) -> dict: ...
def filters_v2(self) -> list: ...
def create_filter_v2(
self,
title: str,
context: list,
filter_action: str = "warn",
expires_in: int = None,
keywords_attributes: list = None
) -> dict: ...Create and manage user lists for organizing followed accounts and customizing timeline feeds.
def lists(self) -> list: ...
def list_create(
self,
title: str,
replies_policy: str = "list",
exclusive: bool = False
) -> dict: ...
def list_accounts(
self,
id: int,
max_id: int = None,
since_id: int = None,
limit: int = None
) -> list: ...
def list_accounts_add(self, id: int, account_ids: list): ...Comprehensive notification management including the new grouped notifications API and notification policies.
def notifications(
self,
id: int = None,
account_id: int = None,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None,
exclude_types: list = None,
types: list = None
) -> list: ...
def grouped_notifications(
self,
max_id: str = None,
since_id: str = None,
limit: int = None,
types: list = None,
exclude_types: list = None,
grouped_types: list = None
) -> dict: ...Create, vote on, and manage polls attached to statuses.
def poll(self, id: int) -> dict: ...
def poll_vote(self, id: int, choices: list) -> dict: ...
def make_poll(
self,
options: list,
expires_in: int,
multiple: bool = False,
hide_totals: bool = False
) -> dict: ...Access various timeline feeds including specialized timelines like link and list timelines.
def timeline_list(
self,
id: int,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None
) -> list: ...
def timeline_link(
self,
url: str,
local: bool = False,
max_id: int = None,
min_id: int = None,
since_id: int = None,
limit: int = None
) -> list: ...Instance management and moderation functionality for admin users including account and report management.
def admin_accounts(
self,
remote: bool = False,
by_domain: str = None,
status: str = "active",
username: str = None,
display_name: str = None,
email: str = None,
ip: str = None,
staff: bool = None,
max_id: int = None,
since_id: int = None,
limit: int = None
) -> list: ...
def admin_account_moderate(
self,
id: int,
action: str = None,
report_id: int = None,
warning_preset_id: int = None,
text: str = None,
send_email_notification: bool = True
) -> dict: ...class Mastodon:
"""Main API client class for interacting with Mastodon instances."""
pass
class StreamListener:
"""Base class for handling streaming events."""
def on_update(self, status: dict): pass
def on_notification(self, notification: dict): pass
def on_delete(self, status_id: int): pass
def on_abort(self, err: Exception): pass
class AttribAccessDict(dict):
"""Dictionary with attribute-style access to keys."""
pass
# Exception hierarchy
class MastodonError(Exception):
"""Base exception for all Mastodon.py errors."""
pass
class MastodonAPIError(MastodonError):
"""Errors returned by the Mastodon API."""
pass
class MastodonNetworkError(MastodonError):
"""Network communication errors."""
pass
class MastodonRatelimitError(MastodonError):
"""Rate limit exceeded errors."""
pass
class MastodonUnauthorizedError(MastodonAPIError):
"""Authentication and authorization errors."""
pass
class MastodonNotFoundError(MastodonAPIError):
"""Resource not found errors."""
pass
class MastodonVersionError(MastodonError):
"""Raised when a function is called that the version of Mastodon does not support."""
pass
class MastodonIllegalArgumentError(ValueError, MastodonError):
"""Raised when an incorrect parameter is passed to a function."""
pass
class MastodonIOError(IOError, MastodonError):
"""Base class for Mastodon.py I/O errors."""
pass
class MastodonFileNotFoundError(MastodonIOError):
"""Raised when a file requested to be loaded can not be opened."""
pass
class MastodonReadTimeout(MastodonNetworkError):
"""Raised when a stream times out."""
pass
class MastodonServerError(MastodonAPIError):
"""Raised if the Server is malconfigured and returns a 5xx error code."""
pass
class CallbackStreamListener(StreamListener):
"""Stream listener that calls registered callback functions for events."""
def __init__(self,
update_handler=None,
local_update_handler=None,
delete_handler=None,
notification_handler=None,
conversation_handler=None,
unknown_event_handler=None): ...