Streaming & Async Jobs

curl -N "https://<your-app-hostname>/" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $DYNAMIQ_ACCESS_KEY" \
  -d '{
    "input": { "question": "What can this app do?" },
    "stream": true
  }'

import os
import requests
import json

endpoint = "https://<your-app-hostname>"
token = os.getenv("DYNAMIQ_ACCESS_KEY")

streaming_event = "data"  # Event name configured in the UI

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {token}",
}

# Payload: keys of "input" must match the input node schema defined in the UI
payload = {
    "input": {"question": "What can this app do?"},
    "stream": True,
}

response = requests.post(endpoint, json=payload, headers=headers, stream=True)
response.raise_for_status()

for line in response.iter_lines(decode_unicode=True):
    if line.startswith("data:"):
        data = line[len("data:"):].strip()
        try:
            json_data = json.loads(data)
            if json_data.get("event") == streaming_event:
                content = json_data.get("data", {}).get("choices", [{}])[0].get("delta", {}).get("content")
                if content:
                    print(content, end="")
        except json.JSONDecodeError as e:
            print(f"Invalid JSON format: {data} - Error: {e}")

const endpoint = "https://<your-app-hostname>";
const token = process.env.DYNAMIQ_ACCESS_KEY;

const streamingEvent = "data"; // Event name configured in the UI

const response = await fetch(endpoint, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${token}`,
  },
  body: JSON.stringify({
    input: { question: "What can this app do?" },
    stream: true,
  }),
});

if (!response.ok || !response.body) {
  throw new Error(`Failed to connect: ${response.status}`);
}

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

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

  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n");
  buffer = lines.pop() || "";

  for (const line of lines) {
    if (!line.startsWith("data:")) continue;
    const jsonData = JSON.parse(line.substring(5).trim());
    if (jsonData.event === streamingEvent) {
      const content = jsonData.data?.choices?.[0]?.delta?.content;
      if (content) process.stdout.write(content);
    }
  }
}

import os
import asyncio
import json
import logging
import websockets
from dynamiq.types.feedback import APPROVAL_EVENT
from dynamiq.types.streaming import StreamingEventMessage

logger = logging.getLogger(__name__)

WS_URI = "wss://<your-app-hostname>"
token = os.getenv("DYNAMIQ_ACCESS_KEY")

streaming_event = "data"  # Event name configured in the UI

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {token}",
}

payload = {
    "input": {"question": "What can this app do?"},
    "stream": True,
}


async def websocket_client():
    async with websockets.connect(WS_URI, extra_headers=headers) as websocket:
        wf_run_event = StreamingEventMessage(entity_id=None, data=payload)
        await websocket.send(wf_run_event.to_json())

        try:
            while True:
                data = await websocket.recv()
                json_data = json.loads(data)
                event = StreamingEventMessage(**json_data)

                if event.event == "approval":
                    feedback = input(event.data["template"])
                    feedback_message = StreamingEventMessage(
                        entity_id=event.entity_id,
                        wf_run_id=event.wf_run_id,
                        data={"feedback": feedback},
                        event=APPROVAL_EVENT,
                    )
                    await websocket.send(feedback_message.to_json())

                if json_data.get("event") == streaming_event:
                    content = json_data.get("data", {}).get("choices", [{}])[0].get("delta", {}).get("content")
                    if content:
                        print(content, end="")

                # The final message comes from the Workflow itself.
                if event.source.name == "Workflow":
                    break

        except websockets.ConnectionClosed:
            logger.error("WebSocket connection closed by the server")


if __name__ == "__main__":
    asyncio.run(websocket_client())

const WS_URI = "wss://<your-app-hostname>";

const streamingEvent = "data"; // Event name configured in the UI
const APPROVAL_EVENT = "approval";

const payload = {
  input: { question: "What can this app do?" },
  stream: true,
};

const websocket = new WebSocket(WS_URI);

websocket.onopen = () => {
  websocket.send(JSON.stringify({ entity_id: null, data: payload }));
};

websocket.onmessage = (event) => {
  const jsonData = JSON.parse(event.data);

  if (jsonData.event === "approval") {
    const userFeedback = prompt(jsonData.data.template);
    websocket.send(
      JSON.stringify({
        entity_id: jsonData.entity_id,
        wf_run_id: jsonData.wf_run_id,
        data: { feedback: userFeedback },
        event: APPROVAL_EVENT,
      }),
    );
  }

  if (jsonData.event === streamingEvent) {
    const content = jsonData.data?.choices?.[0]?.delta?.content;
    if (content) console.log(content);
  }

  // The final message comes from the Workflow itself.
  if (jsonData.source?.name === "Workflow") {
    websocket.close();
  }
};

curl "https://<your-app-hostname>/" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $DYNAMIQ_ACCESS_KEY" \
  -d '{
    "input": { "question": "Generate the weekly report" },
    "execution_mode": "async",
    "callbacks": [
      {
        "url": "https://your-callback-url.example.com/webhook",
        "auth": { "type": "bearer", "token": "your-callback-auth-token" },
        "metadata": { "key": "value" }
      }
    ]
  }'

import os
import requests
import json

endpoint = "https://<your-app-hostname>"
token = os.getenv("DYNAMIQ_ACCESS_KEY")

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {token}",
}

payload = {
    "input": {"question": "Generate the weekly report"},
    "execution_mode": "async",
    "callbacks": [
        {
            "url": "https://your-callback-url.example.com/webhook",
            "auth": {"type": "bearer", "token": "your-callback-auth-token"},
            "metadata": {"key": "value"},
        }
    ],
}

response = requests.post(endpoint, json=payload, headers=headers)
response.raise_for_status()
print("Accepted:", json.dumps(response.json(), indent=4))

const endpoint = "https://<your-app-hostname>";
const token = process.env.DYNAMIQ_ACCESS_KEY;

const response = await fetch(endpoint, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${token}`,
  },
  body: JSON.stringify({
    input: { question: "Generate the weekly report" },
    execution_mode: "async",
    callbacks: [
      {
        url: "https://your-callback-url.example.com/webhook",
        auth: { type: "bearer", token: "your-callback-auth-token" },
        metadata: { key: "value" },
      },
    ],
  }),
});

