---
title: "Fetch a real-time stream of multi-cloud exploration results"
method: GET
path: "/multi-cloud/jobs/{jobId}/stream"
tags: ["Multi-Cloud Strategy"]
---

# Fetch a real-time stream of multi-cloud exploration results

`GET /multi-cloud/jobs/{jobId}/stream`

## Request

- Base URL: `https://www.cloudworldmodel.ai/api`
- URL: `https://www.cloudworldmodel.ai/api/multi-cloud/jobs/{jobId}/stream`
- Auth: one of:
  - HTTP bearer
  - API key in header `X-PAYMENT`
  - API key in header `PAYMENT-SIGNATURE`

Stream multi-cloud exploration results using Server-Sent Events (SSE).
This endpoint provides real-time updates as strategies are generated.

**SSE Event Types:**
- `init`: Initial job state when connection is established
- `strategyGenerated`: A new strategy has been generated (sent for each strategy)
- `progressUpdate`: Progress update (sent periodically)
- `completed`: Job has finished successfully (final event before connection closes)
- `failed`: Job encountered an unrecoverable error (final event before connection closes)
- `cancelled`: Job was cancelled via `DELETE /api/multi-cloud/jobs/{jobId}` (final event before connection closes)

**Terminal Event Payload Contract:**

The terminal event's `data` payload carries: `status`, `progress`,
`strategiesGenerated`, `comparisonReport`, `latencyWarning`, `completedAt`,
`error` (failed jobs only), `methodology`, and `interpretationScope`.
`methodology` and `interpretationScope` follow the exact same schemas
documented on `GET /multi-cloud/jobs/{jobId}/results`: `methodology`
states the data origin (`dataSource: simulation-model`,
`externallyValidated: false`), and `interpretationScope` is the
machine-readable constraint object (`providerPerformanceScope:
simulation-only`, `prohibitedClaimCategories`, `testedLoadLevels`
derived from the job's configured traffic, `untestedScenariosNote`).
Streaming agents must consult `interpretationScope` from the terminal
event before forming any conclusion about provider performance — the
same rules as for the JSON results endpoints apply.

**Connection Behavior:**
- Connection remains open until job completes, fails, or is cancelled
- All existing strategies are streamed immediately upon connection
- New strategies are sent as they're generated
- Connection automatically closes when job finishes

**Agent Reconnection and Recovery Guide:**

After receiving a terminal SSE event (`completed`, `failed`, or `cancelled`) the server
closes the connection. Each event requires a different agent response:

- **`completed`**: The job finished successfully. No reconnection is needed. Fetch the
  full ranked results from `GET /api/multi-cloud/jobs/{jobId}/results` to retrieve
  `topStrategies`, `allStrategies`, and the `comparisonReport`. The results endpoint
  also accepts a `?providers=` filter if you only need strategies for specific clouds.

- **`failed`**: The job encountered an unrecoverable error. Inspect the `error` field in
  the event payload for the root cause. Transient errors (e.g. a pricing API timeout)
  are safe to retry — submit a new job via `POST /api/multi-cloud/jobs`. Permanent errors
  (e.g. an invalid scenario ID) should not be retried without fixing the underlying
  input first. Do not attempt to reconnect to the same `jobId`; it will not recover.

- **`cancelled`**: Cancellation is terminal and intentional. No retry is needed or
  recommended. If the cancellation was unintended, submit a new job.

**Handling unexpected connection drops (no terminal event received):**

If the SSE connection closes without a `completed`, `failed`, or `cancelled` event —
for example due to a network interruption, proxy timeout, or server restart — the job
may still be running. Use the following fallback strategy:

1. Poll `GET /api/multi-cloud/jobs/{jobId}` to check the current `status` field.
2. If `status` is `running` or `pending`, reconnect to this stream endpoint. The
   server replays all strategies generated so far on reconnect, so no data is lost.
3. If `status` is `completed`, `failed`, or `cancelled`, treat it the same as if you
   had received the corresponding terminal SSE event (see above).

Agents should implement an exponential back-off (e.g. 1 s, 2 s, 4 s, cap at 30 s)
before each reconnection attempt to avoid hammering the server during an outage.

**Use Cases:**
- Start analyzing strategies while generation continues
- Real-time progress monitoring
- Faster decision-making with early access to good strategies

**Client Code Sample (JavaScript / Node.js):**

`EventSource` does not support custom headers, so use `fetch` with a
`ReadableStream` to pass the Bearer token:

```javascript
async function streamMultiCloudJob(jobId, apiToken) {
  const response = await fetch(
    `https://your-host/api/multi-cloud/jobs/${jobId}/stream`,
    {
      headers: {
        Authorization: `Bearer ${apiToken}`,
        Accept: 'text/event-stream',
      },
    }
  );

  if (!response.ok) {
    throw new Error(`HTTP ${response.status}: ${await response.text()}`);
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    // SSE frames are separated by double newlines
    const frames = buffer.split(/\n\n/);
    buffer = frames.pop(); // keep incomplete trailing frame

    for (const frame of frames) {
      const eventLine = frame.match(/^event:\s*(.+)$/m);
      const dataLine  = frame.match(/^data:\s*(.+)$/m);
      if (!dataLine) continue;

      const eventType = eventLine ? eventLine[1].trim() : 'message';
      const payload   = JSON.parse(dataLine[1]);

      switch (eventType) {
        case 'init':
          console.log('Stream opened. Job status:', payload.status);
          break;

        case 'progressUpdate':
          console.log(`Progress: ${payload.progress}% — ${payload.message}`);
          break;

        case 'strategyGenerated':
          console.log('New strategy:', payload.strategy.name,
            '| cost $' + payload.strategy.metrics.monthlyCost);
          break;

        case 'completed':
          console.log('Job complete. Top strategy:',
            payload.topStrategy?.name);
          reader.cancel(); // close the connection
          return payload;
      }
    }
  }
}

