Python SDK
The socaity package is the Python client for the Socaity platform. It exposes a typed class per model, returns a job handle from every call, and routes Replicate-backed models through the Socaity backend so you only manage one API key.
pip install socaity Requires Python ≥ 3.10. media-toolkit is installed as a dependency for file handling. There is no separate CLI; pip install socaity is the only install step.
Pass your Socaity key explicitly when you construct a model class. Read it from the environment with os.getenv("SOCAITY_API_KEY"). Constructing a model without an api_key argument fails with Unauthorized unless you have stored credentials from socaity login.
import os
from socaity.sdk.replicate.black_forest_labs import flux_schnell
from socaity.sdk.replicate.deepseek_ai import deepseek_v3
# Both are Replicate-backed catalog models, routed through the Socaity backend.
# Pass your Socaity key explicitly when you construct each model class.
key = os.getenv("SOCAITY_API_KEY")
flux = flux_schnell(api_key=key)
llm = deepseek_v3(api_key=key)flux_schnell, deepseek_v3) and routes through the Socaity backend. A single SOCAITY_API_KEY covers all of them, and you never set REPLICATE_API_KEY yourself. The env var name is REPLICATE_API_KEY (not REPLICATE_API_TOKEN) and is only read if you bypass the Socaity backend with a direct-to-Replicate service address. Socaity's open-source services such as face2face and speechcraft are self-hosted via APIPod and are not part of the hosted catalog. Each model is a typed class. Every method submits the job immediately and returns a handle; call .get_result() on the handle to block until the model finishes. Hosted catalog models are Replicate-backed and live under socaity.sdk.replicate.<vendor>. A few common stubs ship with the wheel (for example flux_schnell); for any other model, run socaity install <vendor>/<model> once to generate its stub before importing.
| Import Path | Service | Category |
|---|---|---|
socaity.sdk.replicate.black_forest_labs | flux_schnell | Image |
socaity.sdk.replicate.deepseek_ai | deepseek_v3 | Text |
socaity.sdk.replicate.tencentarc | gfpgan | Image |
socaity.sdk.replicate.jaaari | kokoro_82m | Audio |
Parameters for flux_schnell.predictions(). The same call is reachable via flux_schnell.run(...) or by invoking the instance directly (flux_schnell()(...)).
| Parameter | Type | Default | Description |
|---|---|---|---|
prompt | str | required | Text description of the image to generate. |
num_outputs | int | 1 | Number of images to generate. |
aspect_ratio | str | '1:1' | Output aspect ratio, e.g. "16:9", "4:3". |
output_quality | int | 80 | JPEG/WebP quality 0-100. |
output_format | str | 'webp' | Output image format: "webp", "jpg", or "png". |
megapixels | str | '1' | Approximate output resolution in megapixels. |
num_inference_steps | int | 4 | Number of diffusion steps. Higher values trade speed for detail. |
go_fast | bool | True | Use the faster inference path. Disables seed-based reproducibility when True. |
disable_safety_checker | bool | False | Bypass the safety filter. Use responsibly. |
seed | int | None | 42 | Random seed for reproducibility. No effect when go_fast=True. |
Construct the model, submit a call, and block on the handle:
import os
from socaity.sdk.replicate.black_forest_labs import flux_schnell
flux = flux_schnell(api_key=os.getenv("SOCAITY_API_KEY"))
# flux(...) submits the job and returns a Job handle immediately.
# .get_result() blocks until the job finishes (default total timeout: 3600s).
# On timeout, .get_result() returns None; it does not raise.
image = flux(
prompt="a lone astronaut on a neon planet, cinematic",
num_outputs=1,
).get_result()
# num_outputs=1 returns a single ImageFile; pass num_outputs>1 for a list.
image.save("astronaut.png")Since submitting never blocks, fanning out many jobs is a plain loop. Block on each handle only when you need the output.
import os
from socaity.sdk.replicate.black_forest_labs import flux_schnell
flux = flux_schnell(api_key=os.getenv("SOCAITY_API_KEY"))
prompts = ["a forest", "a city", "an ocean"]
# Each flux(...) call submits the job and returns immediately, so the
# loop fans out three jobs in parallel before we wait on any of them.
jobs = [flux(prompt=p, num_outputs=1) for p in prompts]
# Collect results in submission order. Each .get_result() blocks on its own job.
# Each job uses num_outputs=1, so each result is a single ImageFile.
results = [job.get_result() for job in jobs]
for i, img in enumerate(results):
img.save(f"output_{i}.png")The SDK accepts local paths, URLs, bytes, PIL Images, and numpy arrays. The client uploads local files for you before submitting the job.
import os
from socaity.sdk.replicate.tencentarc import gfpgan
restore = gfpgan(api_key=os.getenv("SOCAITY_API_KEY"))
# Local paths are uploaded for you before the job is submitted.
result = restore.predictions(
img="./portrait.jpg",
scale=2,
).get_result()
result.save("restored.png")
# The img parameter also accepts: an https:// URL, raw bytes,
# a PIL.Image, or a numpy array.Every model call returns a Job object. The handle exposes status, the parsed response, progress, and cancellation. The polling loop runs once per second with a 3,600-second total timeout. The loop tolerates up to three consecutive polling errors before raising.
| Method / Property | Returns | Description |
|---|---|---|
.get_result(timeout_s=None) | Any | Block until the job reaches a terminal state, then return the parsed result. Returns None on timeout (does not raise). Re-raises TaskException on FAILED and TaskCancelledException on CANCELLED. |
.response | BaseJobResponse | None | Latest parsed response from the backend. Exposes .id, .status, .progress, .error, .result, and provider-specific fields. |
.response.status | APIJobStatus | One of PENDING, QUEUED, PROCESSING, STREAMING, FINISHED, FAILED, TIMEOUT, CANCELLED, UNKNOWN. |
.is_terminal | bool | True once status is FINISHED, FAILED, TIMEOUT, or CANCELLED. |
.runtime_info | tuple[float, float] | None | Returns (delay_seconds, execution_seconds) for RunPod and Replicate jobs. None otherwise. The SDK does not surface per-call cost. |
.cancel(wait=False, timeout_s=30.0, poll_interval_s=0.5) | BaseJobResponse | None | Request cancellation. Issues a remote cancel via cancel_job_url when available; otherwise cancels locally. |
Every model class today exposes method-level aliases for its primary endpoint. They are defined on the Python class itself, so they resolve before any network call is made. For deepseek_v3, predictions is the canonical method and run and __call__ both forward to it.
import os
from socaity.sdk.replicate.deepseek_ai import deepseek_v3
llm = deepseek_v3(api_key=os.getenv("SOCAITY_API_KEY"))
# Canonical endpoint method.
llm.predictions(prompt="hello").get_result()
# run() is an alias for the canonical method.
llm.run(prompt="hello").get_result()
# __call__ also forwards to predictions, so the instance is callable.
llm(prompt="hello").get_result() The pattern is the same across the catalog: a canonical endpoint method, plus run and __call__ pointing at it. flux_schnell.predictions aliases to flux_schnell.run and flux_schnell(), the same way deepseek_v3.predictions aliases to deepseek_v3.run and deepseek_v3().
socaity.sdk.replicate.<vendor>.