> ## Documentation Index
> Fetch the complete documentation index at: https://docs.simular.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python

> The loop with requests, and a streaming variant with sseclient.

## Polling with `requests`

```bash theme={null}
pip install requests
```

```python theme={null}
import os
import time
import webbrowser

import requests

API = os.environ.get("SAI_API_URL", "https://api.simular.ai")
HEADERS = {"Authorization": f"Bearer {os.environ['SAI_API_KEY']}"}


def start(task: str, machine_id: str | None = None) -> str:
    body = {"message": task, "wait": False}
    if machine_id:
        body["machineId"] = machine_id
    r = requests.post(f"{API}/v1/agents/message", json=body, headers=HEADERS, timeout=30)
    if r.status_code == 429:
        raise RuntimeError(r.json()["error"])
    r.raise_for_status()
    return r.json()["sessionId"]


def approve(approval_id: str, response: str = "yes", selections=None) -> None:
    body = {"approvalId": approval_id, "response": response}
    if selections is not None:
        body["selections"] = selections
    r = requests.post(f"{API}/v1/agents/approve", json=body, headers=HEADERS, timeout=30)
    r.raise_for_status()


def run(task: str) -> str:
    session_id = start(task)
    cursor = None
    while True:
        params = {"sessionId": session_id, "wait": 30}
        if cursor:
            params["since"] = cursor
        r = requests.get(f"{API}/v1/agents/events", params=params, headers=HEADERS, timeout=90)
        if r.status_code == 429:
            time.sleep(60)
            continue
        r.raise_for_status()
        page = r.json()
        cursor = page["cursor"]

        for ev in page["events"]:
            if ev["type"] == "data-progress":
                print("  ", ev["data"]["text"])

        status = page["status"]
        if status == "running":
            continue
        if status == "needs_approval":
            data = page["approval"]
            if data["isLinkOnly"]:
                print("Open in a browser:", data["approvalUrl"])
                webbrowser.open(data["approvalUrl"])
                time.sleep(15)
            elif data["approvalType"] == "choice":
                picks = [[q["options"][0]["value"]] for q in data["questions"]]
                approve(data["approvalId"], "yes", picks)
            else:
                print("Approving:", data.get("command", data["title"]))
                approve(data["approvalId"], "yes")
            continue
        if status == "idle":
            return page["text"]
        raise RuntimeError(
            next((e["errorText"] for e in page["events"] if e["type"] == "error"), "task failed")
        )


if __name__ == "__main__":
    print(run("Open Notepad, type hello, save it to the desktop as hello.txt"))
```

The choice handler above picks the first option. Replace it with your own logic. To stop a task early, `POST /v1/agents/abort` with `{"sessionId": session_id}`.

## Uploading a file

```python theme={null}
from pathlib import Path


def upload(path: str) -> dict:
    p = Path(path)
    r = requests.post(
        f"{API}/v1/agents/upload",
        data=p.read_bytes(),
        headers={**HEADERS, "x-filename": p.name},
        timeout=120,
    )
    r.raise_for_status()
    return r.json()


attachment = upload("report.pdf")
requests.post(
    f"{API}/v1/agents/message",
    json={"message": "Summarise the attached PDF", "wait": False, "attachments": [attachment]},
    headers=HEADERS,
    timeout=30,
)
```

## Streaming with `sseclient-py`

Use this from a long-lived process where holding a connection for minutes is fine.

```bash theme={null}
pip install requests sseclient-py
```

```python theme={null}
import json
import os

import requests
import sseclient

API = os.environ.get("SAI_API_URL", "https://api.simular.ai")
HEADERS = {
    "Authorization": f"Bearer {os.environ['SAI_API_KEY']}",
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
}

resp = requests.post(
    f"{API}/v1/agents/message",
    json={"message": "What is on the screen right now?"},
    headers=HEADERS,
    stream=True,
    timeout=(30, 600),
)
resp.raise_for_status()

text = []
for event in sseclient.SSEClient(resp).events():
    if event.data == "[DONE]":
        break
    frame = json.loads(event.data)
    kind = frame["type"]
    if kind == "text-delta":
        text.append(frame["delta"])
    elif kind == "data-approval-request":
        print("approval:", frame["data"]["approvalId"], frame["data"]["isLinkOnly"])
    elif kind == "error":
        raise RuntimeError(frame["errorText"])

print("".join(text))
```

Approvals on the streaming path are answered with `POST /approve` from another thread or process; the stream stays open while it waits.