This page is not available in 繁體中文 yet. You are reading the English version.
Reading the stream
This page gives you a complete client for POST /chat/control/stream in Node (built-in fetch, no
dependencies) and in Python (with httpx). Each one parses frames correctly when the network splits them,
survives heartbeats and unknown events, applies timeouts, can be aborted, and retries only what is safe to
retry. The last section maps each event to what a map interface should do with it.
You need a server key, which is available on Pro and Enterprise. If you have not made a first call yet, start with the Server quickstart. The contract these clients implement is the SSE wire contract.
What a correct client does
| Rule | Why |
|---|---|
| Check the HTTP status before parsing. | A refused request is a JSON error, not a stream. |
| Buffer bytes and split on line breaks; act on a frame at its blank line. | A network read can end in the middle of a line, a frame or a multi-byte character. |
Treat a data: line with no event: line as prose. | The answer's text arrives in unnamed frames. |
| Skip lines that start with a colon, but count them as activity. | : keep-alive is a heartbeat sent after 10 seconds of silence. |
| Ignore event names and fields you do not know. | New ones are added without a version change. |
Keep reading after end, until the body closes. | The quota event comes after end. |
Treat a stream that closes without end or error as failed. | A dropped connection sends no terminal event. |
| Use an idle timeout, not a short total timeout. | A good answer can take a minute; a dead connection is silent. |
| Retry before the stream starts, not after. | A retry is a new request and is charged again. |
EventSource, the browser's built-in SSE client, cannot be used: it cannot send a POST body or an
Authorization header. Read the response body yourself, as these clients do.
Node client
Save this as kaleidr-stream.mjs. It needs Node 18 or later and nothing else.
// kaleidr-stream.mjs: Node 18 or later, no dependencies.
const API = 'https://api.kaleidr.com/inference-api/b2b/v1';
const IDLE_MS = 45_000; // a keep-alive arrives every 10 s, so 45 s of silence means the connection is dead
const TOTAL_MS = 11 * 60_000; // the server ends a stream itself after about 10 minutes
const MAX_ATTEMPTS = 4;
const RETRYABLE_429 = new Set(['rate_limited', 'org_overloaded', 'b2b_overloaded', 'limiter_unavailable']);
export class KaleidrError extends Error {
constructor(message, { status = null, code = null, body = null, retryAfter = null } = {}) {
super(message);
this.name = 'KaleidrError';
Object.assign(this, { status, code, body, retryAfter });
}
}
function sleep(ms, signal) {
return new Promise((resolve, reject) => {
const timer = setTimeout(resolve, ms);
signal.addEventListener(
'abort',
() => {
clearTimeout(timer);
reject(signal.reason);
},
{ once: true },
);
});
}
/** Yield one { event, data } per frame, wherever the network splits the bytes. */
export async function* sseFrames(body, onBytes = () => {}) {
const decoder = new TextDecoder();
let buffer = '';
let event = null;
let data = [];
for await (const chunk of body) {
onBytes(); // comments and half frames count as activity
buffer += decoder.decode(chunk, { stream: true }); // stream: true keeps a split character intact
let newline;
while ((newline = buffer.indexOf('\n')) !== -1) {
let line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
if (line.endsWith('\r')) line = line.slice(0, -1);
if (line === '') {
// A blank line completes the frame.
if (data.length > 0) yield { event, data: data.join('\n') };
event = null;
data = [];
} else if (line.startsWith(':')) {
// A comment such as ": keep-alive". Nothing to do.
} else if (line.startsWith('event:')) {
event = line.slice(6).trim();
} else if (line.startsWith('data:')) {
data.push(line.startsWith('data: ') ? line.slice(6) : line.slice(5));
}
}
}
}
/** Send the request. Retry only refusals that are worth retrying, and only before the stream starts. */
async function openStream(key, payload, signal, idle) {
for (let attempt = 1; ; attempt++) {
let res;
idle.arm();
try {
res = await fetch(`${API}/chat/control/stream`, {
method: 'POST',
headers: {
Authorization: `Bearer ${key}`,
'Content-Type': 'application/json',
Accept: 'text/event-stream',
},
body: JSON.stringify(payload),
signal,
});
} catch (err) {
idle.disarm();
if (signal.aborted || attempt >= MAX_ATTEMPTS) throw err;
await sleep(2 ** attempt * 500 + Math.random() * 500, signal);
continue;
}
if (res.ok) return res;
idle.disarm();
const body = await res.json().catch(() => ({}));
const code =
typeof body.error === 'string'
? body.error
: typeof body.detail === 'string'
? body.detail
: 'validation_error';
const retryAfter = Number(res.headers.get('retry-after')) || null;
const retryable = (res.status === 429 && RETRYABLE_429.has(code)) || res.status >= 500;
if (!retryable || attempt >= MAX_ATTEMPTS) {
throw new KaleidrError(`Kaleidr refused the request: ${res.status} ${code}`, {
status: res.status,
code,
body,
retryAfter,
});
}
await sleep(((retryAfter ?? 2 ** attempt) + Math.random()) * 1000, signal);
}
}
/** Later events add detail to a place. The first coordinates received for a name never change. */
function mergePlace(known, incoming) {
const merged = { ...known };
for (const [field, value] of Object.entries(incoming)) {
if (field === 'coordinates' || field === 'is_update') continue;
if (value !== null && value !== '') merged[field] = value;
}
if (!known.coordinates && incoming.coordinates) merged.coordinates = incoming.coordinates;
return merged;
}
/**
* Ask one question and read the whole stream.
* `on` holds optional callbacks: text, place, actions, route, routeClear, promptOptions, quota.
*/
export async function askKaleidr({ key, messages, sessionId, location, mapZoom, lang = 'en', signal, on = {} }) {
const controller = new AbortController();
const forwardAbort = () => controller.abort(signal.reason);
if (signal?.aborted) forwardAbort();
signal?.addEventListener('abort', forwardAbort, { once: true });
let idleTimer;
const idle = {
arm() {
clearTimeout(idleTimer);
idleTimer = setTimeout(
() => controller.abort(new KaleidrError('No data received for 45 seconds', { code: 'idle_timeout' })),
IDLE_MS,
);
},
disarm() {
clearTimeout(idleTimer);
},
};
const totalTimer = setTimeout(
() => controller.abort(new KaleidrError('The stream ran past its time limit', { code: 'total_timeout' })),
TOTAL_MS,
);
const places = new Map(); // name -> place
const routes = new Map(); // id -> route
const result = {
complete: false, // true once `end` arrives
sessionId: sessionId ?? null,
text: '',
places: [],
actions: [],
routes: [],
promptOptions: null,
quota: null,
end: null,
error: null,
};
try {
const payload = { messages, session_id: sessionId, location, map_zoom: mapZoom, lang };
const res = await openStream(key, payload, controller.signal, idle);
for await (const { event, data: raw } of sseFrames(res.body, idle.arm)) {
let data;
try {
data = JSON.parse(raw);
} catch {
continue; // not JSON: skip the frame
}
if (event === null) {
if (typeof data.text === 'string') {
result.text += data.text;
on.text?.(data.text);
}
} else if (event === 'place') {
if (typeof data.name !== 'string') continue;
const place = places.has(data.name) ? mergePlace(places.get(data.name), data) : data;
places.set(data.name, place);
on.place?.(place);
} else if (event === 'place_linked') {
if (!places.has(data.name)) continue; // detail for a place we never saw
const place = mergePlace(places.get(data.name), data);
places.set(data.name, place);
on.place?.(place);
} else if (event === 'early_actions') {
on.actions?.(data.actions ?? [], { final: false });
} else if (event === 'route_clear') {
routes.clear();
on.routeClear?.();
} else if (event === 'route') {
routes.set(data.id ?? `route-${routes.size}`, data);
on.route?.(data);
} else if (event === 'prompt_options') {
result.promptOptions = data;
on.promptOptions?.(data);
} else if (event === 'phase' || event === 'start') {
if (data.session_id) result.sessionId = data.session_id;
} else if (event === 'end') {
result.complete = true;
result.end = data;
if (typeof data.full_text === 'string') result.text = data.full_text;
for (const place of data.places ?? []) {
if (typeof place?.name !== 'string') continue;
places.set(place.name, places.has(place.name) ? mergePlace(places.get(place.name), place) : place);
}
for (const name of data.dropped_places ?? []) places.delete(name);
result.actions = data.actions ?? [];
on.actions?.(result.actions, { final: true });
} else if (event === 'error') {
result.error = data;
} else if (event === 'quota') {
result.quota = data;
on.quota?.(data);
}
// Any other event name is ignored on purpose.
}
} finally {
idle.disarm();
clearTimeout(totalTimer);
signal?.removeEventListener('abort', forwardAbort);
}
result.places = [...places.values()];
result.routes = [...routes.values()];
return result;
}
Use it like this:
// ask.mjs: run with node ask.mjs "Three bakeries near the Louvre"
import { askKaleidr, KaleidrError } from './kaleidr-stream.mjs';
const controller = new AbortController();
process.on('SIGINT', () => controller.abort(new Error('Cancelled'))); // Ctrl+C aborts the stream
try {
const answer = await askKaleidr({
key: process.env.KALEIDR_KEY,
sessionId: 'trip-42',
messages: [{ role: 'user', content: process.argv[2] ?? 'Three bakeries near the Louvre' }],
location: { lat: 48.8606, lng: 2.3376 },
mapZoom: 14,
signal: controller.signal,
on: {
text: (piece) => process.stdout.write(piece),
place: (place) => {
if (place.coordinates) console.error(`\n[pin] ${place.name}`);
},
},
});
console.log();
if (answer.error) {
console.error(`The answer failed: ${answer.error.message}`);
} else if (!answer.complete) {
console.error('The stream closed before the answer finished.');
}
for (const place of answer.places) {
const at = place.coordinates ? `${place.coordinates.lat}, ${place.coordinates.lng}` : 'no location';
console.log(`- ${place.name} (${at})`);
}
if (answer.quota) console.log(`AI credits used this period: ${answer.quota.used} of ${answer.quota.limit}`);
} catch (err) {
if (err instanceof KaleidrError && err.code === 'quota_exceeded') {
console.error(`Out of AI credits until ${err.body.reset_at}`);
} else if (err instanceof KaleidrError) {
console.error(`${err.message}`);
} else {
throw err;
}
process.exitCode = 1;
}
export KALEIDR_KEY="kld_sk_live_…"
node ask.mjs "Three bakeries near the Louvre"
You should see the answer print as it is written, a [pin] line on standard error for each located place,
then the final list of places and your AI-credit usage.
Python client
This client uses httpx, because its read timeout works as an idle timeout on
a streaming response. Install it with pip install httpx. It needs Python 3.9 or later.
It splits lines itself from raw bytes instead of calling iter_lines(). Line-splitting helpers treat some
Unicode characters as line breaks, and such a character inside a place name or a sentence would cut a JSON
payload in two.
Save this as kaleidr_stream.py:
# kaleidr_stream.py: Python 3.9 or later. pip install httpx
from __future__ import annotations
import json
import os
import random
import time
from typing import Any, Callable, Iterable, Iterator, Optional
import httpx
API = "https://api.kaleidr.com/inference-api/b2b/v1"
IDLE_SECONDS = 45.0 # a keep-alive arrives every 10 s, so 45 s of silence means the connection is dead
TOTAL_SECONDS = 11 * 60 # the server ends a stream itself after about 10 minutes
MAX_ATTEMPTS = 4
RETRYABLE_429 = {"rate_limited", "org_overloaded", "b2b_overloaded", "limiter_unavailable"}
class KaleidrError(Exception):
def __init__(self, message: str, *, status: Optional[int] = None, code: Optional[str] = None,
body: Any = None, retry_after: Optional[float] = None) -> None:
super().__init__(message)
self.status, self.code, self.body, self.retry_after = status, code, body, retry_after
def sse_frames(chunks: Iterable[bytes], on_bytes: Callable[[], None] = lambda: None) -> Iterator[tuple]:
"""Yield one (event, data) pair per frame, wherever the network splits the bytes."""
buffer = b""
event: Optional[str] = None
data: list = []
for chunk in chunks:
on_bytes() # comments and half frames count as activity
buffer += chunk
while True:
newline = buffer.find(b"\n")
if newline == -1:
break
# Decode whole lines only, so a character split across two chunks stays intact.
line = buffer[:newline].rstrip(b"\r").decode("utf-8")
buffer = buffer[newline + 1:]
if line == "":
# A blank line completes the frame.
if data:
yield event, "\n".join(data)
event, data = None, []
elif line.startswith(":"):
pass # a comment such as ": keep-alive"
elif line.startswith("event:"):
event = line[6:].strip()
elif line.startswith("data:"):
data.append(line[6:] if line.startswith("data: ") else line[5:])
def _refusal(response: httpx.Response) -> KaleidrError:
try:
body = response.json()
except ValueError:
body = {}
if isinstance(body.get("error"), str):
code = body["error"]
elif isinstance(body.get("detail"), str):
code = body["detail"]
else:
code = "validation_error"
retry_after = response.headers.get("retry-after")
return KaleidrError(
f"Kaleidr refused the request: {response.status_code} {code}",
status=response.status_code,
code=code,
body=body,
retry_after=float(retry_after) if retry_after and retry_after.isdigit() else None,
)
def _merge_place(known: dict, incoming: dict) -> dict:
"""Later events add detail to a place. The first coordinates received for a name never change."""
merged = dict(known)
for field, value in incoming.items():
if field in ("coordinates", "is_update"):
continue
if value is not None and value != "":
merged[field] = value
if not known.get("coordinates") and incoming.get("coordinates"):
merged["coordinates"] = incoming["coordinates"]
return merged
def ask_kaleidr(
key: str,
messages: list,
*,
session_id: Optional[str] = None,
location: Optional[dict] = None,
map_zoom: Optional[float] = None,
lang: str = "en",
on_text: Optional[Callable[[str], None]] = None,
on_place: Optional[Callable[[dict], None]] = None,
should_stop: Optional[Callable[[], bool]] = None,
) -> dict:
"""Ask one question and read the whole stream."""
payload: dict = {"messages": messages, "lang": lang}
if session_id:
payload["session_id"] = session_id
if location:
payload["location"] = location
if map_zoom is not None:
payload["map_zoom"] = map_zoom
places: dict = {}
routes: dict = {}
result: dict = {
"complete": False, # True once `end` arrives
"session_id": session_id,
"text": "",
"places": [],
"actions": [],
"routes": [],
"prompt_options": None,
"quota": None,
"end": None,
"error": None,
}
deadline = time.monotonic() + TOTAL_SECONDS
def tick() -> None:
if should_stop and should_stop():
raise KaleidrError("Stopped by the caller", code="aborted")
if time.monotonic() > deadline:
raise KaleidrError("The stream ran past its time limit", code="total_timeout")
headers = {"Authorization": f"Bearer {key}", "Accept": "text/event-stream"}
# `read` is the longest wait between two pieces of data: the idle timeout.
timeout = httpx.Timeout(connect=10.0, read=IDLE_SECONDS, write=10.0, pool=10.0)
with httpx.Client(timeout=timeout) as client:
for attempt in range(1, MAX_ATTEMPTS + 1):
started = False
try:
with client.stream("POST", f"{API}/chat/control/stream", headers=headers, json=payload) as response:
if response.status_code != 200:
response.read()
refusal = _refusal(response)
retryable = (refusal.status == 429 and refusal.code in RETRYABLE_429) or refusal.status >= 500
if not retryable or attempt == MAX_ATTEMPTS:
raise refusal
time.sleep((refusal.retry_after or 2 ** attempt) + random.random())
continue
started = True
for event, raw in sse_frames(response.iter_bytes(), tick):
try:
data = json.loads(raw)
except ValueError:
continue # not JSON: skip the frame
if event is None:
if isinstance(data.get("text"), str):
result["text"] += data["text"]
if on_text:
on_text(data["text"])
elif event == "place":
name = data.get("name")
if not isinstance(name, str):
continue
places[name] = _merge_place(places[name], data) if name in places else data
if on_place:
on_place(places[name])
elif event == "place_linked":
name = data.get("name")
if name not in places:
continue # detail for a place we never saw
places[name] = _merge_place(places[name], data)
if on_place:
on_place(places[name])
elif event == "route_clear":
routes.clear()
elif event == "route":
routes[data.get("id") or f"route-{len(routes)}"] = data
elif event == "prompt_options":
result["prompt_options"] = data
elif event in ("phase", "start"):
result["session_id"] = data.get("session_id") or result["session_id"]
elif event == "end":
result["complete"] = True
result["end"] = data
if isinstance(data.get("full_text"), str):
result["text"] = data["full_text"]
for place in data.get("places") or []:
name = place.get("name") if isinstance(place, dict) else None
if isinstance(name, str):
places[name] = _merge_place(places[name], place) if name in places else place
for name in data.get("dropped_places") or []:
places.pop(name, None)
result["actions"] = data.get("actions") or []
elif event == "error":
result["error"] = data
elif event == "quota":
result["quota"] = data
# Any other event name is ignored on purpose.
break
except httpx.ReadTimeout as exc:
raise KaleidrError("No data received for 45 seconds", code="idle_timeout") from exc
except httpx.TransportError:
# Retry a connection failure only if no part of the stream was read.
if started or attempt == MAX_ATTEMPTS:
raise
time.sleep(2 ** attempt * 0.5 + random.random() * 0.5)
result["places"] = list(places.values())
result["routes"] = list(routes.values())
return result
if __name__ == "__main__":
import sys
sys.stdout.reconfigure(encoding="utf-8") # place names are not always in the console's default encoding
answer = ask_kaleidr(
os.environ["KALEIDR_KEY"],
[{"role": "user", "content": sys.argv[1] if len(sys.argv) > 1 else "Three bakeries near the Louvre"}],
session_id="demo-1",
on_text=lambda piece: print(piece, end="", flush=True),
)
print()
for place in answer["places"]:
coordinates = place.get("coordinates")
print("-", place["name"], coordinates if coordinates else "(no location)")
if answer["error"]:
print("The answer failed:", answer["error"].get("message"))
elif not answer["complete"]:
print("The stream closed before the answer finished.")
if answer["quota"]:
print("AI credits used this period:", answer["quota"]["used"], "of", answer["quota"]["limit"])
export KALEIDR_KEY="kld_sk_live_…"
python kaleidr_stream.py "Three bakeries near the Louvre"
You should see the same output as the Node client: text as it streams, then the places and the usage line.
What the clients return
Both return one result object when the stream closes. The Node names are shown; Python uses the same names in snake case.
| Field | Meaning |
|---|---|
complete | true when an end event arrived. If it is false and error is empty, the connection dropped. |
text | The answer. The concatenated prose until end arrives, then end.full_text. |
places | One record per place name, merged across place, place_linked and end.places, without the names in end.dropped_places. A place without coordinates has no confirmed location. |
actions | The map actions from end. |
routes | The routes sent on route events, cleared by route_clear. |
promptOptions | The prompt_options event, when the answer asked a question with choices. |
quota | The quota event, when one arrived. |
error | The error event, when the answer failed after it started. |
sessionId | The conversation id the server used. Send it on the next turn. |
end | The raw end event, for any field not surfaced above. |
A refusal before the stream starts is raised as KaleidrError, with status, code, body and the
Retry-After value.
Conversation history: trim old turns
Kaleidr does not store the transcript. Both clients send exactly the messages you pass, so a follow-up
question means sending the earlier turns again, with the same session_id. If you keep appending and never
remove anything, two things happen:
- Each request costs more. Every turn you resend is read again as input and charged. A stream you abort is charged for at least the conversation it sent, so a long history is paid for even when you close the stream early.
- The request is eventually refused. A request can carry at most 400 messages, 64,000 characters in one
message, and 200,000 characters in all messages together. Over a limit, the API answers 422 with
too_many_messages,message_too_longorconversation_too_longand produces no answer.
Trim before each request: keep the newest turns that fit a budget you choose, well under the limits, and drop
the oldest. Keep sending the same session_id after you trim.
// Add to kaleidr-stream.mjs.
/** Keep the newest turns that fit. The last message, the user's new question, is always kept. */
export function trimHistory(messages, maxMessages = 40, maxChars = 60_000) {
const kept = [];
let chars = 0;
for (let i = messages.length - 1; i >= 0; i--) {
const size = messages[i].content.length;
if (kept.length > 0 && (kept.length >= maxMessages || chars + size > maxChars)) break;
kept.unshift(messages[i]);
chars += size;
}
while (kept.length > 1 && kept[0].role !== 'user') kept.shift(); // start on a user turn
return kept;
}
# Add to kaleidr_stream.py.
def trim_history(messages: list, max_messages: int = 40, max_chars: int = 60_000) -> list:
"""Keep the newest turns that fit. The last message, the user's new question, is always kept."""
kept: list = []
chars = 0
for message in reversed(messages):
size = len(message["content"])
if kept and (len(kept) >= max_messages or chars + size > max_chars):
break
kept.insert(0, message)
chars += size
while len(kept) > 1 and kept[0]["role"] != "user": # start on a user turn
kept.pop(0)
return kept
A conversation loop then looks like this:
const history = [];
async function ask(question) {
history.push({ role: 'user', content: question });
const answer = await askKaleidr({
key: process.env.KALEIDR_KEY,
sessionId: 'trip-42', // the same id for the whole conversation
messages: trimHistory(history),
});
if (answer.complete) history.push({ role: 'assistant', content: answer.text });
return answer;
}
The 40 messages and 60,000 characters are example budgets, not API values: choose yours. Trimming cannot
rescue a single message longer than 64,000 characters; shorten it before you send it. Store the assistant's
turn from answer.text only when complete is true, so a failed answer is not sent back as history.
Timeouts
| Timeout | Value in these clients | Reason |
|---|---|---|
| Idle: longest silence allowed | 45 seconds | The server sends a keep-alive comment after 10 seconds of silence, so 45 seconds with no bytes means the connection is dead. Any bytes reset the timer, comments included. |
| Total | 11 minutes | The server ends a stream itself after about 10 minutes with an error event whose code is upstream_stall. The client limit is a backstop a little above that. |
| Connect (Python) | 10 seconds | A connection that cannot be opened fails fast and is retried. |
Do not put a 30-second or 60-second total timeout on this request, in the client or in a proxy in front of it. Answers that involve several places or a route can run longer.
If your own server relays the stream to a browser, turn off response buffering on that route and send the frames through as they arrive, comments included, so the browser's connection stays alive.
Aborting
- Node: pass an
AbortSignalassignaland abort it. The pending read rejects with the reason you gave. - Python: pass
should_stop, a function that returnsTruewhen the caller has lost interest. It is checked whenever bytes arrive, which is at least every 10 seconds, and raisesKaleidrErrorwith the codeaborted. Leaving thewithblock closes the connection.
Closing the connection is the abort. The server notices within about ten seconds and stops producing the answer. An aborted stream is still charged. The charge is an estimate: at least the size of the conversation you sent, and otherwise what had already been produced. It can therefore be more than the hold placed at the start, and aborting early does not make a long request free. See how a stream is charged.
Abort when the user sends a new question, closes the panel or leaves the page. Every open stream also counts toward your organization's concurrent requests.
Retries
| Situation | Retry? | How |
|---|---|---|
429 with error of rate_limited, org_overloaded, b2b_overloaded or limiter_unavailable | Yes | Wait the Retry-After seconds plus jitter. The clients do this up to three times. |
429 with error: "quota_exceeded" | No | Tell the user. body.reset_at says when the period ends. |
| 401 or 403 | No | The credential or its permissions must change. |
| 422 | No | The request is malformed, or over a size limit. For conversation_too_long or too_many_messages, trim the history and send again. |
| 413 | No | The request body is larger than 2 MiB. Trim the history. |
| 5xx before the stream starts | Yes | A few times, with increasing waits. |
| Connection failure before any byte of the stream | Yes | A few times, with increasing waits. |
error event with code: "upstream_stall" | Let the user decide | Offer a "Try again" control. A retry is a new, charged request. |
error event without a code | Let the user decide | Show message. |
Idle timeout, or a stream that closed without end | Let the user decide | Keep the partial text on screen. Do not retry in a loop. |
The clients retry automatically only in the "Yes" rows, all of which happen before any part of the answer has been received. After the first frame they never retry by themselves, because the request has already been charged and the user has already seen part of an answer.
What to do with each event in a map UI
| Event | Do this |
|---|---|
place with source: "vendor", before phase | Draw the pin at once. These are places from your own data; you may want a distinct pin style. |
phase | Show a "thinking" indicator. Store session_id if you did not send one. |
: keep-alive | Nothing visible. Reset the idle timer. |
prompt_options | Show question with one button per entry in options. When a button is pressed, send its send_text as the next user message in the same conversation. |
route_clear | Remove the routes you have drawn. |
route | Draw primary.coordinates, a list of [lng, lat] pairs, as a line. List the steps from primary.turn_by_turn. Show attribution.text, linked to attribution.url, on or beside the map. |
A card or layer event, such as weather_card | Optional. Render it if you support it; otherwise ignore it. |
start | Change the indicator from "thinking" to "writing". |
| Prose (no event name) | Append text to the message. Remove the [[ and ]] markers, or turn each marked name into a link that focuses its pin. |
place | Add a marker at coordinates, keyed by name. If the name already has a marker, update its details and leave it where it is. If there are no coordinates, add no marker. |
place_linked | Update the details of the marker with that name. |
early_actions | Apply map.fitBounds to frame the places found so far. Apply places.highlight if you have a highlighted state. Skip action types you do not handle. |
end | Replace the message text with full_text. Reconcile markers with places, and remove the markers named in dropped_places. Apply actions. Remove the indicator. |
error | Remove the indicator. Show message. Keep the text already shown, and offer a retry. |
quota | With a server key: update a usage meter from used and limit, if you show one. With a session token the event has no used or limit; ignore it or read reset_at. |
| Any other event name | Ignore it. |
Two cautions when you move the map:
- Coordinates come in two shapes. A place is
{ lat, lng }. A line or a bounds list is[lng, lat]pairs. Passing one where the other is expected puts markers in the wrong hemisphere. early_actionsarrive while the user may be panning. If the user has moved the map since the question was sent, consider skipping the camera move and keeping only the pins.
In a browser
The same parsing works in a web page, with four differences:
- Send a session token (
kld_sess_…) in place of the server key. A server key is refused from a browser. See Auth and scopes. - Read the body with
res.body.getReader()and awhileloop.for awaitover a response body is not available in every browser. - Do not add custom request headers. Only
Authorization,Content-Type,Accept,X-Api-KeyandX-Request-Idpass the CORS preflight. - Do not read usage numbers. With a session token, the
quotaevent carries onlytype,meterandreset_at, and a 429quota_exceededbody has notier,limit,usedorremaining. Showreset_atto the visitor. See Quota and rate limits.
If you only want Kaleidr's chat on your map, you do not need any of this: mount the chat panel with
kaleidr.js and it reads the stream for you. See Chat.
Check that it worked
- The scripts print text progressively, not in one block at the end.
- A question that names places produces at least one place with
coordinates. completeistrueanderroris empty.- Running the script with a wrong key prints
Kaleidr refused the request: 401 Invalid or missing API keyand makes exactly one request.
If it does not work
| Symptom | Cause | Fix |
|---|---|---|
JSON.parse or json.loads fails on a frame | The client parsed a network chunk instead of a complete line. | Split on line breaks and keep the unfinished remainder for the next chunk, as sseFrames does. |
| Garbled characters in place names | Bytes were decoded chunk by chunk, splitting a multi-byte character. | Decode with a streaming decoder (Node) or decode whole lines (Python). |
UnicodeEncodeError when printing on Windows | The console's encoding cannot show the character. | Set standard output to UTF-8, as the Python example does. |
The stream ends after about a minute with no end | A total timeout in your HTTP client, framework or proxy. | Use an idle timeout instead, and raise the proxy's limit. |
| Text arrives all at once | A proxy or framework is buffering the response. | Turn off buffering for the route. |
The client never sees quota | It stopped reading at end. | Read until the body closes. |
403 server_key_in_browser from server code | Your HTTP client or a gateway adds an Origin header. | Remove the header. |
429 org_overloaded under load | Too many streams open at once for the organization. | Abort streams the user has left; queue new questions in your own code. |
422 conversation_too_long after many turns | The history you resend has passed 200,000 characters. | Trim old turns before each request. |
Related
- SSE wire contract: every event and field.
- Endpoints: the request body.
- Quota and rate limits: charging, 429 bodies and limits.
- Errors: every status and code.