Stream an inline workflow

/v1/workflows/execute-stream

Partly available: with a JSON body, pass input. To pass named inputs, send the workflow as text/yaml. Tracked as TI-53.

Headers: every response carries x-request-id (quote it in support requests). Send x-trace-id (optional, a UUID) to correlate this call across planes; reusing an id that already ran answers 409 duplicate_execution_id. Pick the body format with format or Accept (application/json, text/yaml, text/markdown, text/plain).

Authorization: send Authorization: Bearer $PLUNGE_API_KEY; X-API-Key: <ozk_ key> is accepted as an alternative.

Authorizations

Authorizationstringheaderrequired

Bearer token: Authorization: Bearer ozk_…. ozk_ platform API key (Authorization: Bearer ozk_…)

Body

application/json
text/yaml

workflowanyrequired

CNL workflow object, or a YAML string

inputstring

inputsobject

user_contextUserContext · object

The caller's time zone, locale and location. Optional on every execute body; when absent the gateway derives it from the request's network location (time zone, approximate place). The agent uses it for local time and as the default schedule time zone.

Show child attributes

time_zonestring

IANA zone

Example: "America/Los_Angeles"

localestring

BCP-47 tag

Example: "en-US"

placestring

Example: "San Francisco, California, US"

latitudenumber

Example: 37.7749

longitudenumber

Example: -122.4194

accuracy_mnumber

metres; only with a device GPS fix

Example: 35

sourceenum<string>required

Available options: device, network

Example: "device"

Response

SSE stream: request_received, workflow_loaded, workflow_started, task_dispatched, streaming_started (streaming agents), token (one frame per provider delta: {"task_id","delta"}), heartbeat, task_completed, then workflow_completed (the outcome, carries execution_summary; workflow_error on failure) and workflow_result, always the last frame. Single connection: the gateway drains the provider stream itself; terminal events always arrive on this SSE.

text/event-stream

missing_workflow / invalid_json / invalid_body (unreadable body) / invalid_workflow / invalid_yaml — refusals before the first frame are this JSON envelope, never an SSE frame

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

unauthorized — missing, invalid or revoked ozk_ key

Error code: unauthorizedfrom code

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

agent_not_active — a named agent card is not status:active (refused pre-dispatch)

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

unknown_agent — a named agent id does not exist in the registry (refused pre-dispatch); not_found — no route for this method on this path: every other method answers it

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

payload_too_large — body over 1 MiB (MAX_REQUEST_SIZE)

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

rate_limited — Retry-After header

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

engine_error — the engine refused the stream before its first frame

application/json

errorobjectrequired

Show child attributes

codestringrequired

messagestringrequired

connectobject

On a 424 connection_required / credential_required: where the missing connection is made.

Show child attributes

providerstring

methodenum<string>

Available options: oauth, api_key, wallet

linkstring

Example: "POST /v1/connections/stripe/link"

Stream events

Event	Description
`workflow_started`	Execution begins.
`task_dispatched`	A task is sent to its agent.
`token`	One provider delta, injected by the relay.
`task_completed`	A task finished.
`heartbeat`	While a task runs longer than 20 s, about every 25 s.
`workflow_completed`	Terminal event on success.
`workflow_error`	Terminal event on failure.

event: workflow_started
data: {"workflow_id":"00000000-0000-4000-8000-000000000001"}

event: task_dispatched
data: {"task_id":"t1","agent":"one-agent"}

event: token
data: {"task_id":"t1","delta":"pong"}

event: task_completed
data: {"task_id":"t1"}

event: heartbeat
data: {}

event: workflow_completed
data: {"final_task_id":"t1"}

event: workflow_error
data: {"error":"…"}

import os import requests workflow = { 'name': 'stream-demo', 'tasks': [{'type': 'task', 'id': 't1', 'agent': 'one-agent', 'prompt': 'Reply with only the word: pong'}], } with requests.post( 'https://api.plungeai.com/v1/workflows/execute-stream', headers={'Authorization': f"Bearer {os.environ['PLUNGE_API_KEY']}"}, json={'workflow': workflow, 'input': 'go'}, stream=True, ) as res: event = None for line in res.iter_lines(decode_unicode=True): if line.startswith('event: '): event = line[len('event: '):] elif line.startswith('data: '): print(event, line[len('data: '):]) if event in ('workflow_completed', 'workflow_error'): break

import os import requests workflow_yaml = """workflow: name: stream-demo tasks: - type: task id: t1 agent: one-agent prompt: "Reply with only the word: pong" input: go inputs: topic: pong """ with requests.post( 'https://api.plungeai.com/v1/workflows/execute-stream', headers={'Authorization': f"Bearer {os.environ['PLUNGE_API_KEY']}", 'Content-Type': 'text/yaml'}, data=workflow_yaml.encode(), stream=True, ) as res: for line in res.iter_lines(decode_unicode=True): if line: print(line)

