Metadata-Version: 2.4
Name: omnisocials
Version: 0.1.0
Summary: Official Python SDK for the OmniSocials Public API - schedule and publish social media posts, upload media, and read analytics across 11 platforms.
Project-URL: Homepage, https://omnisocials.com
Project-URL: Documentation, https://docs.omnisocials.com
Project-URL: Repository, https://github.com/OmniSocials/omnisocials-python
Author-email: OmniSocials <hello@omnisocials.com>
License-Expression: MIT
License-File: LICENSE
Keywords: api,facebook,instagram,linkedin,omnisocials,scheduling,sdk,social media,tiktok,twitter,youtube
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx<1,>=0.23.0
Description-Content-Type: text/markdown

# OmniSocials Python SDK

The official Python client for the [OmniSocials](https://omnisocials.com) Public API. Schedule and publish posts, upload media, and read analytics across Instagram, Facebook, LinkedIn, YouTube, TikTok, X, Pinterest, Bluesky, Threads, Mastodon, and Google Business.

Full API reference: [https://docs.omnisocials.com](https://docs.omnisocials.com)

## Install

```bash
pip install omnisocials
```

Requires Python 3.9+. The only dependency is [httpx](https://www.python-httpx.org/).

## Quickstart

```python
from omnisocials import OmniSocials

client = OmniSocials()  # reads OMNISOCIALS_API_KEY from the environment
post = client.posts.create(content="Hello from Python", channels=["instagram", "linkedin"])
print(post["data"]["id"])
```

## Authentication

Create an API key in the OmniSocials app under **Settings -> API Keys**. Keys look like `omsk_live_...` (or `omsk_test_...` for test keys) and are bound to one workspace.

Pass the key explicitly or set the `OMNISOCIALS_API_KEY` environment variable:

```python
from omnisocials import OmniSocials

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

The client is a context manager; use `with` (or call `client.close()`) to release connections:

```python
with OmniSocials() as client:
    accounts = client.accounts.list()
```

## Async client

Every resource and method exists on `AsyncOmniSocials` 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())
```

## Posts

### Schedule 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"],
)
```

Omit `scheduled_at` to create a draft, or use `create_and_publish` to publish immediately:

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

### Post to specific platforms with platform options

`content` can be a per-platform mapping (with `default` as the fallback), and each platform has its own options object:

```python
post = client.posts.create(
    content={
        "default": "New feature: best times to post",
        "linkedin": "We just shipped best times to post. Here is how it works.",
    },
    channels=["linkedin", "instagram", "youtube"],
    media_urls=["https://example.com/demo.mp4"],
    scheduled_at="2026-08-01T09:00:00Z",
    instagram={"share_to_feed": True},
    youtube={"title": "Best times to post, explained", "privacy_status": "public"},
)
```

Note that `linkedin` targets a personal LinkedIn profile and `linkedin_page` targets a LinkedIn company page. Both can be connected to the same workspace and posted to independently.

### X thread example

Pass 2 to 25 `thread_parts` to publish a chained thread instead of a single tweet (each part is at most 280 characters). Bluesky and Mastodon support the same `thread_parts` shape:

```python
post = client.posts.create(
    content="How we cut our API response times by 80%",  # fallback text
    channels=["x"],
    scheduled_at="2026-08-01T15:00:00Z",
    x={
        "thread_parts": [
            {"text": "How we cut our API response times by 80%. A thread:"},
            {"text": "1/ We started by profiling every endpoint under real load."},
            {"text": "2/ The biggest win: batching analytics reads into one query."},
            {"text": "3/ Full write-up on the blog. Thanks for reading!"},
        ]
    },
)
```

On update, passing `x={"thread_parts": None}` clears the thread and reverts the post to single-tweet mode. Only top-level `None` values are dropped from request bodies, so nested `None` values like this one are sent as JSON `null`.

### Other post operations

```python
posts = client.posts.list(status="scheduled", limit=50, offset=0)
post = client.posts.get("123")
client.posts.update("123", scheduled_at="2026-08-02T10:00:00Z")
client.posts.publish("123")           # publish a draft or scheduled post now
client.posts.delete("123")            # returns None (204)

# Recent posts fetched live from the connected platforms (including content
# published outside OmniSocials). Requires the analytics:read scope.
recent = client.posts.recent_platform(limit=10, platforms=["instagram", "tiktok"])
```

## Media

### Upload from a URL (recommended, up to 1GB)

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

Files over 100MB are processed in the background: the response has `data["status"] == "processing"`; poll `client.media.get(media_id)` until it is `"ready"` before attaching it to a post.

### Upload a local file (multipart, max 100MB)

```python
media = client.media.upload(file="/path/to/image.jpg", name="hero-shot")
# file can also be raw bytes or a binary file-like object
```

Uploading a PDF splits it into image slides (max 20 pages) and returns `slides` plus a `media_ids` array. Pass all of `media_ids`, in order, to `posts.create` to publish the deck as a carousel (a native swipeable document on LinkedIn, an image carousel elsewhere).

Every upload response also includes a `compatibility` block listing any connected platforms that would reject the file.

### Other media operations

```python
files = client.media.list(search="promo", folder_id="42")
item = client.media.get("1001")
client.media.update("1001", name="renamed", folder_id="42")
client.media.update("1001", folder_id=None)   # move to the root ("All media")
client.media.delete("1001")

# Base64 upload (handy when you have bytes but no URL)
client.media.upload_from_base64(data=b64_string, mime_type="image/png")

# Preflight compatibility BEFORE uploading
result = client.media.check(url="https://example.com/huge.mp4")
if not result["compatible"]:
    print(result["summary"])
```

### Large file uploads with a presigned URL

`create_upload_url` mints a one-time upload URL so any environment (a CI job, a sandbox, a script without the SDK) can upload a file without auth headers:

```python
import httpx

grant = client.media.create_upload_url()
# grant contains: upload_url, upload_token, expires_in_seconds (600),
# method ("POST"), content_type ("multipart/form-data")

with open("/path/to/video.mp4", "rb") as f:
    response = httpx.post(grant["upload_url"], files={"file": f})
media_id = response.json()["data"]["id"]
```

The token is single-use and expires after 10 minutes. The response JSON contains the created media item with its `id` (or a `media_ids` array for a PDF).

## Folders

```python
folders = client.folders.list()
folder = client.folders.create(name="Campaigns")
sub = client.folders.create(name="August", parent_id=folder["data"]["id"])
client.folders.update(sub["data"]["id"], name="August 2026")
client.folders.update(sub["data"]["id"], parent_id=None)  # move to top level
client.folders.delete(folder["data"]["id"])  # files move to root, subfolders move up
```

## Accounts

```python
accounts = client.accounts.list()
for account in accounts["data"]:
    print(account["platform"], account["username"])

account = client.accounts.get("instagram")
```

## 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")
```

### Best times to post

```python
best = client.analytics.best_times(platform="instagram", timezone="Europe/Amsterdam")
for slot in best["data"]["best_times"]:
    print(slot)
```

## Locations (Instagram place tagging)

```python
results = client.locations.search("Blue Bottle Coffee Oakland")
location_id = results["data"][0]["id"]
client.locations.validate(location_id)

client.posts.create(
    content="Great coffee here",
    channels=["instagram"],
    media_urls=["https://example.com/latte.jpg"],
    location_id=location_id,
)
```

## Webhooks

Subscribe to `post.scheduled`, `post.published`, and `post.failed` events:

```python
webhook = client.webhooks.create(
    url="https://example.com/hooks/omnisocials",
    events=["post.published", "post.failed"],
)
secret = webhook["data"]["secret"]  # only shown once, store it

client.webhooks.list()
client.webhooks.get(webhook["data"]["id"])
client.webhooks.update(webhook["data"]["id"], is_active=False)
client.webhooks.rotate_secret(webhook["data"]["id"])  # new secret, shown once
client.webhooks.delete(webhook["data"]["id"])
```

### Verifying webhook signatures (FastAPI example)

Every delivery is signed with HMAC-SHA256. The `X-OmniSocials-Signature` header has the form `t=<unix>,v1=<hex>`, signed over `"{timestamp}.{raw_body}"`. Use `verify_webhook_signature` with the raw request body (do not parse it first):

```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"))
    return {"ok": True}
```

`verify_webhook_signature` uses a constant-time comparison, rejects timestamps older than `tolerance` seconds (default 300) to block replays, and returns the parsed event dict on success. Respond with any 2xx within 10 seconds to acknowledge; deliveries are retried up to 3 times.

## Error handling

Non-2xx responses raise typed exceptions with `status`, `code`, `message`, and the parsed `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)
```

The full hierarchy: `OmniSocialsError` is the base; `APIError` covers all HTTP errors with subclasses `AuthenticationError` (401), `PermissionDeniedError` (403), `NotFoundError` (404), `ValidationError` (400/422), `RateLimitError` (429, with `.retry_after`), and `ServerError` (5xx); `APIConnectionError` covers network failures; `WebhookVerificationError` covers signature failures.

## Rate limits

The API allows **100 requests per minute** per API key. Every response includes `X-RateLimit-Limit`, `X-RateLimit-Remaining`, and `X-RateLimit-Reset` headers. On 429 the SDK automatically waits for the `Retry-After` value and retries. When syncing analytics for many posts, prefer the bulk `client.analytics.posts([...])` endpoint (100 posts per request) over per-post calls.

## Retries and timeouts

The SDK automatically retries on 429, 5xx, and connection errors with exponential backoff (0.5s, 1s, 2s, with jitter), honoring the `Retry-After` header when present. Other 4xx errors are never retried. Defaults: 2 retries, 30 second timeout. Both are configurable:

```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
)
```

## More

- Full API reference and guides: [https://docs.omnisocials.com](https://docs.omnisocials.com)
- MCP server for AI agents: `npx -y @omnisocials/mcp-server`
- Agent skills CLI: `npx skills add OmniSocials/omnisocials-agent-skills`

## License

MIT
