# Quick start

Start with a text-only conversation: no tools, MCP connections or sandbox. You need Python 3, a human-provided platform API key and API base URL (including `/v1`), and either an Anthropic API key with access to `anthropic/claude-sonnet-5-5` or an OpenAI API key with access to `openai/gpt-6.1-sol`. Set `PLATFORM_API_BASE_URL`, `PLATFORM_API_KEY` and `ANTHROPIC_API_KEY` or `OPENAI_API_KEY` in your backend environment, save the following as `quick_start.py`, and run `python3 quick_start.py`. It uses Anthropic when `ANTHROPIC_API_KEY` is set, otherwise OpenAI. The keys stay in your backend; never put them in browser code.

The sequence is: store a model key → create an agent → create an empty session → open its event stream → submit input → read authoritative messages and the run outcome → send a follow-up in the same session. Every returned ID is captured from the actual response. Open the stream before submitting input so previews can cover the new message. This example consumes `message.completed`; previews are optional and are not authoritative.

```python
import json
import os
import urllib.request

base = os.environ["PLATFORM_API_BASE_URL"].rstrip("/")
headers = {"Authorization": "Bearer " + os.environ["PLATFORM_API_KEY"]}

def request(method, path, body=None, expected=200):
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(base + path, data=data, method=method,
        headers={**headers, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as response:
        if response.status != expected:
            raise RuntimeError(f"Unexpected HTTP status: {response.status}")
        return json.load(response)

def events(response):
    data = []
    for raw in response:
        line = raw.decode("utf-8").rstrip("\r\n")
        if not line:
            if data:
                yield json.loads("\n".join(data))
            data = []
        elif line.startswith("data:"):
            data.append(line[5:].lstrip(" "))

provider = "anthropic" if os.environ.get("ANTHROPIC_API_KEY") else "openai"
default_model = {"anthropic": "anthropic/claude-sonnet-5-5", "openai": "openai/gpt-6.1-sol"}[provider]
key = request("POST", "/model-api-keys", {
    "name": "Quick start " + provider, "provider": provider,
    "key": os.environ[provider.upper() + "_API_KEY"],
}, expected=201)
print("model_api_key_id=" + key["id"])
agent = request("POST", "/agents", {"name": "Chat assistant", "config": {
    "model": os.environ.get("MODEL", default_model),
    "model_api_key_id": key["id"],
    "instructions": "Answer the user's questions helpfully.", "tools": [],
}}, expected=201)
print("agent_id=" + agent["id"])
session = request("POST", "/sessions", {
    "agent": {"id": agent["id"], "version": agent["latest_version"]},
    "subject_id": "example/user_123", "environment": {"type": "none"},
}, expected=201)
sid = session["id"]
print("session_id=" + sid)
stream_request = urllib.request.Request(
    base + f"/sessions/{sid}/events/stream?include_deltas=true",
    headers={**headers, "Accept": "text/event-stream"})
with urllib.request.urlopen(stream_request, timeout=60) as stream:
    frames = events(stream)
    for prompt in ["Hello! What can you help me with?", "What did I just ask you?"]:
        receipt = request("POST", f"/sessions/{sid}/inputs", {
            "mode": "queue", "content": [{"type": "text", "text": prompt}],
        }, expected=202)
        print("input_id=" + receipt["id"])
        run_id = receipt["run_id"]
        for event in frames:
            kind = event["type"]
            if kind == "message.delta":
                continue  # Previews have a different shape; wait for authoritative text.
            data = event["data"]
            if kind == "input.cancelled" and data["input"]["id"] == receipt["id"]:
                raise RuntimeError(f"Input {receipt['id']} cancelled: {data['input']['cancellation_reason']}")
            if kind.startswith("run.") and data["run"]["initial_input_id"] == receipt["id"]:
                run_id = data["run"]["id"]
            if kind == "input.applied" and data["input"]["id"] == receipt["id"]:
                run_id = data["input"]["run_id"]
            if kind == "message.completed" and data["run_id"] == run_id:
                print("assistant=" + "".join(block["text"] for block in
                    data["message"]["content"] if block["type"] == "text"))
            if kind in {"run.completed", "run.failed", "run.interrupted", "run.stopped"}:
                if data["run"]["id"] != run_id:
                    continue
                run = request("GET", f"/sessions/{sid}/runs/{run_id}")
                if run["status"] != "completed":
                    raise RuntimeError(f"Run {run_id}: {run['status']}: {run['error']}")
                print("completed_run_id=" + run_id)
                break
        else:
            raise RuntimeError("Stream closed before a run outcome; see Guides → Recovery")
```

HTTP errors raise instead of triggering automatic retries. The stream read timeout is a client limit, not a platform run deadline: a timeout does not stop the agent. Save the session and input IDs before recovering. The sample leaves its resources in place for inspection. Normal completion means execution ended normally; evaluate the answer's correctness separately. Use the same session for follow-ups; `queue` gives each input its own run, while `steer` can join active work. Set `MODEL` to choose another registered model of the same provider. See `GET /sessions/{session_id}/events/stream` for a standalone streaming consumer and replay rules.