aishorty

The official Python SDK for the Shorty public API (https://aishorty.com/v1). Transcribe media, summarize videos and web pages, generate subtitles, and poll async jobs — with typed errors, automatic retries, cursor pagination, and Standard-Webhooks signature verification.

• Sync and async. Shorty and AsyncShorty share one implementation; only the transport differs.
• Fully typed. Ships py.typed; pydantic models for every response.
• Two dependencies. httpx and pydantic, nothing else.
• Python 3.11+.
• Server-side only — your API key is a secret and must never reach a browser or a mobile app.

Install

pip install aishorty
# or: uv add aishorty  /  poetry add aishorty

Quickstart

from aishorty import Shorty

client = Shorty(api_key="shk_live_...")  # or set SHORTY_API_KEY

# Start a summary job for a YouTube video…
result = client.summaries.create(
    "youtube",
    url="https://www.youtube.com/watch?v=dQw4w9WgXcQ",
)

# …then block until it finishes.
if result.job_id:
    job = client.jobs.wait(result.job_id)
    print(job.status, job.output)

The async client is identical, with await:

import asyncio
from aishorty import AsyncShorty


async def main():
    async with AsyncShorty() as client:
        usage = await client.usage.get()
        print(usage.plan.plan_name, usage.limits.conversion.max_file_size_label)


asyncio.run(main())

Authentication

Pass api_key= explicitly, or set the SHORTY_API_KEY environment variable. The constructor raises ValueError if neither is present — the SDK never makes an unauthenticated request to discover the problem at runtime.

client = Shorty(api_key="shk_live_...")
client = Shorty()  # reads SHORTY_API_KEY

Your key is held inside the transport, never as an attribute on the client, and repr(client) prints only the base URL. Logging a client cannot leak the secret.

Client options

Option	Default	Meaning
api_key	$SHORTY_API_KEY	Your shk_live_... key.
base_url	https://aishorty.com	API origin; paths already carry /v1.
timeout	60.0	Per-attempt timeout, in seconds.
max_retries	2	Retries after the first attempt.
http_client	—	Your own httpx.Client / httpx.AsyncClient.
debug	False	True logs one redacted line per attempt to stderr; pass a callable to route it elsewhere.

Both clients are context managers, which is the tidiest way to close the connection pool:

with Shorty() as client:
    ...

async with AsyncShorty() as client:
    ...

Resources

Call	HTTP	Returns
client.usage.get()	GET /v1/usage	UsageResponse
client.articles.list(cursor=…, limit=…)	GET /v1/articles	SyncPage[Article]
client.articles.search(q, limit=…, article_type=…)	GET /v1/articles/search	ArticleSearchResult
client.articles.get(article_id)	GET /v1/articles/{id}	ArticleDetail
client.transcriptions.list(cursor=…, limit=…)	GET /v1/transcriptions	SyncPage[Transcription]
client.transcriptions.create(url, language=…)	POST /v1/transcriptions	JobAccepted
client.transcriptions.get(transcription_id)	GET /v1/transcriptions/{id}	TranscriptionDetail
client.summaries.create(source, url=…, content=…)	POST /v1/summaries	SummaryAccepted | SummaryAlreadyComplete
client.subtitles.create(url, style=…)	POST /v1/subtitles	JobAccepted
client.subtitles.download(job_id, kind=…)	GET /v1/subtitles/{id}/download	SubtitleDownload
client.jobs.get(job_id)	GET /v1/jobs/{id}	JobStatus
client.jobs.wait(job_id)	(client-side polling loop)	JobStatus

Summaries return one of two shapes

summaries.create answers 202 with a started job, or 200 when that source was already summarized. The SDK picks the variant from the status code:

from aishorty import SummaryAlreadyComplete

result = client.summaries.create("url", url="https://example.com/post")

if isinstance(result, SummaryAlreadyComplete):
    article = client.articles.get(result.article_id)  # nothing to poll
elif result.job_id:
    job = client.jobs.wait(result.job_id)
else:
    # Accepted, but no pollable job — the article id is the handle.
    article = client.articles.get(result.article_id)

Pagination

Collections are cursor-paginated. Iterating a page walks every item across all pages, fetching lazily:

for article in client.articles.list(limit=50):
    print(article.id, article.title)

Async iteration mirrors it:

async for article in await client.articles.list(limit=50):
    print(article.id)

Work page by page when you want the boundaries:

page = client.articles.list(limit=50)
while page is not None:
    print(len(page.data), "rows;", page.has_more, "more")
    page = page.next_page()

The cursor is filter-bound server-side, so the SDK resends the original parameters on every fetch and advances only the cursor. Never hand a cursor from one filter to a call with a different one.

Jobs

transcriptions.create, subtitles.create, and summaries.create are asynchronous: they return once the job is accepted. jobs.wait() is a client-side polling loop over jobs.get().

job = client.transcriptions.create("https://example.com/audio.mp3")
final = client.jobs.wait(job.job_id, poll_interval=2.0, timeout=600)
print(final.status, final.output)

It raises JobFailedError if the job ends in error/cancelled, and APITimeoutError if the deadline elapses while the job is still running. Status strings have several spellings (success / completed / done); use normalize_job_status() rather than comparing raw strings.

Note tracking_url on an accepted job is a relative path, not an absolute URL.

Errors

Every non-2xx response raises a typed error, chosen by the problem code first and falling back to the HTTP status class — so a new server code never breaks your except handling.

from aishorty import APIError, QuotaExhaustedError, RateLimitError

try:
    client.usage.get()
except QuotaExhaustedError:
    ...  # period allowance spent — back off until the next window
except RateLimitError as err:
    print("retry after", err.retry_after_seconds, "s")
except APIError as err:
    print(err.status, err.code, err.title, err.detail, err.request_id)
    print(err.errors)  # field-level validation failures, when present

Every APIError carries status, code, problem_type, title, detail, request_id, errors, headers, retry_after_seconds, and rate_limit. The message reads title: detail (request_id=…) and never contains your API key.

Not every failure is a well-formed problem document — a gateway can return HTML, and a disabled feature can answer with a bare JSON error. Those still raise a typed error with a bounded, redacted snippet in detail rather than a parse crash.

Error code reference

Code	Status	SDK error class
unauthorized	401	AuthenticationError
invalid_api_key	401	AuthenticationError
insufficient_scope	403	PermissionDeniedError
feature_not_enabled	403	PermissionDeniedError
resource_not_found	404	NotFoundError
idempotency_conflict	409	ConflictError
idempotency_in_progress	409	ConflictError
resource_not_ready	409	ConflictError
validation_failed	400	ValidationError
idempotency_key_reused	422	ValidationError
request_too_large	413	ValidationError
rate_limited	429	RateLimitError
quota_exhausted	429	QuotaExhaustedError
internal_error	500	APIServerError
service_unavailable	503	APIServerError

Transport failures raise APIConnectionError, or APITimeoutError for a per-attempt timeout. Both derive from ShortyError, as does everything above.

Rate limits

Every response and every APIError exposes the server's rate-limit state, folded from whichever headers arrived (IETF draft-11 or the X-RateLimit-* trio):

usage = client.usage.get()
if usage.rate_limit:
    print(usage.rate_limit.remaining, "left until", usage.rate_limit.reset_at)
print(usage.request_id)  # quote this in a support request

reset_at is always an absolute unix timestamp, whichever header form the server used. Every field is optional — responses that fail before the limiter runs (a pre-auth 401) carry no rate-limit headers at all, and rate_limit is then None.

Retries & idempotency

Transient failures are retried up to max_retries times (default 2) after the first attempt.

Retried: connection errors · per-attempt timeouts · 408 · 429 rate_limited (or a 429 with no parseable code) · 500 / 502 / 503 / 504.

Never retried: 429 quota_exhausted (a spent allowance — retrying cannot help) · any other 4xx.

Backoff: Retry-After is honored when present (both delta-seconds and HTTP-date forms, capped at 60s); otherwise full-jitter exponential backoff, random(0, min(0.5 · 2**attempt, 8)) seconds.

POST safety: a POST is retried only when it carries an Idempotency-Key. The three create methods generate one automatically and reuse the same key across every attempt, so a retry is deduplicated server-side rather than duplicating work. Pass your own to make a call idempotent across process restarts:

client.summaries.create("text", content="…", idempotency_key="my-stable-key-123")

When the server replays a previous result, the returned model reports it:

result = client.summaries.create("text", content="…", idempotency_key="k-1")
if result.was_replayed:
    print("no new work was started")

Webhooks

Verify inbound Shorty webhooks with verify_webhook_signature. It implements the Standard Webhooks v1 scheme: HMAC-SHA256 over webhook-id / webhook-timestamp / webhook-signature, rotation via multiple space-separated signatures, a 5-minute replay tolerance, and a constant-time comparison.

from aishorty import verify_webhook_signature


@app.post("/webhooks/shorty")
async def receive(request: Request):
    raw = await request.body()  # RAW bytes — see the rule below
    if not verify_webhook_signature(raw, request.headers, WEBHOOK_SECRET):
        return Response(status_code=400)
    event = json.loads(raw)
    ...

Use verify_webhook_signature_or_raise when you want the reason in your logs; it raises WebhookVerificationError whose .reason is one of missing_headers, malformed_timestamp, timestamp_out_of_tolerance, or no_matching_signature.

Raw-body rule (load-bearing). Verify the exact bytes you received. Parsing the JSON and re-serializing it reorders keys and changes spacing, and verification will fail. Read the raw body before any JSON middleware touches it: await request.body() (FastAPI), request.get_data() (Flask), request.body (Django).

Type checking

The package ships py.typed, so mypy, pyright, and ty see full types with no stubs. Response parsing is deliberately lenient: unknown fields the server adds later are preserved rather than rejected, and open-vocabulary fields (status, article_type, tier, kind) are typed str so a new server value cannot turn a working call into a validation error. Literal aliases (Tier, SubtitleKind, SubtitleStyle, SummarySource) are exported for the request side, where the set really is closed.

Timestamps are ISO-8601 UTC strings; the SDK does not parse them into datetime in 0.1.

License

MIT © Devino Solutions