planvortex

The official Python client for the PlanVortex API — connect social accounts, schedule and publish posts, read comments and messages, and pull stats, from Python.

pip install planvortex

• Synchronous and asynchronous, same surface: PlanVortex and AsyncPlanVortex.
• Typed, from the same OpenAPI specification the API publishes at https://planvortex.com/openapi.json. Returned shapes are TypedDict, so _id stays _id.
• One runtime dependency, httpx2 — a different package from httpx classic, so it will not collide with whatever your project already uses.
• Server-side. The client_credentials flow needs your client_secret, which must never reach a browser. Connecting an account from one is what the temporal connect token is for.
• Python 3.10 and newer.

Reference: https://taliasoftworks.github.io/PlanVortexPython/ · Guides: planvortex.com/developers

Authentication

Your credentials are a client app's: you create one in the PlanVortex panel and it gives you a client_id and a client_secret. They are read from the environment, so nothing is hardcoded:

export PLANVORTEX_CLIENT_ID=...
export PLANVORTEX_CLIENT_SECRET=...
# Only if you are not talking to production:
export PLANVORTEX_BASE_URL=http://localhost:3000/v1.0.0

from planvortex import PlanVortex

pv = PlanVortex()  # from the environment
pv = PlanVortex(client_id="...", client_secret="...")  # or explicitly

The token is fetched on the first call, cached, and renewed before it expires — you never touch /oauth/token. Use the client as a context manager (with PlanVortex() as pv:) so the connection pool closes, and keep one instance: a new one per request throws away the cache and the pool.

An app sees its own client and that client's organizations, and nothing else.

Publishing

from datetime import datetime, timedelta, timezone

from planvortex import PlanVortex

pv = PlanVortex()

upload = pv.uploads.create(org_id, "./sourdough.jpg")
publication = pv.publications.create(
    org_id,
    account_id,
    {
        "social_network": "instagram",
        "text": "New oven, new loaves",
        "files": [upload["_id"]],
        "publish_date": datetime.now(timezone.utc) + timedelta(hours=1),
    },
)

# A publication that could not be built is NOT an exception: it comes back saved, in `withErrors`,
# with the reason inside. The content is validated against the network, and that is not a failure
# of your request.
if publication["state"] == "withErrors":
    for failure in publication["publication_errors"]:
        print(failure["code"], failure["message"])

publish_date takes a datetime as well as an ISO-8601 string, and it has to carry a timezone: a naive one raises rather than being guessed at, because assuming UTC publishes at the wrong time for whoever is in Madrid and assuming the process's zone does it for whoever is in Docker. With no publish_date at all it goes out in that same request, and the answer already says whether it did.

A file can be a path, an open file, bytes, or a (name, bytes) pair. Per-network limits — characters, images, video length, file size — come from pv.catalog.social_limits(), which is the server's own copy: whoever enforces a limit is who gets to announce it.

Listing gives you a page, and there is a chaining iterator for when you want them all:

page = pv.accounts.list(org_id, limit=50)
page.data, page.total

for publication in pv.publications.iterate(org_id, state=["ready"]):
    ...

The same code in async changes three things and no more — the class, an await, and aiterate:

from planvortex import AsyncPlanVortex

async with AsyncPlanVortex() as pv:
    page = await pv.accounts.list(org_id, limit=50)
    async for publication in pv.publications.aiterate(org_id, state=["ready"]):
        ...

Connecting an account

Connecting is the one flow the library cannot finish on its own: it ends with a person pressing "allow" on Instagram's page. What the library does is hand you a URL to send them to.

connection = pv.organizations.create_connect_token(org_id)
# connection["url"] is where the person goes. Never send them your client_secret.

person = pv.as_temporal_token(connection["token"])  # a client that can only do this
for link in person.accounts.connect_links(org_id):
    ...

Four things about that token, and each one bites separately: it lasts fifteen minutes, it is single-use, it is tied to one organization, and it cannot issue another one. Saving it for "next time" fails four different ways — issue a fresh one per connection, they are free.

And one that trips people without giving an error: branch on link["authorization"]["type"], never on the link. WhatsApp's is the empty string, because its sign-up is Meta's Embedded Signup popup and not an OAuth redirect; walking the list redirecting to link sends your user to your own page.

