Implement Event-Source Send in CopilotStudioClient to handle stream responses (#614)

* Use eventSource to handle streaming responses

* Apply feedback

* Update API docs

* Fix lint issue

* Remove resolved and integrity from package-lock

Author: ceciliaavila

compat/baseline/agents-copilotstudio-client.api.md

@@ -25,10 +25,10 @@ export class ConnectionSettings extends ConnectionOptions {
 // @public
 export class CopilotStudioClient {
     constructor(settings: ConnectionSettings, token: string);
-    askQuestionAsync(question: string, conversationId?: string): Promise<Activity[]>;
+    askQuestionAsync(question: string, conversationId?: string): AsyncGenerator<Activity>;
     static scopeFromSettings: (settings: ConnectionSettings) => string;
-    sendActivity(activity: Activity, conversationId?: string): Promise<Activity[]>;
-    startConversationAsync(emitStartConversationEvent?: boolean): Promise<Activity>;
+    sendActivity(activity: Activity, conversationId?: string): AsyncGenerator<Activity>;
+    startConversationAsync(emitStartConversationEvent?: boolean): AsyncGenerator<Activity>;
 }
 
 // @public

package-lock.json

@@ -28,7 +28,7 @@
   },
   "dependencies": {
     "@microsoft/agents-activity": "file:../agents-activity",
-    "axios": "^1.12.2",
+    "eventsource-client": "^1.2.0",
     "rxjs": "7.8.2",
     "uuid": "^11.1.0"
   },

packages/agents-copilotstudio-client/package.json

@@ -3,8 +3,8 @@
  * Licensed under the MIT License.
  */
 
+import { createEventSource, EventSourceClient } from 'eventsource-client'
 import { ConnectionSettings } from './connectionSettings'
-import axios, { AxiosInstance, AxiosRequestConfig } from 'axios'
 import { getCopilotStudioConnectionUrl, getTokenAudience } from './powerPlatformEnvironment'
 import { Activity, ActivityTypes, ConversationAccount } from '@microsoft/agents-activity'
 import { ExecuteTurnRequest } from './executeTurnRequest'
@@ -14,11 +14,6 @@ import os from 'os'
 
 const logger = debug('copilot-studio:client')
 
-interface streamRead {
-  done: boolean,
-  value: string
-}
-
 /**
  * Client for interacting with Microsoft Copilot Studio services.
  * Provides functionality to start conversations and send messages to Copilot Studio bots.
@@ -33,8 +28,8 @@ export class CopilotStudioClient {
   private conversationId: string = ''
   /** The connection settings for the client. */
   private readonly settings: ConnectionSettings
-  /** The Axios instance used for HTTP requests. */
-  private readonly client: AxiosInstance
+  /** The authenticaton token. */
+  private readonly token: string
 
   /**
    * Returns the scope URL needed to connect to Copilot Studio from the connection settings.
@@ -51,79 +46,70 @@ export class CopilotStudioClient {
    */
   constructor (settings: ConnectionSettings, token: string) {
     this.settings = settings
-    this.client = axios.create()
-    this.client.defaults.headers.common.Authorization = `Bearer ${token}`
-    this.client.defaults.headers.common['User-Agent'] = CopilotStudioClient.getProductInfo()
+    this.token = token
   }
 
-  private async postRequestAsync (axiosConfig: AxiosRequestConfig): Promise<Activity[]> {
-    const activities: Activity[] = []
-
-    logger.debug(`>>> SEND TO ${axiosConfig.url}`)
-
-    const response = await this.client(axiosConfig)
-
-    if (this.settings.useExperimentalEndpoint && !this.settings.directConnectUrl?.trim()) {
-      const islandExperimentalUrl = response.headers?.[CopilotStudioClient.islandExperimentalUrlHeaderKey]
-      if (islandExperimentalUrl) {
-        this.settings.directConnectUrl = islandExperimentalUrl
-        logger.debug(`Island Experimental URL: ${islandExperimentalUrl}`)
-      }
-    }
-
-    this.conversationId = response.headers?.[CopilotStudioClient.conversationIdHeaderKey] ?? ''
-    if (this.conversationId) {
-      logger.debug(`Conversation ID: ${this.conversationId}`)
-    }
-
-    const sanitizedHeaders = { ...response.headers }
-    delete sanitizedHeaders['Authorization']
-    delete sanitizedHeaders[CopilotStudioClient.conversationIdHeaderKey]
-    logger.debug('Headers received:', sanitizedHeaders)
+  /**
+   * Streams activities from the Copilot Studio service using eventsource-client.
+   * @param url The connection URL for Copilot Studio.
+   * @param body Optional. The request body (for POST).
+   * @param method Optional. The HTTP method (default: POST).
+   * @returns An async generator yielding the Agent's Activities.
+   */
+  private async * postRequestAsync (url: string, body?: any, method: string = 'POST'): AsyncGenerator<Activity> {
+    logger.debug(`>>> SEND TO ${url}`)
 
-    const stream = response.data
-    const reader = stream.pipeThrough(new TextDecoderStream()).getReader()
-    let result: string = ''
-    const results: string[] = []
-
-    const processEvents = async ({ done, value }: streamRead): Promise<string[]> => {
-      if (done) {
-        logger.debug('Stream complete')
-        result += value
-        results.push(result)
-        return results
+    const eventSource: EventSourceClient = createEventSource({
+      url,
+      headers: {
+        Authorization: `Bearer ${this.token}`,
+        'User-Agent': CopilotStudioClient.getProductInfo(),
+        'Content-Type': 'application/json',
+        Accept: 'text/event-stream'
+      },
+      body: body ? JSON.stringify(body) : undefined,
+      method,
+      fetch: async (url, init) => {
+        const response = await fetch(url, init)
+        this.processResponseHeaders(response.headers)
+        return response
       }
-      logger.info('Agent is typing ...')
-      result += value
-
-      return await processEvents(await reader.read())
-    }
+    })
 
-    const events: string[] = await reader.read().then(processEvents)
-
-    events.forEach(event => {
-      const values: string[] = event.toString().split('\n')
-      const validEvents = values.filter(e => e.substring(0, 4) === 'data' && e !== 'data: end\r')
-      validEvents.forEach(ve => {
-        try {
-          const act = Activity.fromJson(ve.substring(5, ve.length))
-          if (act.type === ActivityTypes.Message) {
-            activities.push(act)
-            if (!this.conversationId.trim()) {
-              // Did not get it from the header.
-              this.conversationId = act.conversation?.id ?? ''
-              logger.debug(`Conversation ID: ${this.conversationId}`)
+    try {
+      for await (const { data, event } of eventSource) {
+        if (data && event === 'activity') {
+          try {
+            const activity = Activity.fromJson(data)
+            switch (activity.type) {
+              case ActivityTypes.Message:
+                if (!this.conversationId.trim()) { // Did not get it from the header.
+                  this.conversationId = activity.conversation?.id ?? ''
+                  logger.debug(`Conversation ID: ${this.conversationId}`)
+                }
+                yield activity
+                break
+              default:
+                logger.debug(`Activity type: ${activity.type}`)
+                yield activity
+                break
             }
-          } else {
-            logger.debug(`Activity type: ${act.type}`)
+          } catch (error) {
+            logger.error('Failed to parse activity:', error)
           }
-        } catch (e) {
-          logger.error('Error: ', e)
-          throw e
+        } else if (event === 'end') {
+          logger.debug('Stream complete')
+          break
         }
-      })
-    })
-    return activities
+
+        if (eventSource.readyState === 'closed') {
+          logger.debug('Connection closed')
+          break
+        }
+      }
+    } finally {
+      eventSource.close()
+    }
   }
 
   /**
@@ -146,41 +132,50 @@ export class CopilotStudioClient {
     return userAgent
   }
 
+  private processResponseHeaders (responseHeaders: Headers): void {
+    if (this.settings.useExperimentalEndpoint && !this.settings.directConnectUrl?.trim()) {
+      const islandExperimentalUrl = responseHeaders?.get(CopilotStudioClient.islandExperimentalUrlHeaderKey)
+      if (islandExperimentalUrl) {
+        this.settings.directConnectUrl = islandExperimentalUrl
+        logger.debug(`Island Experimental URL: ${islandExperimentalUrl}`)
+      }
+    }
+
+    this.conversationId = responseHeaders?.get(CopilotStudioClient.conversationIdHeaderKey) ?? ''
+    if (this.conversationId) {
+      logger.debug(`Conversation ID: ${this.conversationId}`)
+    }
+
+    const sanitizedHeaders = new Headers()
+    responseHeaders.forEach((value, key) => {
+      if (key.toLowerCase() !== 'authorization' && key.toLowerCase() !== CopilotStudioClient.conversationIdHeaderKey.toLowerCase()) {
+        sanitizedHeaders.set(key, value)
+      }
+    })
+    logger.debug('Headers received:', sanitizedHeaders)
+  }
+
   /**
    * Starts a new conversation with the Copilot Studio service.
    * @param emitStartConversationEvent Whether to emit a start conversation event. Defaults to true.
-   * @returns A promise that resolves to the initial activity of the conversation.
+   * @returns An async generator yielding the Agent's Activities.
    */
-  public async startConversationAsync (emitStartConversationEvent: boolean = true): Promise<Activity> {
+  public async * startConversationAsync (emitStartConversationEvent: boolean = true): AsyncGenerator<Activity> {
     const uriStart: string = getCopilotStudioConnectionUrl(this.settings)
     const body = { emitStartConversationEvent }
 
-    const config: AxiosRequestConfig = {
-      method: 'post',
-      url: uriStart,
-      headers: {
-        Accept: 'text/event-stream',
-        'Content-Type': 'application/json',
-      },
-      data: body,
-      responseType: 'stream',
-      adapter: 'fetch'
-    }
-
     logger.info('Starting conversation ...')
-    const values = await this.postRequestAsync(config)
-    const act = values[0]
-    logger.info(`Conversation '${act.conversation?.id}' started. Received ${values.length} activities.`, values)
-    return act
+
+    yield * this.postRequestAsync(uriStart, body, 'POST')
   }
 
   /**
    * Sends a question to the Copilot Studio service and retrieves the response activities.
    * @param question The question to ask.
    * @param conversationId The ID of the conversation. Defaults to the current conversation ID.
-   * @returns A promise that resolves to an array of activities containing the responses.
+   * @returns An async generator yielding the Agent's Activities.
    */
-  public async askQuestionAsync (question: string, conversationId: string = this.conversationId) {
+  public async * askQuestionAsync (question: string, conversationId: string = this.conversationId) : AsyncGenerator<Activity> {
     const conversationAccount: ConversationAccount = {
       id: conversationId
     }
@@ -191,34 +186,21 @@ export class CopilotStudioClient {
     }
     const activity = Activity.fromObject(activityObj)
 
-    return this.sendActivity(activity)
+    yield * this.sendActivity(activity)
   }
 
   /**
    * Sends an activity to the Copilot Studio service and retrieves the response activities.
    * @param activity The activity to send.
    * @param conversationId The ID of the conversation. Defaults to the current conversation ID.
-   * @returns A promise that resolves to an array of activities containing the responses.
+   * @returns An async generator yielding the Agent's Activities.
    */
-  public async sendActivity (activity: Activity, conversationId: string = this.conversationId) {
+  public async * sendActivity (activity: Activity, conversationId: string = this.conversationId) : AsyncGenerator<Activity> {
     const localConversationId = activity.conversation?.id ?? conversationId
     const uriExecute = getCopilotStudioConnectionUrl(this.settings, localConversationId)
     const qbody: ExecuteTurnRequest = new ExecuteTurnRequest(activity)
 
-    const config: AxiosRequestConfig = {
-      method: 'post',
-      url: uriExecute,
-      headers: {
-        Accept: 'text/event-stream',
-        'Content-Type': 'application/json',
-      },
-      data: qbody,
-      responseType: 'stream',
-      adapter: 'fetch'
-    }
     logger.info('Sending activity...', activity)
-    const values = await this.postRequestAsync(config)
-    logger.info(`Received ${values.length} activities.`, values)
-    return values
+    yield * this.postRequestAsync(uriExecute, qbody, 'POST')
   }
 }

packages/agents-copilotstudio-client/src/copilotStudioClient.ts

@@ -218,12 +218,12 @@ export class CopilotStudioWebChat {
 
       logger.debug('--> Connection established.')
       notifyTyping()
-      const activity = await client.startConversationAsync()
-      // Remove replyToId to avoid timeout issues with WebChat on first activity.
-      delete activity.replyToId
-      conversation = activity.conversation
-      sequence = 0
-      notifyActivity(activity)
+
+      for await (const activity of client.startConversationAsync()) {
+        delete activity.replyToId
+        conversation = activity.conversation
+        notifyActivity(activity)
+      }
     })
 
     const notifyActivity = (activity: Partial<Activity>) => {
@@ -232,9 +232,10 @@ export class CopilotStudioWebChat {
         timestamp: new Date().toISOString(),
         channelData: {
           ...activity.channelData,
-          'webchat:sequence-id': sequence++,
+          'webchat:sequence-id': sequence,
         },
       }
+      sequence++
       logger.debug(`Notify '${newActivity.type}' activity to WebChat:`, newActivity)
       activitySubscriber?.next(newActivity)
     }
@@ -276,15 +277,16 @@ export class CopilotStudioWebChat {
             notifyActivity(newActivity)
             notifyTyping()
 
-            const activities = await client.sendActivity(newActivity)
+            // Notify WebChat immediately that the message was sent
+            subscriber.next(newActivity.id!)
 
-            for (const responseActivity of activities) {
+            // Stream the agent's response, but don't block the UI
+            for await (const responseActivity of client.sendActivity(newActivity)) {
               notifyActivity(responseActivity)
+              logger.info('<-- Activity received correctly from Copilot Studio.')
             }
 
-            subscriber.next(newActivity.id!)
             subscriber.complete()
-            logger.info('<-- Activity received correctly from Copilot Studio.')
           } catch (error) {
             logger.error('Error sending Activity to Copilot Studio:', error)
             subscriber.error(error)