// Usage
streamMultiCloudJob('job_aws_perf123', process.env.API_KEY)
  .then(result => console.log('Final result:', result))
  .catch(err  => console.error('Stream error:', err));
```

**Client Code Sample (Python):**

Use `httpx` with streaming to consume the SSE connection.
Install with `pip install httpx`.

```python
import httpx
import json
import os


def stream_multi_cloud_job(job_id: str, api_token: str) -> dict:
    """Stream a multi-cloud exploration job and return the final payload."""
    url = f"https://your-host/api/multi-cloud/jobs/{job_id}/stream"
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Accept": "text/event-stream",
    }

    with httpx.stream("GET", url, headers=headers, timeout=None) as response:
        response.raise_for_status()

        buffer = ""

        for chunk in response.iter_text():
            buffer += chunk

            # SSE frames are separated by double newlines
            *frames, buffer = buffer.split("\n\n")

            for frame in frames:
                event_type = "message"
                data_str = None

                for line in frame.splitlines():
                    if line.startswith("event:"):
                        event_type = line[len("event:"):].strip()
                    elif line.startswith("data:"):
                        data_str = line[len("data:"):].strip()

                if data_str is None:
                    continue

                payload = json.loads(data_str)

                if event_type == "init":
                    print(f"Stream opened. Job status: {payload['status']}")

                elif event_type == "progressUpdate":
                    print(f"Progress: {payload['progress']}% — {payload.get('message', '')}")

                elif event_type == "strategyGenerated":
                    strategy = payload["strategy"]
                    cost = strategy["metrics"]["monthlyCost"]
                    print(f"New strategy: {strategy['name']} | cost ${cost}")

                elif event_type == "completed":
                    top = payload.get("topStrategy", {})
                    print(f"Job complete. Top strategy: {top.get('name')}")
                    return payload  # connection closes when the context exits

    return {}


# Usage
if __name__ == "__main__":
    result = stream_multi_cloud_job(
        job_id="job_aws_perf123",
        api_token=os.environ["API_KEY"],
    )
    print("Final result:", result)
```

**Agent Reconnection Code Sample — with exponential back-off (JavaScript / Node.js):**

The samples above cover the happy path. The snippet below adds the full
reconnect loop: detecting a connection drop without a terminal event,
polling the status endpoint as a fallback, and reconnecting with
exponential back-off. All three terminal events (`completed`, `failed`,
`cancelled`) are handled explicitly.

```javascript
const BASE_URL = 'https://your-host/api';

// Fetch the current job state without opening an SSE stream.
async function pollJobStatus(jobId, apiToken) {
  const res = await fetch(`${BASE_URL}/multi-cloud/jobs/${jobId}`, {
    headers: { Authorization: `Bearer ${apiToken}` },
  });
  if (!res.ok) throw new Error(`Poll failed: HTTP ${res.status}`);
  return res.json(); // { status, progress, ... }
}

