feat(agents-realtime): add fetch-based WebSocket upgrade for Cloudflare/workerd

[agata]

Summary

• Adds useWorkersFetchUpgrade option to OpenAIRealtimeWebSocket for Cloudflare/workerd (Cloudflare Workers: WebSockets — Make outbound WebSocket connections).
• Implements connectViaFetchUpgrade() to establish WS via fetch() + Upgrade: websocket (see same docs above).
• Keeps default constructor-based WebSocket path unchanged for other environments.

Motivation

Cloudflare Workers and other workerd runtimes (e.g., Miniflare) cannot create client WebSockets via the global WebSocket constructor. Instead, Cloudflare’s documentation specifies using an HTTP fetch() upgrade flow that returns response.webSocket, which must be accept()ed (docs). This PR enables first-class support for that environment. For additional runtime context, see Durable Objects and WebSockets.

Changes

• Option: useWorkersFetchUpgrade?: boolean on OpenAIRealtimeWebSocketOptions.
• Method: connectViaFetchUpgrade() performs the GET + upgrade headers and accepts response.webSocket.
• Connection flow: in fetch-upgrade mode, transition to connected without relying on an 'open' event.
• Errors: failed upgrades include HTTP status and response text for easier debugging.
• Comments: simplify option/property comments; keep detailed explanation on connectViaFetchUpgrade().
• Tests: add coverage for fetch-upgrade success/failure.

Before/After

• Before: workerd environments couldn’t connect using the default constructor-based WebSocket flow.
• After: set useWorkersFetchUpgrade: true to connect under Cloudflare/workerd while preserving default behavior elsewhere.

Example:

const ws = new OpenAIRealtimeWebSocket({
  url: 'ws://test',
  useWorkersFetchUpgrade: true,
});
await ws.connect({ apiKey: 'ek_...', model: 'gpt-4o-realtime' });

Tests

• File: packages/agents-realtime/test/openaiRealtimeWebsocket.test.ts
  • Verifies URL transform (ws: → http:), required headers, and connection transitions in fetch-upgrade mode.
  • Validates error propagation when response.webSocket is missing.
  • Existing WebSocket tests remain passing; test URLs unified.

Docs

• Inlined JSDoc updates: option/property one-liners; detailed method docs on connectViaFetchUpgrade().

Breaking Impact

• None. Default path remains unchanged; the new option is opt-in.

Checklist

• Builds and full test suite pass
• Docs updated
• Changeset added (minor bump for @openai/agents-realtime)

[changeset-bot]

🦋 Changeset detected

Latest commit: eca41e6

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 3 packages

Name	Type
@openai/agents-realtime	Patch
@openai/agents-extensions	Patch
@openai/agents	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[seratch]

Thanks for sending this PR. This project already has TwilioRealtimeTransportLayer in packages/agents-extensions. Having WorkerdRealtimeTransportLayer in the extensions would be cleaner.

[agata]

Adding it to the extensions would definitely be ideal! For context, I previously tried an approach that did not modify OpenAIRealtimeWebSocket.

https://github.com/agata/openai-agents-js/blob/feat/cloudflare-realtime-websocket/packages/agents-realtime/src/cloudflareRealtimeTransport.ts

However, I abandoned that approach for these reasons:

Aside from the timing of WebSocket creation and the post-open handling, the implementation is identical. Copy-pasting would create a large amount of duplicated code, which would be hard to maintain going forward. In particular, event emission is a core behavior of the library, so duplicating it felt wrong.

Utilities like HEADERS and base64ToArrayBuffer are not exported outside the package, which makes it difficult to implement this in the extensions package (in the example above, I ended up re-implementing them inside agents-realtime).

I see a few possible approaches:

A. Allow the OpenAIRealtimeWebSocket constructor to accept a WebSocket factory and post-initialization hook(s), so the creation and “on open” lifecycle can be injected. (This would slightly complicate the public constructor interface.)

B. Export utilities like HEADERS and base64ToArrayBuffer so they’re usable from extensions, then copy OpenAIRealtimeWebSocket and implement CloudflareRealtimeTransport on the extensions side. (This would work but increases maintenance burden due to duplication.)

Which approach do you think would fit the project best? I’m also open to other suggestions if there’s a cleaner path I’m missing.

[seratch]

I prefer the option A. I haven't verified this really works but what do you think about something like this?