console.log("Accepted:", await response.json());

import asyncio
import json
import os

import httpx

endpoint = "https://<your-app-hostname>"
token = os.getenv("DYNAMIQ_ACCESS_KEY")
headers = {"Authorization": f"Bearer {token}"}

payload = {
    "input": {"question": "Draft and send the renewal email"},
    "stream": True,
}

run_id: str | None = None
hitl_request_id: str | None = None
hitl_ready = asyncio.Event()


async def stream_events(client: httpx.AsyncClient) -> None:
    global run_id, hitl_request_id
    async with client.stream(
        "POST",
        f"{endpoint}/v1/runs",
        json=payload,
        headers=headers,
        timeout=None,
    ) as response:
        async for raw in response.aiter_lines():
            if not raw.startswith("data: "):
                continue
            event = json.loads(raw.removeprefix("data: "))
            print(event)
            if event["type"] == "run.created":
                run_id = event["data"]["id"]
            elif event["type"] == "agent.human_feedback.requested":
                hitl_request_id = event["data"]["id"]
                hitl_ready.set()


async def main() -> None:
    async with httpx.AsyncClient() as client:
        stream_task = asyncio.create_task(stream_events(client))

        # Wait for the pause event, then post the reply.
        await hitl_ready.wait()
        await client.post(
            f"{endpoint}/v1/runs/{run_id}/input",
            json={
                "type": "human_feedback",
                "data": {"request_id": hitl_request_id, "feedback": "approve"},
            },
            headers=headers,
        )

        await stream_task


asyncio.run(main())

const endpoint = "https://<your-app-hostname>";
const token = process.env.DYNAMIQ_ACCESS_KEY;
const headers = {
  "Content-Type": "application/json",
  Authorization: `Bearer ${token}`,
};

const payload = {
  input: { question: "Draft and send the renewal email" },
  stream: true,
};

let runId: string | null = null;
let hitlRequestId: string | null = null;
let signalHitlReady!: () => void;
const hitlReady = new Promise<void>((resolve) => (signalHitlReady = resolve));

async function streamEvents() {
  const response = await fetch(`${endpoint}/v1/runs`, {
    method: "POST",
    headers,
    body: JSON.stringify(payload),
  });

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

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    let idx: number;
    while ((idx = buffer.indexOf("\n")) !== -1) {
      const raw = buffer.slice(0, idx).trim();
      buffer = buffer.slice(idx + 1);
      if (!raw.startsWith("data: ")) continue;
      const event = JSON.parse(raw.slice("data: ".length));
      console.log(event);

      if (event.type === "run.created") runId = event.data.id;
      else if (event.type === "agent.human_feedback.requested") {
        hitlRequestId = event.data.id;
        signalHitlReady();
      }
    }
  }
}

async function main() {
  const streamTask = streamEvents();

  // Wait for the pause event, then post the reply.
  await hitlReady;
  await fetch(`${endpoint}/v1/runs/${runId}/input`, {
    method: "POST",
    headers,
    body: JSON.stringify({
      type: "human_feedback",
      data: { request_id: hitlRequestId, feedback: "approve" },
    }),
  });

  await streamTask;
}

main();