Here’s how to use the /invoke/stdout endpoint with Python to stream responses from the Cloud Coding API. We’ll use the requests library to make the HTTP POST request and handle the Server-Sent Events (SSE) stream.

​
Installation

First, make sure you have the requests library installed in your environment:

pip install requests

​
Single Task Invocation

This example demonstrates how to invoke a single task and process the streaming response. The script prints each event received from the API.

import requests
import json

# --- Configuration ---
API_KEY = "YOUR_CLOUDCODING_API_KEY"
GITHUB_TOKEN = "YOUR_GITHUB_TOKEN"
REPO_URL = "https://github.com/username/repo-name"
API_URL = "https://api.cloudcoding.ai/invoke/stdout"

# --- Request Headers ---
headers = {
    "Content-Type": "application/json",
    "x-api-key": API_KEY,
}

# --- Request Body ---
data = {
    "repo_url": REPO_URL,
    "github_token": GITHUB_TOKEN,
    "messages": [
        {
            "type": "human",
            "content": "analyze the repo to see what it is about"
        }
    ]
}

# --- Stream the Response ---
try:
    with requests.post(
        API_URL,
        headers=headers,
        json=data,
        stream=True,
    ) as response:
        response.raise_for_status()  # Raise an exception for bad status codes
        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data:'):
                    try:
                        # Extract the JSON payload from the SSE message
                        event_data = json.loads(decoded_line[len('data:'):])
                        print(json.dumps(event_data, indent=2))
                    except json.JSONDecodeError:
                        print(f"Could not decode JSON from line: {decoded_line}")
except requests.exceptions.RequestException as e:
    print(f"An error occurred: {e}")

​
Parallel Task Invocation

This example runs two tasks in parallel. The output from custom events includes a task_name field, allowing you to distinguish between outputs from different concurrent tasks.

import requests
import json

# --- Configuration ---
API_KEY = "YOUR_CLOUDCODING_API_KEY"
GITHUB_TOKEN = "YOUR_GITHUB_TOKEN"
REPO_URL = "https://github.com/username/my-app"
API_URL = "https://api.cloudcoding.ai/invoke/stdout"

# --- Request Headers ---
headers = {
    "Content-Type": "application/json",
    "x-api-key": API_KEY,
}

# --- Request Body ---
data = {
    "repo_url": REPO_URL,
    "github_token": GITHUB_TOKEN,
    "parallel_tasks": [
      {
        "prompt": "Add comprehensive error handling",
        "task_name": "error-handling"
      },
      {
        "prompt": "Create unit tests for all components",
        "task_name": "testing"
      }
    ]
}

# --- Stream the Response ---
try:
    with requests.post(
        API_URL,
        headers=headers,
        json=data,
        stream=True,
    ) as response:
        response.raise_for_status()
        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data:'):
                    try:
                        event_data = json.loads(decoded_line[len('data:'):])
                        if event_data.get("event") == "custom":
                            task_name = event_data.get("data", {}).get("task_name")
                            stdout = event_data.get("data", {}).get("stdout")
                            print(f"--- Output from {task_name} ---")
                            print(stdout)
                        else:
                            print(json.dumps(event_data, indent=2))
                    except json.JSONDecodeError:
                        print(f"Could not decode JSON from line: {decoded_line}")
except requests.exceptions.RequestException as e:
    print(f"An error occurred: {e}")