diff --git a/packages/agents-extensions/src/CloudflareRealtimeTransport.ts b/packages/agents-extensions/src/CloudflareRealtimeTransport.ts
new file mode 100644
index 0000000..3778f1c
--- /dev/null
+++ b/packages/agents-extensions/src/CloudflareRealtimeTransport.ts
@@ -0,0 +1,50 @@
+import {
+  RealtimeTransportLayer,
+  OpenAIRealtimeWebSocket,
+  OpenAIRealtimeWebSocketOptions,
+} from '@openai/agents/realtime';
+
+export class CloudflareRealtimeTransportLayer
+  extends OpenAIRealtimeWebSocket
+  implements RealtimeTransportLayer {
+  protected _audioLengthMs: number = 0;
+
+  constructor(options: OpenAIRealtimeWebSocketOptions) {
+    super({
+      ...options,
+      createWebSocket: async ({ url, apiKey }) => {
+        return await this.#buildCloudflareWebSocket({ url, apiKey });
+      },
+      skipOpenEventListeners: true,
+    });
+  }
+
+  async #buildCloudflareWebSocket({ url, apiKey }: { url: string, apiKey: string }): Promise<WebSocket> {
+    const transformedUrl = url.replace(/^ws/i, 'http');
+    if (!transformedUrl) {
+      throw new Error('Realtime URL is not defined');
+    }
+
+    const response = await fetch(transformedUrl, {
+      method: 'GET',
+      headers: {
+        Authorization: `Bearer ${apiKey}`,
+        'Sec-WebSocket-Protocol': 'realtime',
+        Connection: 'Upgrade',
+        Upgrade: 'websocket',
+        ...this.getCommonRequestHeaders(),
+      },
+    });
+
+    const upgradedSocket = (response as any).webSocket;
+    if (!upgradedSocket) {
+      const body = await response.text().catch(() => '');
+      throw new Error(
+        `Failed to upgrade websocket: ${response.status} ${body}`,
+      );
+    }
+
+    upgradedSocket.accept();
+    return upgradedSocket as unknown as WebSocket;
+  }
+}
diff --git a/packages/agents-extensions/src/index.ts b/packages/agents-extensions/src/index.ts
index b1cafe0..3b571f4 100644
--- a/packages/agents-extensions/src/index.ts
+++ b/packages/agents-extensions/src/index.ts
@@ -1,2 +1,3 @@
-export * from './TwilioRealtimeTransport';
 export * from './aiSdk';
+export * from './CloudflareRealtimeTransport';
+export * from './TwilioRealtimeTransport';
diff --git a/packages/agents-realtime/src/openaiRealtimeWebsocket.ts b/packages/agents-realtime/src/openaiRealtimeWebsocket.ts
index 22f1ba1..6ee0d85 100644
--- a/packages/agents-realtime/src/openaiRealtimeWebsocket.ts
+++ b/packages/agents-realtime/src/openaiRealtimeWebsocket.ts
@@ -35,6 +35,11 @@ export type WebSocketState =
       websocket: WebSocket;
     };
 
+export interface CreateWebSocketOptions {
+  url: string;
+  apiKey: string;
+}
+
 /**
  * The options for the OpenAI Realtime WebSocket transport layer.
  */
@@ -51,13 +56,25 @@ export type OpenAIRealtimeWebSocketOptions = {
    * The URL to use for the WebSocket connection.
    */
   url?: string;
+
   /**
-   * Cloudflare Workers/workerd only: enable the fetch-based WebSocket upgrade.
-   * Set to `true` only on Cloudflare (including Miniflare). Other environments
-   * use the default `WebSocket` constructor path. See `connectViaFetchUpgrade()`
-   * for details.
+   * Builds a new WebSocket connection.
+   * @param options - The options for the WebSocket connection.
+   * @returns The WebSocket connection.
+   */
+  createWebSocket?: (options: CreateWebSocketOptions) => Promise<WebSocket>;
+
+  /**
+   * When you pass your own createWebSocket function, which completes the connection state transition,
+   * you can set this to true to skip registering the `open` event listener for the same purpose.
+   * If this flag is set to true, the constructor will immediately call the internal operation
+   * to mark the internal connection state to `connected`. Otherwise, the constructor will register
+   * the `open` event listener and wait for it to be triggered.
+   *
+   * By default (meaning if this property is absent), this is set to false.
    */
-  useWorkersFetchUpgrade?: boolean;
+  skipOpenEventListeners?: boolean;
+
 } & OpenAIRealtimeBaseOptions;
 
 /**
@@ -71,8 +88,6 @@ export class OpenAIRealtimeWebSocket
 {
   #apiKey: string | undefined;
   #url: string | undefined;
-  /** Cloudflare/workerd only; see `connectViaFetchUpgrade()`. */
-  useWorkersFetchUpgrade?: boolean;
   #state: WebSocketState = {
     status: 'disconnected',
     websocket: undefined,
@@ -91,11 +106,19 @@ export class OpenAIRealtimeWebSocket
   protected _audioLengthMs: number = 0;
   #ongoingResponse: boolean = false;
 