Accounts come back disabled and take no plan slot until pv.accounts.enable(...), and one authorization can leave several — a Facebook user with four pages is four of them.

Comments and messages

# The comments inbox comes out of PlanVortex's database: free, fast, and a photograph of the last
# time the network was read. The thread asks the network right then, and on X that costs credits.
for comment in pv.comments.iterate(org_id, unread=True, rating=[1, 2]):
    print(comment["rating"], comment["text"])

thread = pv.comments.thread(org_id, publication_id)
thread["credits_consumed"]  # real money on X, 0 everywhere else

# Before painting a button, ask what the network allows: they are not all the same.
if (pv.comments.actions_for("linkedin") or {}).get("hide"):
    ...

Errors

Errors are classified by code, never by the HTTP status — every domain error in this API travels with a 400. Each range has its own exception class, so you can catch a family without memorising numbers:

from planvortex import PlanLimitError, PlanVortexError

try:
    ...
except PlanLimitError as error:
    ...  # not fixed by retrying: fixed by changing plan
except PlanVortexError as error:
    error.code, error.family, error.message, error.data, error.status

The runtime list is PLANVORTEX_ERROR_RANGES. Two more that are not the API's answer: PlanVortexConnectionError (it never got there — retried already, on the methods where retrying is safe) and PlanVortexConfigError (something is wrong on this side, like a missing client_secret).

Webhooks

PlanVortex POSTs to your app when something happens: an account changed state, a message or a comment came in, an integration stopped working. Two things trip up everybody, so they go first.

The body is an array of changes, not an object. And the signature is computed over the raw body — if your framework already parsed the JSON and you serialise it again, the bytes are not the same ones and the signature never matches. The line that gives you the raw body is the only line of the recipe that changes:

# Flask
import os

from flask import request

from planvortex.webhooks import handle_webhook_request, is_comment_change


@app.post("/webhooks/planvortex")
def planvortex_webhook():
    changes = handle_webhook_request(
        body=request.get_data(),  # raw! never request.json
        headers=request.headers,
        secret=os.environ["PLANVORTEX_CLIENT_SECRET"],
    )
    for change in changes:
        if is_comment_change(change):
            moderate(change.get("commentObj"))
    return "", 200

# FastAPI
@app.post("/webhooks/planvortex")
async def planvortex_webhook(request: Request):
    changes = handle_webhook_request(
        body=await request.body(),  # raw! never the parsed model
        headers=request.headers,
        secret=os.environ["PLANVORTEX_CLIENT_SECRET"],
    )
    ...

# Django
@csrf_exempt
def planvortex_webhook(request):
    changes = handle_webhook_request(
        body=request.body,  # raw! never request.POST
        headers=request.headers,
        secret=os.environ["PLANVORTEX_CLIENT_SECRET"],
    )
    ...

handle_webhook_request raises WebhookSignatureError if the signature is missing or does not match (answer 401) and WebhookBodyError if the body is not what it has to be (answer 400). If you would rather do it in two steps, verify_webhook_signature(payload, signature, secret) returns a plain True/False and parse_webhook_body(payload) gives you the changes.

Narrow with the predicates — is_account_state_change, is_message_change, is_comment_change, is_integration_error_change — and let anything else fall through: the event list grows, and a field this release has never heard of is not an error.

PlanVortex does not retry a failed delivery. A 500 of yours loses the event, so if your work is slow, queue it and answer — and use pv.comments.list / pv.messages.list to catch up on anything you missed.

The whole API, in fourteen resources

pv.catalog · pv.clients · pv.organizations · pv.accounts · pv.uploads · pv.publications · pv.comments · pv.messages · pv.contacts · pv.products · pv.integrations · pv.ai_plans · pv.dashboard · pv.apps

That is 112 of the 112 operations the specification documents — everything except the 19 routes of roles and invitations, which are out of scope. A script walks the OpenAPI bundle on every test run and fails if a route is left without a method, so the sentence above stays true.

Examples

Five runnable scripts, each one the whole of its path and with a test of its own:

comments.py only reads unless you set PLANVORTEX_ALLOW_REPLY=1: replying is public, immediate, and reaches a person.

Links

• Reference
• API documentation
• Developers
• Node client
• Contributing · Security · Changelog

MIT © Talia Softworks