Python SDK

Generate specifications and run browser tasks with the published predev-api Python package.

Node.js

npm install predev-api

Python 3.8+ is supported. This guide targets published 1.1.0.

Generate and wait for a specification

import os
import time
from predev_api import PredevAPI

client = PredevAPI(api_key=os.environ["PREDEV_API_KEY"])
job = client.fast_spec_async(
    input_text="Build a team task manager with projects and assignees.",
    current_context="Use TypeScript and preserve organization-scoped access.",
)
print("Save this ID:", job["specId"])
deadline = time.monotonic() + 600

while time.monotonic() < deadline:
    spec = client.get_spec_status(job["specId"])
    if spec["status"] == "completed":
        print(spec.get("codingAgentSpecMarkdown", "No Markdown returned"))
        break
    if spec["status"] == "failed":
        raise RuntimeError(spec.get("errorMessage", "Specification failed"))
    time.sleep(5)
else:
    raise TimeoutError("Polling deadline reached; retrieve the saved spec ID later.")

Specification, status, search, list, and browser methods return dictionaries at runtime, despite dataclass return annotations in the package. Use job["specId"] and spec["status"], not job.specId or spec.status. get_credits_balance() is the exception: it constructs a CreditsBalanceResponse object.

Method reference

Method	Parameters	Runtime return
fast_spec(...)	input_text, optional current_context, doc_urls, file	Completed specification dictionary
deep_spec(...)	Same	Completed specification dictionary
fast_spec_async(...)	Same	Dictionary with specId, status
deep_spec_async(...)	Same	Dictionary with specId, status
get_spec_status(spec_id)	Specification ID	Specification dictionary
list_specs(...)	Optional limit, skip, endpoint, status	Dictionary with specs, total, hasMore
find_specs(query, ...)	Required regex query; same filters	Dictionary with specs, total, hasMore
get_credits_balance()	None	CreditsBalanceResponse object
browser_agent(tasks, ...)	Task list; optional concurrency, stream=False, run_async=False	Run dictionary, or an iterator when stream=True
get_browser_agent(id, ...)	Run ID; optional include_events=False	Run dictionary
list_browser_agents(...)	Optional limit, skip, status	Dictionary with batches, total, hasMore
browser_agent_status()	None	Dictionary with your queue and cap

*_async methods and run_async=True submit server-side work and return its ID; they are ordinary blocking Python calls, not coroutines. Do not await them. Spec generation calls use a 300-second HTTP timeout and status, list, and search calls 60 seconds. Browser submissions use 300 seconds, retrieval and list 60 seconds, and queue status 30 seconds. A polling deadline is separate from an individual HTTP timeout, and there is no client-wide timeout or retry option.

Specification history and credits

page = client.find_specs(query="report|export", status="completed", limit=20)
for summary in page["specs"]:
    print(summary["_id"], summary.get("input"))

balance = client.get_credits_balance()
print(balance.creditsRemaining)

List and search return summaries. Fetch the ID with get_spec_status for full Markdown, JSON, and graphs. See inputs and outputs.

File uploads

The SDK accepts a file path or binary file object via file. In 1.1.0, it does not set an explicit file-part MIME type or JSON-encode multipart doc_urls. Use direct HTTP when uploading reference documents, especially when supplying documentation URLs:

import json
import requests

with open("requirements.pdf", "rb") as document:
    response = requests.post(
        "https://api.pre.dev/fast-spec",
        headers={"Authorization": "Bearer " + os.environ["PREDEV_API_KEY"]},
        data={
            "input": "Create an implementation plan from these requirements.",
            "docURLs": json.dumps(["https://www.postgresql.org/docs/current/"]),
            "async": "true",
        },
        files={"file": ("requirements.pdf", document, "application/pdf")},
        timeout=60,
    )
    response.raise_for_status()
    job = response.json()
print(job["specId"])

See accepted formats and limits.

Run browser tasks and poll

tasks = [{
    "url": "https://example.com",
    "instruction": "Extract the main page heading.",
    "output": {
        "type": "object",
        "properties": {"heading": {"type": "string"}},
        "required": ["heading"],
    },
    "mode": "extract",
    "maxSteps": 10,
    "maxDurationSeconds": 120,
}]
browser_job = client.browser_agent(tasks, run_async=True)
print("Save this run ID:", browser_job["id"])
run_deadline = time.monotonic() + 600

while time.monotonic() < run_deadline:
    run = client.get_browser_agent(browser_job["id"])
    if run["status"] in ("completed", "failed"):
        for task in run["results"]:
            if task and task.get("status") == "SUCCESS":
                print(task.get("data"))
            else:
                print("Task failed:", (task or {}).get("error", "No result"))
        break
    time.sleep(5)
else:
    raise TimeoutError("Polling deadline reached; retrieve the saved run later.")

The SDK passes task dictionaries through to REST, including the current per-task controls such as mode, maxSteps, and maxDurationSeconds.

Stream a new browser submission

saw_done = False
for message in client.browser_agent(tasks, stream=True):
    if message["event"] == "task_event":
        print(message["data"]["taskIndex"], message["data"]["type"])
    elif message["event"] == "done":
        saw_done = True
        final = message["data"]
        print(final["status"], final["totalCreditsUsed"])
if not saw_done:
    raise RuntimeError("The stream ended before a final result arrived.")

This call creates a new run. Streamed error frames raise SDK exceptions. A clean iterator exit without done still requires recovery; the SDK does not make EOF a terminal success. To attach to a saved async ID, use existing-run SSE over direct HTTP.

Browser history and evidence

queue = client.browser_agent_status()
print(queue["total"], queue["cap"])

runs = client.list_browser_agents(limit=20, status="completed")
for summary in runs["batches"]:
    print(summary["id"], summary["totalCreditsUsed"])

evidence = client.get_browser_agent(browser_job["id"], include_events=True)

include_events=True adds available timelines and screenshots. Routine polling should omit it. Status and results explains pending stubs, null slots, uppercase task statuses, and lowercase batch statuses.

Errors

from predev_api import (
    AuthenticationError,
    InsufficientCreditsError,
    PredevAPIError,
    QueueFullError,
    RateLimitError,
)

try:
    page = client.list_specs(limit=20)
except AuthenticationError:
    print("Check the API key and account access.")
except RateLimitError:
    print("Back off before retrying.")
except PredevAPIError as error:
    print(str(error))

try:
    browser_job = client.browser_agent(tasks, run_async=True)
except InsufficientCreditsError as error:
    print(str(error), error.action_url)
except QueueFullError:
    print("Wait for tasks to finish before submitting more work.")
except PredevAPIError as error:
    print(str(error))

The base exception has no status_code attribute. Browser gate errors have typed subclasses; see errors and retries. An Architect 402 without a machine-readable browser gate code raises PredevAPIError. Use direct REST when a retry needs an idempotency key; the SDK has no idempotency or arbitrary-header option.

See the PyPI package or the SDK source.