Streaming Chat Responses Specification¶

// StreamingChatClient extends ChatClient with streaming support.
// Implementations that support streaming implement this interface
// in addition to ChatClient.
type StreamingChatClient interface {
    ChatClient

    // StreamChat sends a message and streams the response via the callback.
    // The callback is invoked for each event in the stream. If the callback
    // returns an error, the stream is cancelled.
    // The final return value is the complete response text (concatenation of
    // all text deltas) or an error.
    StreamChat(ctx context.Context, prompt string, callback StreamCallback) (string, error)
}

// StreamCallback receives streaming events. Return a non-nil error to cancel the stream.
type StreamCallback func(event StreamEvent) error

// StreamEvent represents a single event in a streaming response.
type StreamEvent struct {
    // Type indicates the kind of event.
    Type StreamEventType

    // Delta contains the text fragment for TextDelta events.
    Delta string

    // ToolCall contains tool call information for ToolCallStart/ToolCallDelta events.
    ToolCall *StreamToolCall

    // Error contains error information for Error events.
    Error error
}

// StreamEventType identifies the kind of stream event.
type StreamEventType int

const (
    // EventTextDelta is a partial text response.
    EventTextDelta StreamEventType = iota
    // EventToolCallStart indicates a tool call is beginning.
    EventToolCallStart
    // EventToolCallDelta contains partial tool call arguments.
    EventToolCallDelta
    // EventToolCallEnd indicates a tool call is complete and will be executed.
    EventToolCallEnd
    // EventComplete indicates the stream has finished successfully.
    EventComplete
    // EventError indicates an error occurred during streaming.
    EventError
)

// StreamToolCall contains information about a streaming tool call.
type StreamToolCall struct {
    ID        string
    Name      string
    Arguments string // partial or complete JSON arguments
}

// Consumer usage:
client, err := chat.New(ctx, props, cfg)
if streamer, ok := client.(chat.StreamingChatClient); ok {
    result, err := streamer.StreamChat(ctx, "prompt", func(e chat.StreamEvent) error {
        if e.Type == chat.EventTextDelta {
            fmt.Print(e.Delta) // progressive output
        }
        return nil
    })
}

func (c *Claude) StreamChat(ctx context.Context, prompt string, callback StreamCallback) (string, error) {
    c.messages = append(c.messages, anthropic.NewUserMessage(anthropic.NewTextBlock(prompt)))

    params := anthropic.MessageNewParams{
        Model:     c.model,
        MaxTokens: c.maxTokens,
        Messages:  c.messages,
        System:    []anthropic.TextBlockParam{{Text: c.system}},
    }

    stream := c.client.Messages.NewStreaming(ctx, params)
    var fullText strings.Builder

    for stream.Next() {
        event := stream.Current()
        switch e := event.(type) {
        case *anthropic.ContentBlockDeltaEvent:
            if e.Delta.Type == "text_delta" {
                fullText.WriteString(e.Delta.Text)
                if err := callback(StreamEvent{Type: EventTextDelta, Delta: e.Delta.Text}); err != nil {
                    return fullText.String(), err
                }
            }
        case *anthropic.MessageStopEvent:
            callback(StreamEvent{Type: EventComplete})
        }
    }

    if err := stream.Err(); err != nil {
        return fullText.String(), errors.Wrap(err, "claude stream error")
    }

    return fullText.String(), nil
}

func (a *OpenAI) StreamChat(ctx context.Context, prompt string, callback StreamCallback) (string, error) {
    a.params.Messages = append(a.params.Messages, openai.UserMessage(prompt))
    a.params.StreamOptions = &openai.ChatCompletionStreamOptionsParam{IncludeUsage: true}

    stream := a.oai.Chat.Completions.NewStreaming(ctx, a.params)
    var fullText strings.Builder

    for stream.Next() {
        chunk := stream.Current()
        for _, choice := range chunk.Choices {
            if choice.Delta.Content != "" {
                fullText.WriteString(choice.Delta.Content)
                if err := callback(StreamEvent{Type: EventTextDelta, Delta: choice.Delta.Content}); err != nil {
                    return fullText.String(), err
                }
            }
        }
    }

    if err := stream.Err(); err != nil {
        return fullText.String(), errors.Wrap(err, "openai stream error")
    }

    callback(StreamEvent{Type: EventComplete})
    return fullText.String(), nil
}

func (g *Gemini) StreamChat(ctx context.Context, prompt string, callback StreamCallback) (string, error) {
    g.history = append(g.history, &genai.Content{
        Role:  "user",
        Parts: []*genai.Part{genai.NewPartFromText(prompt)},
    })

    var fullText strings.Builder
    for result, err := range g.client.Models.GenerateContentStream(ctx, g.model, g.history, g.genCfg) {
        if err != nil {
            return fullText.String(), errors.Wrap(err, "gemini stream error")
        }
        for _, candidate := range result.Candidates {
            for _, part := range candidate.Content.Parts {
                if part.Text != "" {
                    fullText.WriteString(part.Text)
                    if err := callback(StreamEvent{Type: EventTextDelta, Delta: part.Text}); err != nil {
                        return fullText.String(), err
                    }
                }
            }
        }
    }

    callback(StreamEvent{Type: EventComplete})
    return fullText.String(), nil
}

// Simplified streaming ReAct loop
for step := range maxSteps {
    // Stream response
    // If tool calls detected → execute tools → continue loop
    // If no tool calls → stream is complete
}

func newMockSSEServer(t *testing.T, events []string) *httptest.Server {
    t.Helper()
    return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "text/event-stream")
        flusher := w.(http.Flusher)
        for _, event := range events {
            fmt.Fprintf(w, "data: %s\n\n", event)
            flusher.Flush()
        }
    }))
}