# Python Social Media API SDK

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.

## Install

```bash
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/GIFs),
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](/inbox) 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](https://github.com/OmniSocials/omnisocials-python)