+  #createWebSocket?: (options: CreateWebSocketOptions) => Promise<WebSocket>;
+  #skipOpenEventListeners?: boolean;
+
   constructor(options: OpenAIRealtimeWebSocketOptions = {}) {
     super(options);
     this.#url = options.url;
     this.#useInsecureApiKey = options.useInsecureApiKey ?? false;
-    this.useWorkersFetchUpgrade = options.useWorkersFetchUpgrade ?? false;
+    this.#createWebSocket = options.createWebSocket;
+    this.#skipOpenEventListeners = options.skipOpenEventListeners ?? false;
+  }
+
+  protected getCommonRequestHeaders() {
+    return HEADERS;
   }
 
   /**
@@ -166,8 +189,8 @@ export class OpenAIRealtimeWebSocket
 
     let ws: WebSocket | null = null;
 
-    if (this.useWorkersFetchUpgrade) {
-      ws = await this.connectViaFetchUpgrade();
+    if (this.#createWebSocket) {
+      ws = await this.#createWebSocket({ url: this.#url!, apiKey: this.#apiKey });
     } else {
       // browsers and workerd should use the protocols argument, node should use the headers argument
       const websocketArguments = useWebSocketProtocols
@@ -181,7 +204,7 @@ export class OpenAIRealtimeWebSocket
         : {
             headers: {
               Authorization: `Bearer ${this.#apiKey}`,
-              ...HEADERS,
+              ...this.getCommonRequestHeaders(),
             },
           };
 
@@ -203,7 +226,9 @@ export class OpenAIRealtimeWebSocket
       resolve();
     };
 
-    if (this.useWorkersFetchUpgrade) {
+    if (this.#skipOpenEventListeners === true) {
+      // Some platforms like Cloudflare's fetch-upgrade mode do not rely on `open` events
+      // for transitioning the connection state.
       onSocketOpenReady();
     } else {
       ws.addEventListener('open', onSocketOpenReady);
@@ -439,46 +464,4 @@ export class OpenAIRealtimeWebSocket
     this._audioLengthMs = 0;
     this.#currentAudioContentIndex = undefined;
   }
-
-  /**
-   * Connect using the Cloudflare/workerd Fetch-based WebSocket upgrade path.
-   *
-   * workerd does not support creating client WebSockets via the global `WebSocket`
-   * constructor. Instead, a `fetch()` request with `Upgrade: websocket` is issued and
-   * the upgraded socket is returned as `response.webSocket`, which must be accepted
-   * via `accept()`.
-   *
-   * This method is intended for use when `useWorkersFetchUpgrade` is `true` and the
-   * code is running under Cloudflare Workers (or compatible workerd runtimes like
-   * Miniflare). Other environments should use the normal constructor-based path.
-   */
-  async connectViaFetchUpgrade(): Promise<WebSocket> {
-    const transformedUrl = this.#url?.replace(/^ws/i, 'http');
-    if (!transformedUrl) {
-      throw new Error('Realtime URL is not defined');
-    }
-
-    const response = await fetch(transformedUrl, {
-      method: 'GET',
-      headers: {
-        Authorization: `Bearer ${this.#apiKey}`,
-        // 'OpenAI-Beta': 'realtime=v1',
-        'Sec-WebSocket-Protocol': 'realtime',
-        Connection: 'Upgrade',
-        Upgrade: 'websocket',
-        ...HEADERS,
-      },
-    });
-
-    const upgradedSocket = (response as any).webSocket;
-    if (!upgradedSocket) {
-      const body = await response.text().catch(() => '');
-      throw new Error(
-        `Failed to upgrade websocket: ${response.status} ${body}`,
-      );
-    }
-
-    upgradedSocket.accept();
-    return upgradedSocket as unknown as WebSocket;
-  }
 }

[agata]

I thought this was an ideal solution. I actually applied it in a real project, and it’s working without any issues. Thank you very much.

Will this fix be incorporated on your side, or should I go ahead and make the changes myself?

[seratch]

If you update this PR, we're happy to have the change as your contribution! 🙌 If you're busy for other things now, I can take time to complete this on behalf of you.

[seratch]

If you have document updates in this PR, we have to release a new version including the change right after merging the PR. Can you make document changes in a separate PR?

[agata]

Got it. I’ll create a separate PR.

[seratch]

Looks a bit long in the side bar; how about "Realtime Agents on Cloudflare"?

[agata]

Yes, I’ll do that.

[seratch]

the latest version supports only gpt-realtime model

[agata]

I’ll fix it.

[seratch]

delete this comment

[agata]

I’ve fixed it.

[agata]

I’ve incorporated your patch and recreated the pull request.
I had also included some documentation changes, but I’ve discarded those for now.

In addition to the patch, I’ve added tests and updated the JSDoc.
I also verified the final behavior in a real application running on Cloudflare, and confirmed once again that it’s working without any issues.