Copy page

SDKs

Python SDK reference: install, auth and methods

The official Python SDK. Python 3.9+, the only dependency is httpx, and both a synchronous and an async client ship in the same package.

This page is the reference. For an overview of the Python SDK, see omnisocials.com/sdk/python.

Install

Terminal
pip install omnisocials

Authentication

Python
from omnisocials import OmniSocials

client = OmniSocials(api_key="omsk_live_...")
# or: export OMNISOCIALS_API_KEY=omsk_live_... and just OmniSocials()

Context manager form:

Python
with OmniSocials() as client:
    accounts = client.accounts.list()

Create a post

Python
post = client.posts.create(
    content="Big announcement coming Friday",
    channels=["instagram", "facebook", "linkedin"],
    scheduled_at="2026-08-01T09:00:00Z",
    media_urls=["https://example.com/teaser.jpg"],
)

Any media_urls (or media_ids) entry can also be a dict carrying per-media alt text (max 1500 chars), delivered to Mastodon, Bluesky, X (photos), Pinterest, Instagram (images), and LinkedIn (images):

Python
media_urls=[{"url": "https://example.com/teaser.jpg", "alt": "Red sneaker on a white background"}],

Publish immediately:

Python
post = client.posts.create_and_publish(
    content="We are live!",
    channels=["x", "bluesky"],
)

Upload media

Python
media = client.media.upload_from_url(
    url="https://example.com/promo.mp4",
    name="promo-august",
    folder="Campaigns",
)
media_id = media["data"]["id"]

Local files: client.media.upload(file="/path/to/image.jpg", name="hero-shot") (file accepts a path, bytes, or a binary file-like object). Base64: client.media.upload_from_base64(data=b64_string, mime_type="image/png").

Analytics

Python
# One post
stats = client.analytics.post("123")

# Up to 100 posts in one request (avoids rate limit trouble when syncing)
bulk = client.analytics.posts(["123", "124", "125"])

# Workspace overview
overview = client.analytics.overview(period="30d")

# Account-level stats (followers etc.)
accounts = client.analytics.accounts(platform="instagram")

Social Inbox

Read and answer DMs, comments and mentions from your connected accounts. The API key needs the opt-in inbox:read / inbox:write scopes. The lists use cursor pagination (cursor in, pagination["next_cursor"] out).

Python
conversations = client.inbox.list_conversations(platform="instagram", unread=True)
conversation_id = conversations["data"][0]["conversation_id"]

thread = client.inbox.get_messages(conversation_id)
client.inbox.reply(conversation_id, "Thanks for reaching out!")
client.inbox.mark_read(conversation_id)

# Comment moderation: hide (hide=False unhides) or delete
client.inbox.hide(thread["data"][0]["id"])
client.inbox.delete_message(thread["data"][0]["id"])

inbox.next() hands out the next conversation that still needs an answer, with the thread and the post it belongs to. On a comment thread, pass message_id so the reply goes under the comment you answer and not under the newest comment on the post.

Python
item = client.inbox.next()
while item["data"]:
    message = item["data"]["message"]

    if not item["data"]["reply_window"]["open"]:
        # An Instagram/Facebook DM past Meta's 24-hour window: answer it in the app
        client.inbox.mark_read(message["conversation_id"])
        item = client.inbox.next()
        continue

    reply = client.inbox.reply(
        message["conversation_id"],
        "Thanks! DM sent.",
        message_id=message["id"],
        include_next=True,
    )
    item = {"data": reply.get("next"), "remaining": reply.get("remaining", 0)}

The same methods are on AsyncOmniSocials (await client.inbox.next()). See the Social Inbox guide for the platforms, the reply rules and the error codes.

Pinterest product tags

Tag products on a Pin so people can shop the items in the image. List the product Pins of the connected Pinterest account, then pass up to 24 of their Pin ids as product_tags in the pinterest options of the post. Only product Pins of your own account can be tagged; products of other merchants cannot.

Python
result = client.pinterest.list_products()

