Predictions
A guide on how to work with predictions using the WriftAI Python client
This section demonstrates common prediction operations you can perform with the WriftAI Python client. These examples are not exhaustive — check the client reference for all options.
Get a Prediction by ID
prediction = wriftai.predictions.get("your-prediction-id")Cancel a Prediction by ID
wriftai.predictions.cancel("your-prediction-id")Create a Prediction
With the latest version of a model
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
)With a specific version of a model
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1:2",
input={
"prompt": "Summarize quantum computing.",
},
)With a webhook for prediction updates
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
webhook={
"url": "https://example.com/webhooks/wriftai",
"secret": "top-secret", # This is optional
},
)With file inputs
File inputs are automatically uploaded when you create a prediction. Wrap bytes,
file paths, or file-like objects in File, and no need to call wriftai.files.upload() separately.
Files can be nested anywhere in input, including inside lists, dicts, or combinations of
both, and you can include as many as you need. They'll all be uploaded concurrently.
With file paths:
from pathlib import Path
from wriftai import File
prediction = wriftai.predictions.create(
model="pyannote/speaker-diarization-community-1",
input={"audio": File(Path("/path/to/audio.wav"))},
)With raw bytes:
from wriftai import File
prediction = wriftai.predictions.create(
model="pyannote/speaker-diarization-community-1",
input={"audio": File(content=b"some audio bytes", mime_type="audio/wav")},
)With file-like objects:
from wriftai import File
with open("audio.wav", "rb") as f:
prediction = wriftai.predictions.create(
model="pyannote/speaker-diarization-community-1",
input={"audio": File(f)},
)For more details on working with files, see the Files guide.
With input validation enabled
Enable early input validation against the model’s input schema before a prediction is created. This catches invalid inputs upfront and prevents unnecessary model execution and cost.
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
validate_input=True,
)Create and wait for completion
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
wait=True,
)This returns a finished prediction. The server holds the connection open for its default wait time, and if the prediction hasn't finished by then, the client polls until it reaches a terminal state.
Create and wait with a custom duration
Set how long the server holds the initial request open before the client falls back to polling:
from wriftai.predictions import CreateWaitOptions
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
wait=True,
wait_options=CreateWaitOptions(duration=30),
)This still returns a finished prediction. duration only changes how long the
initial request stays open before polling starts. Use a lower value if your
environment has a short HTTP timeout, or a higher one (up to the server's
maximum) to get the result as soon as it's ready with fewer polling requests.
Create and wait without polling
Set poll=False to return as soon as the server's wait ends, even if the
prediction is still running:
from wriftai.predictions import CreateWaitOptions
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
wait=True,
wait_options=CreateWaitOptions(duration=6, poll=False),
)The returned prediction may not be in a terminal state, so check its status.
Create and wait with progress updates
from wriftai.predictions import CreateWaitOptions, PredictionWithDetails
def on_poll(prediction: PredictionWithDetails) -> None:
print(prediction["status"])
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
wait=True,
wait_options=CreateWaitOptions(poll_interval=2, on_poll=on_poll),
)poll_interval and on_poll only take effect while the client is polling.
They do not affect the initial wait.
Wait for an existing prediction to complete
prediction = wriftai.predictions.wait("your-prediction-id")This is also what create(wait=True) uses to poll when the prediction hasn't
finished after the server's wait, so calling it directly is useful when you want
to skip straight to polling, or you already have a prediction ID from somewhere
else.
It takes the same poll_interval and on_poll options shown above, via
WaitOptions instead of CreateWaitOptions:
from wriftai.predictions import WaitOptions, PredictionWithDetails
def on_poll(prediction: PredictionWithDetails) -> None:
print(prediction["status"])
prediction = wriftai.predictions.wait(
"your-prediction-id",
options=WaitOptions(poll_interval=2, on_poll=on_poll),
)Stream a Prediction's events
A prediction's events can be streamed while it runs. You can stream them straight
from the create call, or create the prediction first and stream its events by ID.
Either way, the prediction has to be created with stream=True.
Streaming stops on its own once the prediction sends a terminal event, so there's nothing extra to handle. If the server closes an idle connection before then, the client reconnects automatically from the last event it received, so you don't miss or repeat events.
Create a prediction and stream its events in one call
Set wait=True together with stream=True to get the events directly from the
create request. This returns the events instead of the prediction:
for event in wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
wait=True,
stream=True,
):
print(event)With the async client, async_create returns the events after you await it, so
iterate with async for event in await wriftai.predictions.async_create(...).
Create a prediction, then stream its events by ID
With stream=True but without wait=True, create returns the prediction as
usual, and you stream its events afterwards with the stream() method, which
takes the prediction's ID:
prediction = wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={
"prompt": "Summarize quantum computing.",
},
stream=True,
)
for event in wriftai.predictions.stream(prediction["id"]):
print(event)With the async client, use async_stream and iterate with async for.
The sections below use the stream() method, which takes a prediction's ID.
The same applies when streaming from create(), apart from the differences
noted in each section.
Handle events by type
from wriftai.predictions import PredictionStreamEventType
for event in wriftai.predictions.stream("your-prediction-id"):
if event["event"] == PredictionStreamEventType.OUTPUT:
print(event)Start from a specific event
Pass after_id to only receive events after a given event ID. Without it, the
stream starts from the first event.
for event in wriftai.predictions.stream("your-prediction-id", after_id="event-id"):
print(event)create() has no after_id, since it starts from the first event.
Set how long the server waits for events
By default, the server decides how long an idle connection stays open between
events before closing it, and the client reconnects automatically when that
happens. This is usually fine, but you can set it with the wait argument of
stream(), in seconds. Use a lower value if something in your environment, like
a proxy or load balancer, closes idle connections sooner than the server does, or
a higher one to reconnect less often when there are long gaps between events:
for event in wriftai.predictions.stream("your-prediction-id", wait=10):
print(event)This is the longest the server waits between events, not how long to wait for the prediction to finish. After an idle close, the client reconnects with the same value.
create() takes the same setting as duration in wait_options, since its
wait argument is a boolean:
from wriftai.predictions import CreateWaitOptions
for event in wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={"prompt": "Summarize quantum computing."},
wait=True,
stream=True,
wait_options=CreateWaitOptions(duration=10),
):
print(event)When streaming from create(), poll, poll_interval and on_poll are
ignored, since the client never polls.
Resume after a network error
Streaming connections can stay open for a long time, so they can drop before the
stream ends, for example because of a network issue on your side. Opening the
connection is retried on transient errors, like any other request, but if the
connection drops after events have started arriving, the error is raised rather
than retried. To pick up where you left off, call stream() again with after_id
set to the ID of the last event you received:
import httpx
last_id = None
try:
for event in wriftai.predictions.stream("your-prediction-id"):
last_id = event["id"]
print(event)
except httpx.TransportError:
# Continue from the last event received.
for event in wriftai.predictions.stream("your-prediction-id", after_id=last_id):
print(event)The first event of a prediction's stream is an init event, and its data contains
the prediction's ID (event["data"]["id"]). When streaming from create(),
that's how you get the ID to resume with stream() the same way. If the error is
raised before the first event, there is no ID yet, so there is nothing to resume
from:
import httpx
from wriftai.predictions import PredictionStreamEventType
prediction_id = None
last_id = None
try:
for event in wriftai.predictions.create(
model="deepseek-ai/deepseek-r1",
input={"prompt": "Summarize quantum computing."},
wait=True,
stream=True,
):
if event["event"] == PredictionStreamEventType.INIT:
prediction_id = event["data"]["id"]
last_id = event["id"]
print(event)
except httpx.TransportError:
if prediction_id is None:
raise # No prediction ID yet, so there is nothing to resume from.
# Continue from the last event received.
for event in wriftai.predictions.stream(prediction_id, after_id=last_id):
print(event)Stop streaming early
Leaving the for statement early, for example with break, keeps the connection
open until the stream is garbage collected. Wrap the stream in
contextlib.closing to release the connection immediately:
from contextlib import closing
with closing(wriftai.predictions.stream("your-prediction-id")) as events:
for event in events:
print(event)
break # Stop after the first event.With the async client, use contextlib.aclosing instead:
from contextlib import aclosing
async with aclosing(wriftai.predictions.async_stream("your-prediction-id")) as events:
async for event in events:
print(event)
break # Stop after the first event.The same works for create(): wrap the call in closing, or, with the async
client, wrap the awaited call in aclosing, for example
aclosing(await wriftai.predictions.async_create(...)).
List Predictions
predictions = wriftai.predictions.list()Filter by status
from wriftai.predictions import PredictionPaginationOptions, Status
predictions = wriftai.predictions.list(
PredictionPaginationOptions(
statuses=[Status.cancelled, Status.failed],
)
)