An async Python SDK for the Meta Threads API — covering publishing, media retrieval, insights, reply management, and keyword search.
- Fully async — built on
aiohttpwith anasyncio-native interface - Typed responses — every API response is a frozen Pydantic v2 model
- Six API namespaces —
client.user,client.publishing,client.media,client.insights,client.replies,client.search - Convenience helpers — one-call publishing for text/image/video/carousel, container polling, auto-pagination
- Clean exception hierarchy — catch broad or specific errors as needed
- Forward-compatible — unknown fields from newer API versions are silently ignored
- Python ≥ 3.11
- A Meta developer account with a Threads app
- A valid user access token (OAuth 2.0 handled externally)
pip install threads-python-sdkimport asyncio
from threads import ThreadsClient, UserField, MediaField, MediaMetric
async def main() -> None:
async with ThreadsClient(access_token="your_token_here") as client:
# --- User Profile ---
me = await client.user.get_profile()
print(f"@{me.username} — {me.threads_biography}")
# --- Publish a text post ---
post = await client.publishing.post_text("me", "Hello Threads!")
print(f"Published: {post.id}")
# --- Publish an image ---
img = await client.publishing.post_image(
"me",
"https://example.com/photo.jpg",
text="Check this out",
)
# --- Publish a video (waits for processing automatically) ---
vid = await client.publishing.post_video(
"me",
"https://example.com/clip.mp4",
text="New video drop",
)
# --- Insights ---
insights = await client.insights.get_media_insights(post.id)
for metric in insights.data:
print(f"{metric.name}: {metric.values[0].value}")
# --- Paginate through all threads ---
async for thread in client.media.iter_threads():
print(thread.id, thread.text)
# --- Search ---
results = await client.search.search("#python")
for hit in results.data:
print(hit.text)
asyncio.run(main())ThreadsClient(access_token: str, *, timeout: float = 30.0)The main client. Use as an async context manager or call await client.aclose() manually.
| Method | Description |
|---|---|
get_profile(user_id, fields) |
Authenticated user's profile (or another user by ID) |
ID, USERNAME, NAME, THREADS_PROFILE_PICTURE_URL, THREADS_BIOGRAPHY, IS_VERIFIED
| Method | Description |
|---|---|
create_container(user_id, ...) |
Low-level: create a media container |
create_text_post(user_id, text, ...) |
Create a text-only container |
create_image_post(user_id, image_url, ...) |
Create an image container |
create_video_post(user_id, video_url, ...) |
Create a video container |
create_carousel_post(user_id, children, ...) |
Create a carousel container (2–20 items) |
publish(user_id, creation_id) |
Publish a container |
get_container_status(container_id) |
Check processing status |
wait_for_container(container_id, ...) |
Poll until terminal state |
post_text(user_id, text, ...) |
Convenience: create + publish text |
post_image(user_id, image_url, ...) |
Convenience: create + wait + publish image |
post_video(user_id, video_url, ...) |
Convenience: create + wait + publish video |
post_carousel(user_id, children, ...) |
Convenience: create + wait + publish carousel |
delete_post(thread_id) |
Delete a published thread |
get_publishing_limit() |
Check rate-limit quota |
TEXT, IMAGE, VIDEO, CAROUSEL
EVERYONE, ACCOUNTS_YOU_FOLLOW, MENTIONED_ONLY
IN_PROGRESS, FINISHED, ERRORED, EXPIRED, PUBLISHED
# Step 1: Create individual item containers
item1 = await client.publishing.create_image_post(
"me", "https://example.com/img1.jpg", is_carousel_item=True,
)
item2 = await client.publishing.create_image_post(
"me", "https://example.com/img2.jpg", is_carousel_item=True,
)
# Step 2: Create and publish carousel (convenience method)
post = await client.publishing.post_carousel(
"me", [item1.id, item2.id], text="My carousel",
)| Method | Description |
|---|---|
list_threads(user_id, *, fields, limit, since, until, after, before) |
One page of the user's threads |
get_thread(media_id, *, fields) |
Fetch a single thread by ID |
iter_threads(user_id, *, fields, limit) |
Async iterator over all threads (auto-paginates) |
ID, MEDIA_PRODUCT_TYPE, MEDIA_TYPE, MEDIA_URL, PERMALINK, OWNER,
USERNAME, TEXT, TIMESTAMP, SHORTCODE, THUMBNAIL_URL, CHILDREN, IS_QUOTE_POST
TEXT_POST, IMAGE, VIDEO, CAROUSEL_ALBUM, REPOST_FACADE
| Method | Description |
|---|---|
get_media_insights(media_id, metrics) |
Metrics for a specific post |
get_account_insights(*, metrics, period, since, until, breakdown) |
Account-level analytics |
get_metric(media_id, metric) |
Convenience: single metric for a post |
VIEWS, LIKES, REPLIES, REPOSTS, QUOTES, SHARES
All media metrics plus: FOLLOWERS_COUNT, FOLLOWER_DEMOGRAPHICS
DAY, WEEK, DAYS_28, LIFETIME
COUNTRY, CITY, AGE, GENDER
# Get follower demographics by country (requires 100+ followers)
from threads import AccountMetric, InsightPeriod, DemographicBreakdown
response = await client.insights.get_account_insights(
metrics=[AccountMetric.FOLLOWER_DEMOGRAPHICS],
period=InsightPeriod.LIFETIME,
breakdown=DemographicBreakdown.COUNTRY,
)| Method | Description |
|---|---|
get_replies(media_id, *, fields, reverse) |
Top-level replies to a thread |
get_conversation(media_id, *, fields, reverse) |
Full conversation tree |
hide_reply(reply_id) |
Hide a reply |
unhide_reply(reply_id) |
Unhide a reply |
get_mentions(*, fields) |
Threads that mention the authenticated user |
| Method | Description |
|---|---|
search(query, *, media_type, fields, since, until) |
Search public threads by keyword or #topic |
TEXT, IMAGE, VIDEO
ThreadsSDKError
├── ThreadsAPIError ← error response from the Graph API
│ ├── ThreadsAuthError ← invalid / expired token, OAuthException
│ ├── ThreadsRateLimitError
│ ├── ThreadsNotFoundError
│ └── ThreadsServerError ← 5xx from Meta
├── ThreadsPublishingError ← container processing failure or timeout
└── ThreadsConfigError ← SDK misconfiguration (e.g. empty token)
from threads import ThreadsAuthError, ThreadsRateLimitError, ThreadsAPIError
try:
me = await client.user.get_profile()
except ThreadsAuthError as e:
print(f"Auth failed: {e.error_code} — {e.message}")
except ThreadsRateLimitError:
print("Rate limited, back off and retry.")
except ThreadsAPIError as e:
print(f"API error [{e.http_status}] {e.error_type}: {e.message}")# Clone and install with dev extras
git clone https://github.com/inoue-ai/threads-python-sdk.git
cd threads-python-sdk
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
# Run tests
pytest
# Lint and format
ruff check .
ruff format .
# Type-check
mypy threads/This SDK does not implement the Threads OAuth 2.0 flow. Obtain an access
token externally (e.g. via your web server or the Meta App Dashboard) and
pass it to ThreadsClient. Token refresh must also be handled externally.
Required OAuth scopes per feature:
| Feature | Scope |
|---|---|
| User profile | threads_basic |
| Publish posts | threads_content_publish |
| Read media | threads_basic |
| Insights / analytics | threads_manage_insights |
| Read replies | threads_read_replies |
| Manage replies (hide/unhide) | threads_manage_replies |
| Keyword search | threads_keyword_search |
| Delete posts | threads_delete |