梦兽编程™AI_SUITE A streaming backend is only useful when users can see the output as it arrives. This tutorial covers the end-to-end chat interface: server event production, browser consumption, incremental rendering, retries, and the SSE versus WebSocket decision.
← Multimodal | Agent Runtime Architecture →
SSE Versus WebSocket
Both protocols keep a live connection open. They optimize for different interaction patterns.
| Concern | SSE | WebSocket |
|---|---|---|
| Direction | Server to client | Full duplex |
| Browser support | Native | Native |
| HTTP compatibility | High | Separate framing |
| Proxy support | Usually easier | Often needs explicit upgrade support |
| Best use case | Chat output and notifications | Voice, collaboration, and fast control loops |
| Reconnect model | Automatic with EventSource | Usually custom |
For most ADK Go chat applications, SSE is enough. Use WebSocket when the frontend also sends frequent real-time control commands, such as live microphone frames, cursor sharing, or tool approval interactions.
Backend SSE Endpoint
The Go backend should set streaming headers, write events in the SSE format, and flush after each event.
func chatHandler(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var req ChatRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming not supported", http.StatusInternalServerError)
return
}
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
stream, err := agent.RunStream(ctx, req.Message)
if err != nil {
writeSSE(w, "error", map[string]string{"error": err.Error()})
return
}
defer stream.Close()
seq := int64(0)
for {
event, err := stream.Recv()
if err == io.EOF {
writeSSE(w, "done", map[string]any{"sequence": seq, "status": "completed"})
flusher.Flush()
return
}
if err != nil {
writeSSE(w, "error", map[string]string{"error": err.Error()})
flusher.Flush()
return
}
seq++
if err := writeStreamEvent(w, seq, event); err != nil {
return
}
flusher.Flush()
}
}
The sequence number is the contract between frontend and backend. It matters more than perfect latency because the UI can recover from slow networks, duplicate events, and temporary failures.
Writing SSE Events
SSE formatting is simple but must be exact. Each event has optional id, event, and one or more data lines, followed by a blank line.
func writeSSE(w http.ResponseWriter, eventName string, payload any) error {
data, err := json.Marshal(payload)
if err != nil {
return err
}
if eventName != "" {
fmt.Fprintf(w, "event: %s\n", eventName)
}
fmt.Fprintf(w, "data: %s\n\n", data)
return nil
}
A common production bug is forgetting the final blank line. Without it, some browsers wait for more data before dispatching the event.
Frontend EventSource Client
The browser EventSource API is the simplest path to streaming chat. It handles reconnection automatically, but you still need to manage duplicate delivery and UI state.
function streamChat(message) {
const source = new EventSource(`/api/chat?message=${encodeURIComponent(message)}`);
const messageEl = document.createElement('p');
messageEl.className = 'assistant-message';
document.querySelector('#chat').appendChild(messageEl);
source.addEventListener('text_delta', (evt) => {
const data = JSON.parse(evt.data);
messageEl.textContent += data.text;
scrollToBottom();
});
source.addEventListener('tool_call', (evt) => {
const data = JSON.parse(evt.data);
showToolState(data.tool, 'running');
});
source.addEventListener('done', (evt) => {
const data = JSON.parse(evt.data);
source.close();
saveMessage(data);
});
source.addEventListener('error', (evt) => {
source.close();
showConnectionError(evt);
});
}
The frontend should render incrementally and treat done as the only signal that the response is complete. Partial text events are not necessarily final.
Ordered Rendering
A robust UI uses sequence numbers rather than raw event arrival order.
let lastSeenSequence = 0;
function applyEvent(name, data) {
if (data.sequence <= lastSeenSequence) return;
lastSeenSequence = data.sequence;
switch (name) {
case 'text_delta':
appendText(data.text);
break;
case 'tool_call':
appendToolCall(data.tool, data.arguments);
break;
case 'error':
showError(data.error);
break;
case 'done':
markComplete();
break;
}
}
Deduplication prevents the same token from appearing twice after EventSource reconnects. Some backends include an id header or event field for this purpose; if yours does not, use the sequence number from data.
Connection Management
Every open stream uses server memory and model-call resources. Limit concurrent streams per user or API key, add a per-stream timeout separate from HTTP server timeouts, cancel the agent stream when the HTTP request context ends, keep a Redis-backed session store for reconnect or resume scenarios, emit heartbeat events only if your load balancer needs them, and close idle connections after a fixed period.
type StreamManager struct {
active sync.Map
limiter *rate.Limiter
}
func (m *StreamManager) Start(ctx context.Context, streamID string, run func()) error {
if !m.limiter.Allow() {
return fmt.Errorf("too many active streams")
}
m.active.Store(streamID, ctx.Done())
go func() {
run()
m.active.Delete(streamID)
}()
return nil
}
Resuming and Replay
If a user disconnects mid-response, the product can restart from scratch or resume from the last sequence. Resume is more complex but feels much smoother. A minimal resume contract is that the backend emits monotonically increasing sequence values, the frontend sends last_sequence on reconnect, and the backend re-emits missing events or continues generation where possible.
func resumeHandler(w http.ResponseWriter, r *http.Request) {
lastSeq, _ := strconv.ParseInt(r.URL.Query().Get("last_sequence"), 10, 64)
stream, err := session.LoadStream(r.Context(), lastSeq)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
defer stream.Close()
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("X-Accel-Buffering", "no")
flusher, _ := w.(http.Flusher)
for event := range stream.Events() {
writeSSE(w, event.Kind, event.Payload)
flusher.Flush()
}
}
Resume is especially useful for voice, multimodal, and long-analysis workflows where restarting wastes time and tokens.
Frontend Rendering Optimization
A chat UI can feel slow even when the server is fast. Optimize rendering the same way you optimize streaming: render text deltas into a single message node, avoid full-page re-renders on every event, lazy-render tool-call panels, throttle scroll updates, cache completed assistant messages locally, and show a clear disconnected or retrying state.
For large code blocks or structured outputs, render a skeleton first and then hydrate details when the done event arrives.
Production Deployment Notes
When deploying a streaming chat UI, verify that load balancers and reverse proxies do not buffer SSE responses, TLS keep-alive is compatible with long-lived connections, the backend has a circuit breaker for failed model calls, logs include session_id, stream_id, and sequence, metrics track first-token latency and reconnects, and the frontend shows graceful errors for canceled or timed-out streams.
Summary
Module 6 is complete. You now understand streaming principles and the event model, event handling with filtering and recovery, multimodal streaming for audio and images, and chat UI implementation with SSE or WebSocket.
← Multimodal Streaming | Agent Runtime Architecture →
Want to keep learning hands-on Go ADK? Follow the “Mengshou Programming” channel for weekly Go and AI programming posts.
