Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | 24x 8x 8x 8x 8x 8x 2x 6x 24x 5x 1x 2x 16x 16x 16x 16x 16x 16x 17x 16x 16x 16x 16x 16x 16x 18x 18x 16x 16x 1x 15x 16x 15x 15x 14x 16x 2x 1x 1x | import EventSource from "react-native-sse";
import { mapAguiEvent } from "./agui-map";
import type {
ActionResult,
AgentTransport,
StreamHandlers,
Thread,
TransportConfig,
TurnInput,
} from "./transport";
/**
* SSE implementation of the AgentTransport port. This is the ONLY file allowed
* to import `react-native-sse`. Everything above it (hook, UI) depends solely
* on the `AgentTransport` interface + `AgentEvent`, so a future CopilotKit
* transport is a drop-in sibling of this file.
*
* REST (thread/confirm/cancel) is plain JSON `fetch`; `streamTurn` opens a POST
* EventSource against the native AG-UI route and maps each frame to AgentEvent.
*/
export function createSseTransport(config: TransportConfig): AgentTransport {
const { baseUrl, getToken } = config;
async function authHeaders(json: boolean): Promise<Record<string, string>> {
const token = await getToken();
return {
...(json ? { "Content-Type": "application/json" } : {}),
Accept: "application/json",
...(token ? { Authorization: `Bearer ${token}` } : {}),
};
}
async function postJson<T>(path: string, body: unknown): Promise<T> {
const response = await fetch(`${baseUrl}${path}`, {
method: "POST",
headers: await authHeaders(true),
body: JSON.stringify(body ?? {}),
});
const data = (await response.json()) as T;
if (!response.ok) {
throw new Error(`${path} failed (${response.status})`);
}
return data;
}
return {
createThread(scope) {
return postJson<Thread>("/v1/chat/threads", scope ? { scope } : {});
},
confirm(threadId, pendingActionId) {
return postJson<ActionResult>(`/v1/chat/threads/${threadId}/actions/confirm`, {
pendingActionId,
});
},
cancel(threadId, pendingActionId) {
return postJson<ActionResult>(`/v1/chat/threads/${threadId}/actions/cancel`, {
pendingActionId,
});
},
async streamTurn(input: TurnInput, handlers: StreamHandlers): Promise<void> {
const token = await getToken();
const url = `${baseUrl}/v1/chat/threads/${input.threadId}/messages`;
await new Promise<void>((resolve) => {
const source = new EventSource(url, {
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "text/event-stream",
...(token ? { Authorization: `Bearer ${token}` } : {}),
},
body: JSON.stringify({
message: input.message,
...(input.system ? { system: input.system } : {}),
...(input.images?.length ? { images: input.images } : {}),
}),
// One-shot turn: no auto-reconnect / heartbeat polling.
pollingInterval: 0,
});
let settled = false;
const finish = (): void => {
if (settled) return;
settled = true;
source.removeAllEventListeners();
source.close();
resolve();
};
handlers.signal?.addEventListener("abort", finish);
source.addEventListener("message", (event) => {
const data = event.data;
if (!data || data === "[DONE]") return;
let parsed: unknown;
try {
parsed = JSON.parse(data);
} catch {
return; // ignore keep-alives / non-JSON frames
}
for (const agentEvent of mapAguiEvent(parsed)) {
handlers.onEvent(agentEvent);
}
const frameType = (parsed as { type?: string }).type;
if (frameType === "RUN_FINISHED" || frameType === "RUN_ERROR") {
finish();
}
});
source.addEventListener("error", () => {
// A close after RUN_FINISHED also surfaces here — already settled, so
// this only fires for a genuine transport failure before completion.
if (!settled) {
handlers.onEvent({
_tag: "RunError",
message: "Connection to the agent failed.",
});
finish();
}
});
});
},
};
}
|