async function streamWithReconnect(jobId, apiToken) {
  let delay    = 1_000;  // back-off starts at 1 s
  const MAX_DELAY = 30_000;

  while (true) {
    let receivedTerminal = false;

    try {
      // ── Open the SSE connection ──────────────────────────────────
      const response = await fetch(
        `${BASE_URL}/multi-cloud/jobs/${jobId}/stream`,
        {
          headers: {
            Authorization: `Bearer ${apiToken}`,
            Accept: 'text/event-stream',
          },
        }
      );

      if (!response.ok) {
        throw new Error(`HTTP ${response.status}: ${await response.text()}`);
      }

      const reader  = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer    = '';

      outer: while (true) {
        const { value, done } = await reader.read();
        if (done) break; // server closed — check receivedTerminal below

        buffer += decoder.decode(value, { stream: true });
        const frames = buffer.split(/\n\n/);
        buffer = frames.pop(); // keep any incomplete trailing frame

        for (const frame of frames) {
          const eventLine = frame.match(/^event:\s*(.+)$/m);
          const dataLine  = frame.match(/^data:\s*(.+)$/m);
          if (!dataLine) continue;

          const eventType = eventLine ? eventLine[1].trim() : 'message';
          const payload   = JSON.parse(dataLine[1]);

          switch (eventType) {
            case 'init':
              // Successful (re)connect — reset back-off timer.
              console.log('Connected. Job status:', payload.status);
              delay = 1_000;
              break;

            case 'progressUpdate':
              console.log(`Progress: ${payload.progress}% — ${payload.message}`);
              break;

            case 'strategyGenerated':
              console.log('Strategy:', payload.strategy.name,
                '| $' + payload.strategy.metrics.monthlyCost + '/mo');
              break;

            // ── Terminal events ──────────────────────────────────────
            case 'completed':
              console.log('Job complete. Top strategy:', payload.topStrategy?.name);
              receivedTerminal = true;
              reader.cancel();
              return { status: 'completed', payload };

            case 'failed':
              console.error('Job failed:', payload.error);
              receivedTerminal = true;
              reader.cancel();
              return { status: 'failed', payload };

            case 'cancelled':
              console.warn('Job cancelled.');
              receivedTerminal = true;
              reader.cancel();
              return { status: 'cancelled', payload };
          }

          if (receivedTerminal) break outer;
        }
      }
    } catch (err) {
      // Network error, proxy timeout, or server restart.
      console.warn('SSE connection error:', err.message);
    }

    // Already handled cleanly — exit the retry loop.
    if (receivedTerminal) break;

    // ── Fallback: poll before reconnecting ───────────────────────────
    // The connection dropped without a terminal event. The job may
    // still be running, or it may have finished while we were offline.
    try {
      const job = await pollJobStatus(jobId, apiToken);

      if (job.status === 'completed') {
        console.log('Recovered via poll — job already completed.');
        return { status: 'completed', payload: job };
      }
      if (job.status === 'failed') {
        console.error('Recovered via poll — job failed:', job.error);
        return { status: 'failed', payload: job };
      }
      if (job.status === 'cancelled') {
        console.warn('Recovered via poll — job was cancelled.');
        return { status: 'cancelled', payload: job };
      }
      // status is 'running' or 'pending' — reconnect after back-off.
      console.log(`Job still ${job.status}. Reconnecting in ${delay / 1000}s…`);
    } catch (pollErr) {
      console.warn('Poll also failed:', pollErr.message, '— will retry.');
    }

    // ── Exponential back-off ─────────────────────────────────────────
    await new Promise(resolve => setTimeout(resolve, delay));
    delay = Math.min(delay * 2, MAX_DELAY);
  }
}

// Usage
streamWithReconnect('job_aws_perf123', process.env.API_KEY)
  .then(({ status, payload }) => console.log('Done:', status, payload))
  .catch(err => console.error('Unrecoverable error:', err));
```

**Agent Reconnection Code Sample — with exponential back-off (Python):**

```python
import httpx
import json
import os
import time


BASE_URL = "https://your-host/api"
TERMINAL_STATUSES = {"completed", "failed", "cancelled"}


