Streaming & human-in-the-loop in the SDKs

curl -N https://api.modulex.dev/workflows/listen/run_abc \
  -H "Authorization: Bearer mx_live_xxx" \
  -H "X-Organization-ID: 6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c" \
  -H "Accept: text/event-stream"

import asyncio
from modulex import Modulex

async def main():
    async with Modulex(api_key="mx_live_xxx", organization_id="6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c") as client:
        async with client.executions.listen("run_abc") as stream:
            async for event in stream:
                print(event.event, event.data)
                if event.is_terminal:
                    break

asyncio.run(main())

import { Modulex } from "modulex-js";

const client = new Modulex({
  apiKey: "mx_live_xxx",
  organizationId: "6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c",
});

for await (const event of client.executions.listen("run_abc")) {
  if (event.type === "response_chunk") process.stdout.write(event.content ?? "");
  if (event.type === "done" || event.type === "error") break;
}

# Stop the server-side run (separate from closing the stream connection)
curl -X POST https://api.modulex.dev/workflows/cancel/run_abc \
  -H "Authorization: Bearer mx_live_xxx" \
  -H "X-Organization-ID: 6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c" \
  -H "Content-Type: application/json" \
  -d '{"reason": "user_requested"}'

import asyncio
from modulex import Modulex

async def main():
    async with Modulex(api_key="mx_live_xxx", organization_id="org_uuid") as client:
        stream = client.executions.listen("run_abc")
        async with stream:
            async for event in stream:
                if event.event == "node_update":
                    await stream.close()          # stop reading the stream
                    break
        # Terminate the run itself, server-side:
        await client.executions.cancel("run_abc", reason="user_requested")

asyncio.run(main())

const controller = new AbortController();
const stream = client.executions.listen("run_abc", { signal: controller.signal });

// Stop listening after 5 seconds
setTimeout(() => controller.abort(), 5_000);

try {
  for await (const evt of stream) {
    if (evt.type === "done" || evt.type === "error") break;
  }
} catch (err) {
  // Aborting surfaces as the fetch's abort error — handle or ignore as needed
}

// Terminate the run itself, server-side:
await client.executions.cancel("run_abc", { reason: "user_requested" });

# 1. Answer the paused question (returns a NEW run_id + stream_url)
curl -X POST https://api.modulex.dev/composer/chat/cmp_123/resume \
  -H "Authorization: Bearer mx_live_xxx" \
  -H "X-Organization-ID: org_uuid" \
  -H "Content-Type: application/json" \
  -d '{
    "request_id": "req_42",
    "response": { "kind": "yes_no", "answer": true },
    "llm": { "integration_name": "openai", "provider_id": "openai", "model_id": "gpt-4o" }
  }'

# 2. Re-listen on the NEW run_id from the response
curl -N https://api.modulex.dev/composer/chat/cmp_123/listen/<new_run_id> \
  -H "Authorization: Bearer mx_live_xxx" \
  -H "X-Organization-ID: org_uuid" \
  -H "Accept: text/event-stream"

from modulex.types.realtime import user_input_request_from_event, YesNoResponse, ComposerLLMConfig

llm = ComposerLLMConfig(integration_name="openai", provider_id="openai", model_id="gpt-4o")
resp = await client.composer.chat("Build me a Slack alert workflow", llm=llm)
run_id = resp.run_id

while True:
    paused = None
    async with client.composer.listen(resp.composer_chat_id, run_id) as stream:
        async for event in stream:
            if event.event == "user_input_request":
                paused = user_input_request_from_event(event.data)
                break                       # stop reading; go answer it
            if event.is_terminal:
                break                       # run finished
    if paused is None:
        break

    new = await client.composer.resume(
        resp.composer_chat_id,
        request_id=paused.request_id,
        response=YesNoResponse(answer=True),
        llm=llm,
    )
    run_id = new.run_id                      # NEW run_id — loop re-listens on it

const llm = { integration_name: "openai", provider_id: "openai", model_id: "gpt-4o" };
const start = await client.composer.chat({ message: "Build me a Slack alert workflow", llm });
let runId = start.run_id;

for (;;) {
  let paused = null;
  for await (const evt of client.composer.listen(start.composer_chat_id, runId)) {
    if (evt.type === "user_input_request") {
      paused = evt.data;                     // UserInputRequest; discriminate on paused.kind
      break;
    }
    if (evt.type === "done" || evt.type === "error") break;
  }
  if (!paused) break;

  const resumed = await client.composer.resume(start.composer_chat_id, {
    requestId: paused.request_id,
    response: { kind: "yes_no", answer: true },
    llm,
  });
  runId = resumed.run_id;                    // NEW run_id — loop re-listens on it
}