fix: set index of tool calls in stream mode (#1468)

**Description**

There is an `index` field
(https://github.com/openai/openai-python/blob/4e8856576211064b09c0cc4a1ed35b82b169abe2/src/openai/types/chat/chat_completion_chunk.py#L48)
in the tool calls definition in stream mode: and it's not optional.
Third party libs like
opentelemetry(https://github.com/open-telemetry/opentelemetry-python-contrib/blob/447aac2b0fbe88998e970941c1160147f355ec73/instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/patch.py#L285)
is implemented based on this assumption. The index field should be set
according the number of tool calls in the streaming events which can be
tracked by checking message start/stop event types.

- aws bedrock: track the tool index by contentBlockStart/Stop events
- gcp anthropic: track the tool index by content_block_start/stop events

**Discuss/suggestion**: 

- `ChatCompletionMessageToolCallParam` should not be reused in the
response type.
- For both converse api and anthropic api, the stream chunks have strong
dependencies. We should use complete responses to write unit tests.

---------

Signed-off-by: yxia216 <[email protected]>
Signed-off-by: secustor <[email protected]>
Signed-off-by: ydu208 <[email protected]>
Signed-off-by: Takeshi Yoneda <[email protected]>
Signed-off-by: Dan Sun <[email protected]>
Co-authored-by: Sebastian Poxhofer <[email protected]>
Co-authored-by: yuhongd <[email protected]>
Co-authored-by: ydu208 <[email protected]>
Co-authored-by: Dan Sun <[email protected]>
Co-authored-by: Takeshi Yoneda <[email protected]>

Author: 6 people

internal/apischema/awsbedrock/awsbedrock.go

@@ -403,9 +403,44 @@ type TokenUsage struct {
 	CacheWriteInputTokens *int `json:"cacheWriteInputTokens,omitempty"`
 }
 
+// ConverseStreamEventType represents a distinct event type received from the Bedrock ConverseStream API.
+// Using a string type provides a clear, human-readable value, which is often helpful
+// when logging or debugging stream events.
+type ConverseStreamEventType string
+
+const (
+	// ConverseStreamEventTypeUnknown is the zero value and represents an uninitialized or unknown event type.
+	// This is a Go best practice to ensure the zero value is not a valid constant.
+	ConverseStreamEventTypeUnknown ConverseStreamEventType = ""
+
+	// ConverseStreamEventTypeMessageStart signals the start of the assistant's message.
+	ConverseStreamEventTypeMessageStart ConverseStreamEventType = "messageStart"
+
+	// ConverseStreamEventTypeContentBlockStart signals the start of a content block within the message (e.g., text or tool use).
+	ConverseStreamEventTypeContentBlockStart ConverseStreamEventType = "contentBlockStart"
+
+	// ConverseStreamEventTypeContentBlockDelta contains a chunk of content (e.g., text or tool input).
+	ConverseStreamEventTypeContentBlockDelta ConverseStreamEventType = "contentBlockDelta"
+
+	// ConverseStreamEventTypeContentBlockStop signals the end of a content block.
+	ConverseStreamEventTypeContentBlockStop ConverseStreamEventType = "contentBlockStop"
+
+	// ConverseStreamEventTypeMessageStop signals the end of the entire message.
+	ConverseStreamEventTypeMessageStop ConverseStreamEventType = "messageStop"
+
+	// ConverseStreamEventTypeMetadata contains usage and latency information.
+	ConverseStreamEventTypeMetadata ConverseStreamEventType = "metadata"
+)
+
+// String implements the fmt.Stringer interface for better logging/printing.
+func (c ConverseStreamEventType) String() string {
+	return string(c)
+}
+
 // ConverseStreamEvent is the union of all possible event types in the AWS Bedrock API:
 // https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ConverseStream.html
 type ConverseStreamEvent struct {
+	EventType         string                                `json:"eventType"`
 	ContentBlockIndex int                                   `json:"contentBlockIndex,omitempty"`
 	Delta             *ConverseStreamEventContentBlockDelta `json:"delta,omitempty"`
 	Role              *string                               `json:"role,omitempty"`

internal/apischema/openai/openai.go

@@ -546,8 +546,6 @@ type ChatCompletionMessageToolCallFunctionParam struct {
 }
 
 type ChatCompletionMessageToolCallParam struct {
-	// Add this Index field. It is required for streaming.
-	Index *int `json:"index,omitempty"`
 	// The ID of the tool call.
 	ID *string `json:"id"`
 	// The function that the model called.
@@ -1350,14 +1348,24 @@ type ChatCompletionResponseChunkChoice struct {
 	FinishReason ChatCompletionChoicesFinishReason       `json:"finish_reason,omitempty"`
 }
 
+type ChatCompletionChunkChoiceDeltaToolCall struct {
+	Index int64 `json:"index"`
+	// The ID of the tool call.
+	ID *string `json:"id"`
+	// The function that the model called.
+	Function ChatCompletionMessageToolCallFunctionParam `json:"function"`
+	// The type of the tool. Currently, only `function` is supported.
+	Type ChatCompletionMessageToolCallType `json:"type,omitempty"`
+}
+
 // ChatCompletionResponseChunkChoiceDelta is described in the OpenAI API documentation:
 // https://platform.openai.com/docs/api-reference/chat/streaming#chat/streaming-choices
 type ChatCompletionResponseChunkChoiceDelta struct {
-	Content          *string                              `json:"content,omitempty"`
-	Role             string                               `json:"role,omitempty"`
-	ToolCalls        []ChatCompletionMessageToolCallParam `json:"tool_calls,omitempty"`
-	Annotations      *[]Annotation                        `json:"annotations,omitempty"`
-	ReasoningContent *AWSBedrockStreamReasoningContent    `json:"reasoning_content,omitempty"`
+	Content          *string                                  `json:"content,omitempty"`
+	Role             string                                   `json:"role,omitempty"`
+	ToolCalls        []ChatCompletionChunkChoiceDeltaToolCall `json:"tool_calls,omitempty"`
+	Annotations      *[]Annotation                            `json:"annotations,omitempty"`
+	ReasoningContent *AWSBedrockStreamReasoningContent        `json:"reasoning_content,omitempty"`
 }
 
 // Error is described in the OpenAI API documentation

internal/apischema/openai/openai_test.go

@@ -1288,19 +1288,20 @@ func TestChatCompletionResponseChunkChoice(t *testing.T) {
 				Index: 0,
 				Delta: &ChatCompletionResponseChunkChoiceDelta{
 					Role: "assistant",
-					ToolCalls: []ChatCompletionMessageToolCallParam{
+					ToolCalls: []ChatCompletionChunkChoiceDeltaToolCall{
 						{
 							ID:   ptr.To("tooluse_QklrEHKjRu6Oc4BQUfy7ZQ"),
 							Type: "function",
 							Function: ChatCompletionMessageToolCallFunctionParam{
 								Name:      "cosine",
 								Arguments: "",
 							},
+							Index: 0,
 						},
 					},
 				},
 			},
-			expected: `{"index":0,"delta":{"role":"assistant","tool_calls":[{"id":"tooluse_QklrEHKjRu6Oc4BQUfy7ZQ","function":{"arguments":"","name":"cosine"},"type":"function"}]}}`,
+			expected: `{"index":0,"delta":{"role":"assistant","tool_calls":[{"id":"tooluse_QklrEHKjRu6Oc4BQUfy7ZQ","function":{"arguments":"","name":"cosine"},"type":"function", "index": 0}]}}`,
 		},
 		{
 			name: "streaming chunk with annotations",

internal/extproc/translator/gemini_helper.go

@@ -647,6 +647,48 @@ func extractToolCallsFromGeminiParts(parts []*genai.Part) ([]openai.ChatCompleti
 	return toolCalls, nil
 }
 
+// extractToolCallsFromGeminiPartsStream extracts tool calls from Gemini parts for streaming responses.
+// Each tool call is assigned an incremental index starting from 0, matching OpenAI's streaming protocol.
+// Returns ChatCompletionChunkChoiceDeltaToolCall types suitable for streaming responses, or nil if no tool calls are found.
+func extractToolCallsFromGeminiPartsStream(parts []*genai.Part) ([]openai.ChatCompletionChunkChoiceDeltaToolCall, error) {
+	var toolCalls []openai.ChatCompletionChunkChoiceDeltaToolCall
+	toolCallIndex := int64(0)
+
+	for _, part := range parts {
+		if part == nil || part.FunctionCall == nil {
+			continue
+		}
+
+		// Convert function call arguments to JSON string.
+		args, err := json.Marshal(part.FunctionCall.Args)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal function arguments: %w", err)
+		}
+
+		// Generate a random ID for the tool call.
+		toolCallID := uuid.New().String()
+
+		toolCall := openai.ChatCompletionChunkChoiceDeltaToolCall{
+			ID:   &toolCallID,
+			Type: openai.ChatCompletionMessageToolCallTypeFunction,
+			Function: openai.ChatCompletionMessageToolCallFunctionParam{
+				Name:      part.FunctionCall.Name,
+				Arguments: string(args),
+			},
+			Index: toolCallIndex,
+		}
+
+		toolCalls = append(toolCalls, toolCall)
+		toolCallIndex++
+	}
+
+	if len(toolCalls) == 0 {
+		return nil, nil
+	}
+
+	return toolCalls, nil
+}
+
 // geminiUsageToOpenAIUsage converts Gemini usage metadata to OpenAI usage.
 func geminiUsageToOpenAIUsage(metadata *genai.GenerateContentResponseUsageMetadata) openai.Usage {
 	if metadata == nil {
@@ -746,7 +788,7 @@ func geminiCandidatesToOpenAIStreamingChoices(candidates []*genai.Candidate, res
 			}
 
 			// Extract tool calls if any.
-			toolCalls, err := extractToolCallsFromGeminiParts(candidate.Content.Parts)
+			toolCalls, err := extractToolCallsFromGeminiPartsStream(candidate.Content.Parts)
 			if err != nil {
 				return nil, fmt.Errorf("error extracting tool calls: %w", err)
 			}