11package client
22
33import (
4+ "bufio"
45 "context"
6+ "encoding/json"
57 "fmt"
8+ "io"
9+ "log/slog"
10+ "net/http"
11+ "strings"
612
713 "github.com/harness/harness-mcp/client/dto"
814)
@@ -15,28 +21,153 @@ type GenaiService struct {
1521 Client * Client
1622}
1723
18- func (g * GenaiService ) SendAIDevOpsChat (ctx context.Context , scope dto.Scope , request * dto.ServiceChatParameters ) (* dto.ServiceChatResponse , error ) {
19- path := aiDevopsChatPath
20- params := make (map [string ]string )
24+ // SendAIDevOpsChat sends a request to the AI DevOps service and returns the response.
25+ // If request.Stream is true and onProgress is provided, it will handle streaming responses.
26+ // For non-streaming requests or when onProgress is nil, it will use the standard request flow.
27+ func (g * GenaiService ) SendAIDevOpsChat (ctx context.Context , scope dto.Scope , request * dto.ServiceChatParameters , onProgress ... func (progressUpdate dto.ProgressUpdate ) error ) (* dto.ServiceChatResponse , error ) {
28+ if g == nil || g .Client == nil {
29+ return nil , fmt .Errorf ("genai service or client is nil" )
30+ }
31+
32+ if request == nil {
33+ return nil , fmt .Errorf ("request is nil" )
34+ }
35+
36+ // Validate context
37+ if ctx == nil {
38+ return nil , fmt .Errorf ("context is nil" )
39+ }
40+
41+ // Extract the progress callback if provided
42+ var progressCB func (progressUpdate dto.ProgressUpdate ) error
43+ if len (onProgress ) > 0 && onProgress [0 ] != nil {
44+ progressCB = onProgress [0 ]
45+ }
2146
47+ params := make (map [string ]string )
2248 // Only add non-empty scope parameters
2349 if scope .AccountID != "" {
2450 params ["accountIdentifier" ] = scope .AccountID
2551 }
26-
2752 if scope .OrgID != "" {
2853 params ["orgIdentifier" ] = scope .OrgID
2954 }
30-
3155 if scope .ProjectID != "" {
3256 params ["projectIdentifier" ] = scope .ProjectID
3357 }
3458
35- var response dto.ServiceChatResponse
36- err := g .Client .Post (ctx , path , params , request , & response )
59+ // Handle non-streaming case with early return
60+ isStreaming := request .Stream && progressCB != nil
61+ if ! isStreaming {
62+ var response dto.ServiceChatResponse
63+ err := g .Client .Post (ctx , aiDevopsChatPath , params , request , & response )
64+ if err != nil {
65+ return nil , fmt .Errorf ("failed to send request to genai service: %w" , err )
66+ }
67+
68+ return & response , nil
69+ }
70+
71+ // Execute the streaming request
72+ resp , err := g .Client .PostStream (ctx , aiDevopsChatPath , params , request )
73+ if err != nil {
74+ slog .Warn ("Failed to execute streaming request" , "error" , err .Error ())
75+ return nil , fmt .Errorf ("failed to execute streaming request: %w" , err )
76+ }
77+ defer resp .Body .Close ()
78+
79+ // Check response status
80+ if resp .StatusCode != http .StatusOK {
81+ body , _ := io .ReadAll (resp .Body )
82+ return nil , fmt .Errorf ("streaming request failed with status %d: %s" , resp .StatusCode , string (body ))
83+ }
84+
85+ // Initialize the response with conversation ID from request
86+ finalResponse := & dto.ServiceChatResponse {
87+ ConversationID : request .ConversationID ,
88+ }
89+
90+ // Process the streaming response
91+ err = g .processStreamingResponse (resp .Body , finalResponse , progressCB )
3792 if err != nil {
38- return nil , fmt .Errorf ("failed to send request to genai service: %w" , err )
93+ slog .Warn ("Error processing streaming response" , "error" , err .Error ())
94+ return finalResponse , fmt .Errorf ("error processing streaming response: %w" , err )
95+ }
96+
97+ return finalResponse , nil
98+ }
99+
100+ // processStreamingResponse handles Server-Sent Events (SSE) streaming responses
101+ // and accumulates complete events before forwarding them with appropriate event types
102+ func (g * GenaiService ) processStreamingResponse (body io.ReadCloser , finalResponse * dto.ServiceChatResponse , onProgress func (dto.ProgressUpdate ) error ) error {
103+ scanner := bufio .NewScanner (body )
104+ var allContent , currentEvent strings.Builder
105+ inEvent := false
106+ var eventType string
107+ var eventData string
108+
109+ for scanner .Scan () {
110+ line := scanner .Text ()
111+ allContent .WriteString (line + "\n " )
112+
113+ switch {
114+ case strings .HasPrefix (line , "event: " ):
115+ eventType = strings .TrimPrefix (line , "event: " )
116+ currentEvent .Reset ()
117+ currentEvent .WriteString (line + "\n " )
118+ inEvent = true
119+
120+ case strings .HasPrefix (line , "data: " ) && inEvent :
121+ eventData = strings .TrimPrefix (line , "data: " )
122+ currentEvent .WriteString (line + "\n " )
123+
124+ case line == "" && inEvent :
125+ // Empty line ends the event
126+ currentEvent .WriteString ("\n " )
127+
128+ if onProgress != nil {
129+ eventPayload := map [string ]any {
130+ "type" : eventType ,
131+ "data" : eventData ,
132+ }
133+
134+ // Convert to JSON string
135+ jsonPayload , err := json .Marshal (eventPayload )
136+ if err != nil {
137+ slog .Warn ("Error creating JSON payload" , "error" , err .Error ())
138+ } else {
139+ progress := dto.ProgressUpdate {
140+ Message : string (jsonPayload ),
141+ }
142+
143+ if err := onProgress (progress ); err != nil {
144+ slog .Warn ("Error forwarding event" , "type" , eventType , "error" , err .Error ())
145+ }
146+ }
147+
148+ if eventType == "error" {
149+ finalResponse .Error = eventData
150+ }
151+ }
152+
153+ // Reset for next event
154+ inEvent = false
155+ }
156+ }
157+
158+ // Add a header note to inform the Uber Agent that these events have already been shown to the user
159+ // TODO: move this to a prompt template for uber agent to ingest
160+ instructionNote := "NOTE TO AGENT: The SSE events below have already been streamed to the end user in real-time.\n " +
161+ "Do not repeat the full content of these events in your response.\n " +
162+ "Focus on summarizing key outcomes and providing additional context or next steps.\n " +
163+ "---\n \n "
164+
165+ finalResponse .Response = instructionNote + allContent .String ()
166+
167+ if err := scanner .Err (); err != nil {
168+ slog .Warn ("Error in scanner" , "error" , err .Error ())
169+ return fmt .Errorf ("error reading response stream: %w" , err )
39170 }
40171
41- return & response , nil
172+ return nil
42173}
0 commit comments