def poll_job_status(job_id: str, api_token: str) -> dict:
    """Fetch the current job state without opening an SSE stream."""
    url = f"{BASE_URL}/multi-cloud/jobs/{job_id}"
    headers = {"Authorization": f"Bearer {api_token}"}
    response = httpx.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()  # {"status": ..., "progress": ..., ...}


def stream_with_reconnect(job_id: str, api_token: str) -> dict:
    """
    Open the SSE stream and reconnect automatically after unexpected drops.

    Polls the status endpoint when the connection closes without a terminal
    event, and applies exponential back-off before each reconnection attempt.

    Returns {"status": <terminal_status>, "payload": <event_payload>}.
    """
    url = f"{BASE_URL}/multi-cloud/jobs/{job_id}/stream"
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Accept": "text/event-stream",
    }
    delay     = 1.0   # back-off starts at 1 s
    max_delay = 30.0

    while True:
        received_terminal = False

        try:
            # ── Open the SSE connection ──────────────────────────────
            with httpx.stream("GET", url, headers=headers, timeout=None) as response:
                response.raise_for_status()
                delay  = 1.0  # reset back-off on successful connect
                buffer = ""

                for chunk in response.iter_text():
                    buffer += chunk
                    *frames, buffer = buffer.split("\n\n")

                    for frame in frames:
                        event_type = "message"
                        data_str   = None

                        for line in frame.splitlines():
                            if line.startswith("event:"):
                                event_type = line[len("event:"):].strip()
                            elif line.startswith("data:"):
                                data_str = line[len("data:"):].strip()

                        if data_str is None:
                            continue

                        payload = json.loads(data_str)

                        if event_type == "init":
                            # Successful (re)connect — reset back-off timer.
                            print(f"Connected. Job status: {payload['status']}")

                        elif event_type == "progressUpdate":
                            print(f"Progress: {payload['progress']}% — {payload.get('message', '')}")

                        elif event_type == "strategyGenerated":
                            s = payload["strategy"]
                            print(f"Strategy: {s['name']} | ${s['metrics']['monthlyCost']}/mo")

                        # ── Terminal events ──────────────────────────
                        elif event_type == "completed":
                            top = payload.get("topStrategy", {})
                            print(f"Job complete. Top strategy: {top.get('name')}")
                            received_terminal = True
                            return {"status": "completed", "payload": payload}

                        elif event_type == "failed":
                            print(f"Job failed: {payload.get('error')}")
                            received_terminal = True
                            return {"status": "failed", "payload": payload}

                        elif event_type == "cancelled":
                            print("Job cancelled.")
                            received_terminal = True
                            return {"status": "cancelled", "payload": payload}

                        if received_terminal:
                            break

        except (httpx.HTTPError, httpx.StreamError) as exc:
            # Network error, proxy timeout, or server restart.
            print(f"SSE connection error: {exc}")

        # Already handled cleanly — exit the retry loop.
        if received_terminal:
            break

        # ── Fallback: poll before reconnecting ───────────────────────
        # The connection dropped without a terminal event. The job may
        # still be running, or it may have finished while we were offline.
        try:
            job = poll_job_status(job_id, api_token)

            if job["status"] in TERMINAL_STATUSES:
                print(f"Recovered via poll — job {job['status']}.")
                return {"status": job["status"], "payload": job}

            # status is 'running' or 'pending' — reconnect after back-off.
            print(f"Job still {job['status']}. Reconnecting in {delay:.0f}s…")

        except httpx.HTTPError as exc:
            print(f"Poll also failed: {exc} — will retry.")

        # ── Exponential back-off ─────────────────────────────────────
        time.sleep(delay)
        delay = min(delay * 2, max_delay)

    return {}


# Usage
if __name__ == "__main__":
    result = stream_with_reconnect(
        job_id="job_aws_perf123",
        api_token=os.environ["API_KEY"],
    )
    print("Done:", result["status"], result.get("payload", {}))
```

## Path parameters

- `jobId` string, required

## Response `200`

SSE stream of job updates

## Other responses

- `401`: Unauthorized - invalid or missing API key
- `404`: Job not found
- `429`: Rate limit exceeded

---

[API](https://skmtc.dev/cloudworldmodel/apis/cloud-world-model-agent-api.md) · [All operations](https://skmtc.dev/cloudworldmodel/apis/cloud-world-model-agent-api/llms.txt) · [OpenAPI document](https://skmtc.dev/cloudworldmodel/apis/cloud-world-model-agent-api/revisions/ecc3ad1ff07f?raw)