From 997c32245a7aca7fc8239752f4f2bcd7440534df Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 15 Feb 2026 11:52:56 +0000 Subject: [PATCH] feat: support dynamic accessToken function for token refresh The accessToken option now accepts either a string or a function returning a string. This enables dynamic token refresh patterns: new TriggerChatTransport({ taskId: 'my-task', accessToken: () => getLatestToken(), }) The function is called on each sendMessages() call, allowing fresh tokens to be used for each task trigger. Co-authored-by: Eric Allam --- packages/ai/src/transport.test.ts | 85 +++++++++++++++++++++++++++++++ packages/ai/src/transport.ts | 22 +++++--- packages/ai/src/types.ts | 18 +++++-- 3 files changed, 113 insertions(+), 12 deletions(-) diff --git a/packages/ai/src/transport.test.ts b/packages/ai/src/transport.test.ts index cc946596c..53d3ab868 100644 --- a/packages/ai/src/transport.test.ts +++ b/packages/ai/src/transport.test.ts @@ -77,6 +77,19 @@ describe("TriggerChatTransport", () => { expect(transport).toBeInstanceOf(TriggerChatTransport); }); + + it("should accept a function for accessToken", () => { + let tokenCallCount = 0; + const transport = new TriggerChatTransport({ + taskId: "my-chat-task", + accessToken: () => { + tokenCallCount++; + return `dynamic-token-${tokenCallCount}`; + }, + }); + + expect(transport).toBeInstanceOf(TriggerChatTransport); + }); }); describe("sendMessages", () => { @@ -627,6 +640,78 @@ describe("TriggerChatTransport", () => { }); }); + describe("dynamic accessToken", () => { + it("should call the accessToken function for each sendMessages call", async () => { + let tokenCallCount = 0; + + global.fetch = vi.fn().mockImplementation(async (url: string | URL) => { + const urlStr = typeof url === "string" ? url : url.toString(); + + if (urlStr.includes("/trigger")) { + return new Response( + JSON.stringify({ id: `run_dyn_${tokenCallCount}` }), + { + status: 200, + headers: { + "content-type": "application/json", + "x-trigger-jwt": "stream-token", + }, + } + ); + } + + if (urlStr.includes("/realtime/v1/streams/")) { + const chunks: UIMessageChunk[] = [ + { type: "text-start", id: "p1" }, + { type: "text-end", id: "p1" }, + ]; + return new Response(createSSEStream(sseEncode(chunks)), { + status: 200, + headers: { + "content-type": "text/event-stream", + "X-Stream-Version": "v1", + }, + }); + } + + throw new Error(`Unexpected fetch URL: ${urlStr}`); + }); + + const transport = new TriggerChatTransport({ + taskId: "my-task", + accessToken: () => { + tokenCallCount++; + return `dynamic-token-${tokenCallCount}`; + }, + baseURL: "https://api.test.trigger.dev", + }); + + // First call — the token function should be invoked + await transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-dyn-1", + messageId: undefined, + messages: [createUserMessage("first")], + abortSignal: undefined, + }); + + const firstCount = tokenCallCount; + expect(firstCount).toBeGreaterThanOrEqual(1); + + // Second call — the token function should be invoked again + await transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-dyn-2", + messageId: undefined, + messages: [createUserMessage("second")], + abortSignal: undefined, + }); + + // Token function was called at least once more + expect(tokenCallCount).toBeGreaterThan(firstCount); + }); + }); + describe("body merging", () => { it("should merge ChatRequestOptions.body into the task payload", async () => { const fetchSpy = vi.fn().mockImplementation(async (url: string | URL) => { diff --git a/packages/ai/src/transport.ts b/packages/ai/src/transport.ts index b9df702b3..d785a471c 100644 --- a/packages/ai/src/transport.ts +++ b/packages/ai/src/transport.ts @@ -66,12 +66,11 @@ const DEFAULT_STREAM_TIMEOUT_SECONDS = 120; */ export class TriggerChatTransport implements ChatTransport { private readonly taskId: string; - private readonly accessToken: string; + private readonly resolveAccessToken: () => string; private readonly baseURL: string; private readonly streamKey: string; private readonly extraHeaders: Record; private readonly streamTimeoutSeconds: number; - private readonly apiClient: ApiClient; /** * Tracks active chat sessions for reconnection support. @@ -81,12 +80,18 @@ export class TriggerChatTransport implements ChatTransport { constructor(options: TriggerChatTransportOptions) { this.taskId = options.taskId; - this.accessToken = options.accessToken; + this.resolveAccessToken = + typeof options.accessToken === "function" + ? options.accessToken + : () => options.accessToken as string; this.baseURL = options.baseURL ?? DEFAULT_BASE_URL; this.streamKey = options.streamKey ?? DEFAULT_STREAM_KEY; this.extraHeaders = options.headers ?? {}; this.streamTimeoutSeconds = options.streamTimeoutSeconds ?? DEFAULT_STREAM_TIMEOUT_SECONDS; - this.apiClient = new ApiClient(this.baseURL, this.accessToken); + } + + private getApiClient(): ApiClient { + return new ApiClient(this.baseURL, this.resolveAccessToken()); } /** @@ -118,8 +123,11 @@ export class TriggerChatTransport implements ChatTransport { ...(body ?? {}), }; + const currentToken = this.resolveAccessToken(); + // Trigger the task - const triggerResponse = await this.apiClient.triggerTask(this.taskId, { + const apiClient = this.getApiClient(); + const triggerResponse = await apiClient.triggerTask(this.taskId, { payload: JSON.stringify(payload), options: { payloadType: "application/json", @@ -135,11 +143,11 @@ export class TriggerChatTransport implements ChatTransport { // Store session state for reconnection this.sessions.set(chatId, { runId, - publicAccessToken: publicAccessToken ?? this.accessToken, + publicAccessToken: publicAccessToken ?? currentToken, }); // Subscribe to the realtime stream for this run - return this.subscribeToStream(runId, publicAccessToken ?? this.accessToken, abortSignal); + return this.subscribeToStream(runId, publicAccessToken ?? currentToken, abortSignal); }; /** diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index bbffb5024..88bf43173 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -11,13 +11,21 @@ export type TriggerChatTransportOptions = { taskId: string; /** - * A public access token or trigger token for authenticating with the Trigger.dev API. - * This is used both to trigger the task and to subscribe to the realtime stream. + * An access token for authenticating with the Trigger.dev API. * - * You can generate one using `auth.createTriggerPublicToken()` or - * `auth.createPublicToken()` from the `@trigger.dev/sdk`. + * This must be a token with permission to trigger the task. You can use: + * - A **trigger public token** created via `auth.createTriggerPublicToken(taskId)` (recommended for frontend use) + * - A **secret API key** (for server-side use only — never expose in the browser) + * + * The token returned from triggering the task (`publicAccessToken`) is automatically + * used for subscribing to the realtime stream. + * + * Can also be a function that returns a token string, useful for dynamic token refresh: + * ```ts + * accessToken: () => getLatestToken() + * ``` */ - accessToken: string; + accessToken: string | (() => string); /** * Base URL for the Trigger.dev API.