if "error" in result:
    print(result["error"]["code"], result["error"]["message"])
else:
    client.posts.create(
        content="Our summer picks",
        channels=["pinterest"],
        media_urls=["https://example.com/summer-look.jpg"],
        scheduled_at="2026-08-01T09:00:00Z",
        pinterest={
            "board_id": "1234567890",
            "product_tags": [product["pin_id"] for product in result["products"][:3]],
        },
    )

The list answers with HTTP 200 in both cases: "products" on success, or an "error" object (pinterest_not_connected, pinterest_catalog_access_required, platform_error) when it could not be read. It reads the Pinterest catalog when the connection has catalog access (given one time in the OmniSocials composer: Pinterest options, Add products, Connect catalog), else the account's own Pins. The tags are added right after the Pin is published. A product that Pinterest refuses never fails the post: the post then carries pinterest["product_tags_result"] with the tagged and the skipped Pin ids. client.pinterest.validate_product(id) checks one Pin id or Pin link before you post. On posts.update() the pinterest object replaces the stored one, so leave product_tags out to remove the tags. The same methods are on AsyncOmniSocials.

Verify webhooks

Python
from fastapi import FastAPI, Header, HTTPException, Request
from omnisocials import WebhookVerificationError, verify_webhook_signature

app = FastAPI()
WEBHOOK_SECRET = "whsec_from_webhook_creation"

@app.post("/hooks/omnisocials")
async def omnisocials_webhook(
    request: Request,
    x_omnisocials_signature: str = Header(None),
):
    payload = await request.body()  # raw bytes, exactly as received
    try:
        event = verify_webhook_signature(
            payload, x_omnisocials_signature, WEBHOOK_SECRET, tolerance=300
        )
    except WebhookVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    if event["type"] == "post.published":
        for target in event["data"]["targets"]:
            print(target["platform"], target["status"], target.get("native_post_id"))
    elif event["type"] == "post.rejected":
        print("Rejected:", event["data"]["post_id"], event["data"]["approval"]["reason"])
    return {"ok": True}

Events: post.scheduled, post.published, post.failed, post.approved and post.rejected. The two approval events carry data.approval with status, decided_by (the approver's user id) and reason (None on post.approved). To read the full review of a post (steps, approvers, the rejection reason and the comment thread), call client.posts.get_approval(post_id).

Errors

The base class is OmniSocialsError. APIError covers HTTP errors with subclasses AuthenticationError (401), PermissionDeniedError (403), NotFoundError (404), ValidationError (400/422), RateLimitError (429, with .retry_after), and ServerError (5xx), plus APIConnectionError and WebhookVerificationError. Each carries status, code, message, and body.

Python
from omnisocials import (
    OmniSocials,
    APIError,
    APIConnectionError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

client = OmniSocials()

try:
    client.posts.get("does-not-exist")
except NotFoundError as exc:
    print("Gone:", exc.code, exc.message)
except RateLimitError as exc:
    print(f"Slow down, retry in {exc.retry_after}s")
except ValidationError as exc:
    print("Bad request:", exc.body)
except AuthenticationError:
    print("Check your API key")
except APIConnectionError:
    print("Network problem")
except APIError as exc:
    print("API error:", exc.status, exc.message)

Configuration

Python
client = OmniSocials(
    api_key="omsk_live_...",
    timeout=60.0,       # seconds
    max_retries=5,      # retries after the first attempt
    base_url="https://api.omnisocials.com/v1",  # override for testing
)

Async client

AsyncOmniSocials mirrors every resource and method with identical names:

Python
import asyncio
from omnisocials import AsyncOmniSocials

async def main():
    async with AsyncOmniSocials() as client:
        posts = await client.posts.list(status="scheduled")
        print(len(posts["data"]))

asyncio.run(main())

Notes

• Responses are plain dicts; list endpoints include data and pagination
• Top-level None values are dropped from request bodies; nested None is sent as JSON null
• Source: github.com/OmniSocials/omnisocials-python