const workflow = { name: 'stream-demo', tasks: [{ type: 'task', id: 't1', agent: 'one-agent', prompt: 'Reply with only the word: pong' }], } const res = await fetch('https://api.plungeai.com/v1/workflows/execute-stream', { method: 'POST', headers: { Authorization: `Bearer ${process.env.PLUNGE_API_KEY}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ workflow, input: 'go' }), }) const reader = res.body!.getReader() const decoder = new TextDecoder() let buffer = '' for (;;) { const { done, value } = await reader.read() if (done) break buffer += decoder.decode(value, { stream: true }) const frames = buffer.split('\n\n') buffer = frames.pop() ?? '' for (const frame of frames) { const event = /^event: (.*)$/m.exec(frame)?.[1] ?? 'message' const data = /^data: (.*)$/m.exec(frame)?.[1] ?? '' console.log(event, data) if (event === 'workflow_completed' || event === 'workflow_error') process.exit(0) } }

const workflowYaml = `workflow: name: stream-demo tasks: - type: task id: t1 agent: one-agent prompt: "Reply with only the word: pong" input: go inputs: topic: pong ` const res = await fetch('https://api.plungeai.com/v1/workflows/execute-stream', { method: 'POST', headers: { Authorization: `Bearer ${process.env.PLUNGE_API_KEY}`, 'Content-Type': 'text/yaml' }, body: workflowYaml, }) const reader = res.body!.getReader() const decoder = new TextDecoder() for (;;) { const { done, value } = await reader.read() if (done) break process.stdout.write(decoder.decode(value, { stream: true })) }

import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; public class Main { public static void main(String[] args) throws Exception { String json = "{\"workflow\":{\"name\":\"stream-demo\",\"tasks\":[{\"type\":\"task\",\"id\":\"t1\",\"agent\":\"one-agent\",\"prompt\":\"Reply with only the word: pong\"}]},\"input\":\"go\"}"; HttpRequest req = HttpRequest.newBuilder(URI.create("https://api.plungeai.com/v1/workflows/execute-stream")) .header("Authorization", "Bearer " + System.getenv("PLUNGE_API_KEY")) .header("Content-Type", "application/json") .POST(HttpRequest.BodyPublishers.ofString(json)) .build(); HttpClient.newHttpClient().send(req, HttpResponse.BodyHandlers.ofLines()).body() .filter(line -> line.startsWith("event: ") || line.startsWith("data: ")) .takeWhile(line -> !line.equals("event: workflow_completed") && !line.equals("event: workflow_error")) .forEach(System.out::println); } }

import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; public class Main { public static void main(String[] args) throws Exception { String yaml = String.join("\n", "workflow:", " name: stream-demo", " tasks:", " - type: task", " id: t1", " agent: one-agent", " prompt: \"Reply with only the word: pong\"", "input: go", "inputs:", " topic: pong", ""); HttpRequest req = HttpRequest.newBuilder(URI.create("https://api.plungeai.com/v1/workflows/execute-stream")) .header("Authorization", "Bearer " + System.getenv("PLUNGE_API_KEY")) .header("Content-Type", "text/yaml") .POST(HttpRequest.BodyPublishers.ofString(yaml)) .build(); HttpClient.newHttpClient().send(req, HttpResponse.BodyHandlers.ofLines()).body() .filter(line -> !line.isEmpty()) .forEach(System.out::println); } }

curl -sN https://api.plungeai.com/v1/workflows/execute-stream \ -H "Authorization: Bearer $PLUNGE_API_KEY" \ -H 'Content-Type: application/json' \ -d '{"workflow":{"name":"stream-demo","tasks":[{"type":"task","id":"t1","agent":"one-agent","prompt":"Reply with only the word: pong"}]},"input":"go"}'

cat > workflow.yaml <<'YAML' workflow: name: stream-demo tasks: - type: task id: t1 agent: one-agent prompt: "Reply with only the word: pong" input: go inputs: topic: pong YAML curl -sN https://api.plungeai.com/v1/workflows/execute-stream \ -H "Authorization: Bearer $PLUNGE_API_KEY" \ -H 'Content-Type: text/yaml' \ --data-binary @workflow.yaml

event: workflow_started data: {"workflow_id":"00000000-0000-4000-8000-000000000001"} event: task_dispatched data: {"task_id":"t1","agent":"one-agent"} event: token data: {"task_id":"t1","delta":"pong"} event: task_completed data: {"task_id":"t1"} event: heartbeat data: {} event: workflow_completed data: {"final_task_id":"t1"} event: workflow_error data: {"error":"…"}