From 650a002a5a2df58f4844f314ab2b17d718d7b896 Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 11 Feb 2026 12:05:07 +0800 Subject: [PATCH] Implement support for Anthropic connectors in the LLM and sandbox components - Add handling for Anthropic connector types in the sandbox, allowing direct connections without a proxy. - Enhance capability retrieval to support both OpenAI and Anthropic formats, ensuring a unified interface. - Update the executor and command logic to differentiate between OpenAI and Anthropic configurations, streamlining environment setup. - Modify the provider selection logic to accommodate Anthropic capabilities, improving flexibility in LLM provider management. - Refactor API detection to include Anthropic, ensuring accurate identification of connector types. --- agent/assistant/sandbox.go | 9 + agent/llm/capabilities.go | 27 + agent/llm/providers/anthropic/anthropic.go | 1173 ++++++++++++++++++++ agent/llm/providers/anthropic/types.go | 154 +++ agent/llm/providers/factory.go | 30 +- agent/sandbox/claude/command.go | 24 +- agent/sandbox/claude/executor.go | 7 + agent/sandbox/executor.go | 1 + agent/sandbox/types.go | 4 + openapi/llm/llm.go | 13 +- 10 files changed, 1424 insertions(+), 18 deletions(-) create mode 100644 agent/llm/providers/anthropic/anthropic.go create mode 100644 agent/llm/providers/anthropic/types.go diff --git a/agent/assistant/sandbox.go b/agent/assistant/sandbox.go index 60750589..0f1675bd 100644 --- a/agent/assistant/sandbox.go +++ b/agent/assistant/sandbox.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/yaoapp/gou/connector" gouMCP "github.com/yaoapp/gou/mcp" mcpProcess "github.com/yaoapp/gou/mcp/process" "github.com/yaoapp/yao/agent/context" @@ -257,6 +258,14 @@ func (ast *Assistant) buildSandboxOptions(ctx *context.Context, opts *context.Op return nil, fmt.Errorf("failed to get connector: %w", err) } + // Determine connector type for sandbox proxy behavior + // Anthropic connectors bypass the proxy (Claude CLI connects directly) + if conn.Is(connector.ANTHROPIC) { + execOpts.ConnectorType = "anthropic" + } else { + execOpts.ConnectorType = "openai" + } + setting := conn.Setting() if host, ok := setting["host"].(string); ok { execOpts.ConnectorHost = host diff --git a/agent/llm/capabilities.go b/agent/llm/capabilities.go index 63072805..85482f5c 100644 --- a/agent/llm/capabilities.go +++ b/agent/llm/capabilities.go @@ -2,6 +2,7 @@ package llm import ( "github.com/yaoapp/gou/connector" + "github.com/yaoapp/gou/connector/anthropic" "github.com/yaoapp/gou/connector/openai" ) @@ -68,6 +69,14 @@ func GetCapabilitiesFromConn(conn connector.Connector, modelCapabilities map[str if capabilities, ok := caps.(openai.Capabilities); ok { return &capabilities } + // Try to convert from *anthropic.Capabilities + if capabilities, ok := caps.(*anthropic.Capabilities); ok { + return convertAnthropicCaps(capabilities) + } + // Try to convert from anthropic.Capabilities (value type) + if capabilities, ok := caps.(anthropic.Capabilities); ok { + return convertAnthropicCaps(&capabilities) + } } } @@ -125,3 +134,21 @@ func ToMap(caps *openai.Capabilities) map[string]interface{} { return result } + +// convertAnthropicCaps converts anthropic.Capabilities to openai.Capabilities +// This provides a unified capabilities interface across connector types +func convertAnthropicCaps(caps *anthropic.Capabilities) *openai.Capabilities { + if caps == nil { + return getDefaultCapabilities() + } + return &openai.Capabilities{ + Vision: caps.Vision, + Audio: caps.Audio, + ToolCalls: caps.ToolCalls, + Reasoning: caps.Reasoning, + Streaming: caps.Streaming, + JSON: caps.JSON, + Multimodal: caps.Multimodal, + TemperatureAdjustable: caps.TemperatureAdjustable, + } +} diff --git a/agent/llm/providers/anthropic/anthropic.go b/agent/llm/providers/anthropic/anthropic.go new file mode 100644 index 00000000..4639bc22 --- /dev/null +++ b/agent/llm/providers/anthropic/anthropic.go @@ -0,0 +1,1173 @@ +package anthropic + +import ( + gocontext "context" + "fmt" + "strings" + "time" + + jsoniter "github.com/json-iterator/go" + "github.com/yaoapp/gou/connector" + gouAnthropicConn "github.com/yaoapp/gou/connector/anthropic" + gouOpenAI "github.com/yaoapp/gou/connector/openai" + "github.com/yaoapp/gou/http" + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/i18n" + "github.com/yaoapp/yao/agent/llm/adapters" + "github.com/yaoapp/yao/agent/llm/providers/base" + "github.com/yaoapp/yao/agent/output/message" +) + +// Provider Anthropic Messages API provider +type Provider struct { + *base.Provider + adapters []adapters.CapabilityAdapter +} + +// New create a new Anthropic provider +func New(conn connector.Connector, capabilities *gouOpenAI.Capabilities) *Provider { + return &Provider{ + Provider: base.NewProvider(conn, capabilities), + adapters: buildAdapters(capabilities), + } +} + +// NewFromAnthropicCaps create a new Anthropic provider from Anthropic capabilities +func NewFromAnthropicCaps(conn connector.Connector, caps *gouAnthropicConn.Capabilities) *Provider { + // Convert anthropic capabilities to openai capabilities for base provider compatibility + openaiCaps := &gouOpenAI.Capabilities{ + Vision: caps.Vision, + Audio: caps.Audio, + ToolCalls: caps.ToolCalls, + Reasoning: caps.Reasoning, + Streaming: caps.Streaming, + JSON: caps.JSON, + Multimodal: caps.Multimodal, + TemperatureAdjustable: caps.TemperatureAdjustable, + } + return &Provider{ + Provider: base.NewProvider(conn, openaiCaps), + adapters: buildAdapters(openaiCaps), + } +} + +// buildAdapters builds capability adapters based on model capabilities +func buildAdapters(cap *gouOpenAI.Capabilities) []adapters.CapabilityAdapter { + if cap == nil { + return []adapters.CapabilityAdapter{} + } + + result := make([]adapters.CapabilityAdapter, 0) + + // Tool call adapter + result = append(result, adapters.NewToolCallAdapter(cap.ToolCalls)) + + // Vision adapter + visionSupport, visionFormat := context.GetVisionSupport(cap) + if visionSupport { + result = append(result, adapters.NewVisionAdapter(true, visionFormat)) + } else if cap.Vision != nil { + result = append(result, adapters.NewVisionAdapter(false, context.VisionFormatNone)) + } + + // Audio adapter + result = append(result, adapters.NewAudioAdapter(cap.Audio)) + + // Reasoning adapter + if cap.Reasoning { + result = append(result, adapters.NewReasoningAdapter(adapters.ReasoningFormatOpenAI, cap)) + } else { + result = append(result, adapters.NewReasoningAdapter(adapters.ReasoningFormatNone, cap)) + } + + return result +} + +// Stream stream completion from Anthropic API +func (p *Provider) Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error) { + trace, _ := ctx.Trace() + if trace != nil { + trace.Debug("Anthropic Stream: Starting stream request", map[string]any{ + "message_count": len(messages), + }) + } + + maxRetries := 3 + var lastErr error + + goCtx := ctx.Context + if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Context != nil { + goCtx = ctx.Stack.Options.Context + } + if goCtx == nil { + goCtx = gocontext.Background() + } + + currentMessages := make([]context.Message, len(messages)) + copy(currentMessages, messages) + + for attempt := 0; attempt < maxRetries; attempt++ { + select { + case <-goCtx.Done(): + return nil, fmt.Errorf("context cancelled: %w", goCtx.Err()) + default: + } + + if ctx.Interrupt != nil { + if signal := ctx.Interrupt.Peek(); signal != nil && signal.Type == context.InterruptForce { + return nil, fmt.Errorf("force interrupted by user") + } + } + + if attempt > 0 { + backoff := time.Duration(1< + // data: + var currentEventType string + + streamHandler := func(data []byte) int { + select { + case <-goCtx.Done(): + return http.HandlerReturnBreak + default: + } + + if ctx.Interrupt != nil { + if signal := ctx.Interrupt.Peek(); signal != nil && signal.Type == context.InterruptForce { + return http.HandlerReturnBreak + } + } + + if len(data) == 0 { + return http.HandlerReturnOk + } + + dataStr := string(data) + trimmed := strings.TrimSpace(dataStr) + + if trimmed == "" { + return http.HandlerReturnOk + } + + // Parse event type line + // Support both "event: type" (with space) and "event:type" (without space) formats + if strings.HasPrefix(trimmed, "event:") { + currentEventType = strings.TrimSpace(strings.TrimPrefix(trimmed, "event:")) + return http.HandlerReturnOk + } + + // Parse data line + // Support both "data: {...}" (with space) and "data:{...}" (without space) formats + if !strings.HasPrefix(trimmed, "data:") { + // Check for error response + if strings.HasPrefix(trimmed, "{") && strings.Contains(trimmed, `"error"`) { + var apiErr APIError + if err := jsoniter.UnmarshalFromString(trimmed, &apiErr); err == nil && apiErr.Error.Message != "" { + if handler != nil { + handler(message.ChunkError, []byte(apiErr.Error.Message)) + } + } + } + return http.HandlerReturnOk + } + + jsonStr := strings.TrimSpace(strings.TrimPrefix(trimmed, "data:")) + + if jsonStr == "" { + return http.HandlerReturnOk + } + + // Process based on event type + switch currentEventType { + case "message_start": + var event MessageStartEvent + if err := jsoniter.UnmarshalFromString(jsonStr, &event); err == nil { + accumulator.id = event.Message.ID + accumulator.model = event.Message.Model + accumulator.role = event.Message.Role + if event.Message.Usage != nil { + accumulator.usage = &message.UsageInfo{ + PromptTokens: event.Message.Usage.InputTokens, + TotalTokens: event.Message.Usage.InputTokens, + } + } + } + + case "content_block_start": + var event ContentBlockStartEvent + if err := jsoniter.UnmarshalFromString(jsonStr, &event); err == nil { + accumulator.currentBlockIndex = event.Index + accumulator.currentBlockType = event.ContentBlock.Type + + switch event.ContentBlock.Type { + case "thinking": + startMessage(msgTracker, message.ChunkThinking, handler) + case "text": + startMessage(msgTracker, message.ChunkText, handler) + case "tool_use": + accumulator.toolCalls[event.Index] = &accumulatedToolCall{ + id: event.ContentBlock.ID, + name: event.ContentBlock.Name, + } + toolCallInfo := &message.EventToolCallInfo{ + ID: event.ContentBlock.ID, + Name: event.ContentBlock.Name, + Index: event.Index, + } + startToolCallMessage(msgTracker, toolCallInfo, handler) + } + } + + case "content_block_delta": + var event ContentBlockDeltaEvent + if err := jsoniter.UnmarshalFromString(jsonStr, &event); err == nil { + switch event.Delta.Type { + case "thinking_delta": + if event.Delta.Thinking != "" { + accumulator.thinkingContent += event.Delta.Thinking + if handler != nil { + handler(message.ChunkThinking, []byte(event.Delta.Thinking)) + incrementChunk(msgTracker) + } + } + + case "text_delta": + if event.Delta.Text != "" { + accumulator.content += event.Delta.Text + if handler != nil { + handler(message.ChunkText, []byte(event.Delta.Text)) + incrementChunk(msgTracker) + } + } + + case "input_json_delta": + if event.Delta.PartialJSON != "" { + if tc, exists := accumulator.toolCalls[event.Index]; exists { + tc.inputJSON += event.Delta.PartialJSON + // Update tracker + if msgTracker.active && msgTracker.toolCallInfo != nil { + msgTracker.toolCallInfo.Arguments = tc.inputJSON + } + } + if handler != nil { + // Send tool call delta + toolCallData, _ := jsoniter.Marshal([]map[string]interface{}{ + { + "index": event.Index, + "function": map[string]interface{}{ + "arguments": event.Delta.PartialJSON, + }, + }, + }) + handler(message.ChunkToolCall, toolCallData) + incrementChunk(msgTracker) + } + } + + case "signature_delta": + // Handle thinking signature delta (for extended thinking) + // The signature is accumulated but not sent to handler + var sigDelta struct { + Type string `json:"type"` + Signature string `json:"signature"` + } + if err := jsoniter.UnmarshalFromString(jsonStr, &struct { + Delta *struct { + Signature string `json:"signature"` + } `json:"delta"` + }{Delta: &struct { + Signature string `json:"signature"` + }{}}); err == nil { + _ = sigDelta // signature tracking if needed + } + } + } + + case "content_block_stop": + endMessage(msgTracker, handler) + + case "message_delta": + var event MessageDeltaEvent + if err := jsoniter.UnmarshalFromString(jsonStr, &event); err == nil { + accumulator.stopReason = event.Delta.StopReason + if event.Usage != nil { + if accumulator.usage == nil { + accumulator.usage = &message.UsageInfo{} + } + accumulator.usage.CompletionTokens = event.Usage.OutputTokens + accumulator.usage.TotalTokens = accumulator.usage.PromptTokens + event.Usage.OutputTokens + } + } + + case "message_stop": + // Message complete + endMessage(msgTracker, handler) + + case "ping": + // Keep-alive, ignore + + case "error": + var apiErr struct { + Type string `json:"type"` + Error struct { + Type string `json:"type"` + Message string `json:"message"` + } `json:"error"` + } + if err := jsoniter.UnmarshalFromString(jsonStr, &apiErr); err == nil && apiErr.Error.Message != "" { + if handler != nil { + handler(message.ChunkError, []byte(apiErr.Error.Message)) + } + } + } + + return http.HandlerReturnOk + } + + // Log request + if trace != nil { + if requestBodyJSON, marshalErr := jsoniter.Marshal(requestBody); marshalErr == nil { + trace.Debug("Anthropic Stream Request", map[string]any{ + "url": url, + "body": string(requestBodyJSON), + }) + } + } + + // Error buffer for non-SSE error responses + var errorBuffer strings.Builder + errorDetected := false + + wrappedHandler := func(data []byte) int { + dataStr := string(data) + trimmed := strings.TrimSpace(dataStr) + + if trimmed == "" { + return http.HandlerReturnOk + } + + // SSE event/data lines - pass to stream handler + // Support both "event: type" (with space) and "event:type" (without space) formats + if strings.HasPrefix(trimmed, "event:") || strings.HasPrefix(trimmed, "data:") { + return streamHandler(data) + } + + // Detect JSON error response + if strings.HasPrefix(trimmed, "{") && strings.Contains(dataStr, `"error"`) { + errorDetected = true + } + + if errorDetected { + errorBuffer.Write(data) + errorBuffer.WriteString("\n") + return http.HandlerReturnOk + } + + return streamHandler(data) + } + + // Make streaming request + log.Trace("[LLM] Starting Anthropic Stream request: url=%s", url) + err = req.Stream(goCtx, "POST", requestBody, wrappedHandler) + _ = streamStartTime + + // Check for captured error response + if errorDetected && errorBuffer.Len() > 0 { + errorJSON := errorBuffer.String() + if trace != nil { + trace.Error(i18n.T(ctx.Locale, "llm.anthropic.stream.api_error"), map[string]any{"response": errorJSON}) + } + + var apiErr APIError + if parseErr := jsoniter.UnmarshalFromString(errorJSON, &apiErr); parseErr == nil && apiErr.Error.Message != "" { + err = fmt.Errorf("Anthropic API error: %s (type: %s)", apiErr.Error.Message, apiErr.Error.Type) + } else { + err = fmt.Errorf("Anthropic API error: %s", strings.TrimSpace(errorJSON)) + } + } + + // Handle context cancellation + if err != nil && goCtx.Err() != nil { + return nil, fmt.Errorf("stream cancelled: %w", goCtx.Err()) + } + + if err != nil { + endMessage(msgTracker, handler) + if handler != nil { + handler(message.ChunkError, []byte(err.Error())) + } + return nil, fmt.Errorf("streaming request failed: %w", err) + } + + // Check for empty response + if accumulator.id == "" { + endMessage(msgTracker, handler) + errMsg := fmt.Errorf("no data received from Anthropic API") + if handler != nil { + handler(message.ChunkError, []byte(errMsg.Error())) + } + return nil, errMsg + } + + // Build final response (convert to unified CompletionResponse) + response := &context.CompletionResponse{ + ID: accumulator.id, + Object: "message", + Model: accumulator.model, + Role: accumulator.role, + Content: accumulator.content, + ReasoningContent: accumulator.thinkingContent, + FinishReason: mapStopReason(accumulator.stopReason), + Usage: accumulator.usage, + } + + // Convert accumulated tool calls + if len(accumulator.toolCalls) > 0 { + toolCalls := make([]context.ToolCall, 0, len(accumulator.toolCalls)) + for i := 0; i < len(accumulator.toolCalls); i++ { + if tc, exists := accumulator.toolCalls[i]; exists { + toolCalls = append(toolCalls, context.ToolCall{ + ID: tc.id, + Type: "function", + Function: context.Function{ + Name: tc.name, + Arguments: tc.inputJSON, + }, + }) + } + } + response.ToolCalls = toolCalls + } + + endMessage(msgTracker, handler) + return response, nil +} + +// Post non-streaming completion request to Anthropic API +func (p *Provider) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error) { + trace, _ := ctx.Trace() + + maxRetries := 3 + var lastErr error + + goCtx := ctx.Context + if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Context != nil { + goCtx = ctx.Stack.Options.Context + } + if goCtx == nil { + goCtx = gocontext.Background() + } + + currentMessages := make([]context.Message, len(messages)) + copy(currentMessages, messages) + + for attempt := 0; attempt < maxRetries; attempt++ { + select { + case <-goCtx.Done(): + return nil, fmt.Errorf("context cancelled: %w", goCtx.Err()) + default: + } + + if attempt > 0 { + backoff := time.Duration(1< 0 { + contentBlocks := make([]map[string]interface{}, 0) + + // Add text content if present + if contentStr, ok := msg.Content.(string); ok && contentStr != "" { + contentBlocks = append(contentBlocks, map[string]interface{}{ + "type": "text", + "text": contentStr, + }) + } + + // Add tool_use blocks + for _, tc := range msg.ToolCalls { + var input interface{} + if tc.Function.Arguments != "" { + jsoniter.UnmarshalFromString(tc.Function.Arguments, &input) + } + if input == nil { + input = map[string]interface{}{} + } + contentBlocks = append(contentBlocks, map[string]interface{}{ + "type": "tool_use", + "id": tc.ID, + "name": tc.Function.Name, + "input": input, + }) + } + + apiMsg["content"] = contentBlocks + } + + apiMessages = append(apiMessages, apiMsg) + } + + // Build request body + body := map[string]interface{}{ + "model": model, + } + + if len(apiMessages) > 0 { + body["messages"] = apiMessages + } + + if systemContent != "" { + body["system"] = systemContent + } + + if streaming { + body["stream"] = true + } + + // max_tokens is required for Anthropic + maxTokens := 4096 // default + if options.MaxTokens != nil { + maxTokens = *options.MaxTokens + } else if options.MaxCompletionTokens != nil { + maxTokens = *options.MaxCompletionTokens + } else if mt, ok := setting["max_tokens"].(int); ok && mt > 0 { + maxTokens = mt + } + body["max_tokens"] = maxTokens + + // Temperature + if options.Temperature != nil { + body["temperature"] = *options.Temperature + } + + if options.TopP != nil { + body["top_p"] = *options.TopP + } + + if options.Stop != nil { + body["stop_sequences"] = options.Stop + } + + // Tools (convert from OpenAI format to Anthropic format) + if len(options.Tools) > 0 { + anthropicTools := convertTools(options.Tools) + if len(anthropicTools) > 0 { + body["tools"] = anthropicTools + } + } + + if options.ToolChoice != nil { + body["tool_choice"] = convertToolChoice(options.ToolChoice) + } + + // Thinking configuration from connector settings + if thinking, exists := setting["thinking"]; exists && thinking != nil { + body["thinking"] = thinking + } + + return body, nil +} + +// convertTools converts OpenAI-format tools to Anthropic format +func convertTools(tools []map[string]interface{}) []map[string]interface{} { + result := make([]map[string]interface{}, 0, len(tools)) + for _, tool := range tools { + function, ok := tool["function"].(map[string]interface{}) + if !ok { + continue + } + + anthropicTool := map[string]interface{}{ + "name": function["name"], + } + if desc, ok := function["description"]; ok { + anthropicTool["description"] = desc + } + if params, ok := function["parameters"]; ok { + anthropicTool["input_schema"] = params + } + + result = append(result, anthropicTool) + } + return result +} + +// convertToolChoice converts OpenAI tool_choice to Anthropic format +func convertToolChoice(choice interface{}) interface{} { + switch v := choice.(type) { + case string: + switch v { + case "auto": + return map[string]interface{}{"type": "auto"} + case "none": + return map[string]interface{}{"type": "none"} + case "required": + return map[string]interface{}{"type": "any"} + } + case map[string]interface{}: + if fn, ok := v["function"].(map[string]interface{}); ok { + if name, ok := fn["name"].(string); ok { + return map[string]interface{}{ + "type": "tool", + "name": name, + } + } + } + } + return map[string]interface{}{"type": "auto"} +} + +// convertImagePart converts an OpenAI image_url content part to Anthropic image format +func convertImagePart(part context.ContentPart) map[string]interface{} { + if part.ImageURL == nil { + return map[string]interface{}{"type": "text", "text": "[image not available]"} + } + + url := part.ImageURL.URL + + // Check if it's a base64 data URL + if strings.HasPrefix(url, "data:") { + // Parse data URL: data:image/jpeg;base64, + parts := strings.SplitN(url, ",", 2) + if len(parts) == 2 { + mediaInfo := strings.TrimPrefix(parts[0], "data:") + mediaInfo = strings.TrimSuffix(mediaInfo, ";base64") + return map[string]interface{}{ + "type": "image", + "source": map[string]interface{}{ + "type": "base64", + "media_type": mediaInfo, + "data": parts[1], + }, + } + } + } + + // URL-based image (Anthropic supports URL images) + return map[string]interface{}{ + "type": "image", + "source": map[string]interface{}{ + "type": "url", + "url": url, + }, + } +} + +// buildAPIURL builds the API URL for Anthropic +func buildAPIURL(host, endpoint string) string { + return connector.BuildAPIURL(host, endpoint) +} + +// mapStopReason maps Anthropic stop_reason to OpenAI finish_reason +func mapStopReason(stopReason string) string { + switch stopReason { + case "end_turn": + return "stop" + case "max_tokens": + return "length" + case "tool_use": + return "tool_calls" + case "stop_sequence": + return "stop" + default: + return stopReason + } +} + +// Message tracker helper functions + +func startMessage(mt *messageTracker, messageType message.StreamChunkType, handler message.StreamFunc) { + if mt.active { + endMessage(mt, handler) + } + + mt.active = true + if mt.idGenerator != nil { + mt.messageID = mt.idGenerator.GenerateMessageID() + } else { + mt.messageID = message.GenerateNanoID() + } + mt.messageType = messageType + mt.startTime = time.Now().UnixMilli() + mt.chunkCount = 0 + mt.toolCallInfo = nil + + if handler != nil { + startData := &message.EventMessageStartData{ + MessageID: mt.messageID, + Type: string(messageType), + Timestamp: mt.startTime, + } + if startJSON, err := jsoniter.Marshal(startData); err == nil { + handler(message.ChunkMessageStart, startJSON) + } + } +} + +func startToolCallMessage(mt *messageTracker, toolCallInfo *message.EventToolCallInfo, handler message.StreamFunc) { + if mt.active { + endMessage(mt, handler) + } + + mt.active = true + if mt.idGenerator != nil { + mt.messageID = mt.idGenerator.GenerateMessageID() + } else { + mt.messageID = message.GenerateNanoID() + } + mt.messageType = message.ChunkToolCall + mt.startTime = time.Now().UnixMilli() + mt.chunkCount = 0 + mt.toolCallInfo = toolCallInfo + + if handler != nil { + startData := &message.EventMessageStartData{ + MessageID: mt.messageID, + Type: string(message.ChunkToolCall), + Timestamp: mt.startTime, + ToolCall: toolCallInfo, + } + if startJSON, err := jsoniter.Marshal(startData); err == nil { + handler(message.ChunkMessageStart, startJSON) + } + } +} + +func incrementChunk(mt *messageTracker) { + if mt.active { + mt.chunkCount++ + } +} + +func endMessage(mt *messageTracker, handler message.StreamFunc) { + if !mt.active { + return + } + + if handler != nil { + endData := &message.EventMessageEndData{ + MessageID: mt.messageID, + Type: string(mt.messageType), + Timestamp: time.Now().UnixMilli(), + DurationMs: time.Now().UnixMilli() - mt.startTime, + ChunkCount: mt.chunkCount, + Status: "completed", + } + if mt.toolCallInfo != nil { + endData.ToolCall = mt.toolCallInfo + } + if endJSON, err := jsoniter.Marshal(endData); err == nil { + handler(message.ChunkMessageEnd, endJSON) + } + } + + mt.active = false + mt.messageID = "" + mt.toolCallInfo = nil +} + +// isRetryableError checks if an error is retryable +func isRetryableError(err error) bool { + if err == nil { + return false + } + + errStr := err.Error() + retryablePatterns := []string{ + "timeout", + "connection refused", + "connection reset", + "EOF", + "HTTP 429", + "HTTP 500", + "HTTP 502", + "HTTP 503", + "HTTP 504", + "overloaded", + } + + for _, pattern := range retryablePatterns { + if strings.Contains(strings.ToLower(errStr), strings.ToLower(pattern)) { + return true + } + } + + return false +} diff --git a/agent/llm/providers/anthropic/types.go b/agent/llm/providers/anthropic/types.go new file mode 100644 index 00000000..5e109ac6 --- /dev/null +++ b/agent/llm/providers/anthropic/types.go @@ -0,0 +1,154 @@ +package anthropic + +import ( + "github.com/yaoapp/yao/agent/output/message" +) + +// ============================================================ +// Anthropic Messages API types +// Reference: https://docs.anthropic.com/en/api/messages +// ============================================================ + +// StreamEvent represents an SSE event from Anthropic streaming API +type StreamEvent struct { + Type string `json:"type"` +} + +// MessageStartEvent represents the message_start SSE event +type MessageStartEvent struct { + Type string `json:"type"` + Message MessageStart `json:"message"` +} + +// MessageStart represents the message object in message_start event +type MessageStart struct { + ID string `json:"id"` + Type string `json:"type"` + Role string `json:"role"` + Content []ContentBlock `json:"content"` + Model string `json:"model"` + StopReason *string `json:"stop_reason"` + StopSequence *string `json:"stop_sequence"` + Usage *UsageInfo `json:"usage,omitempty"` +} + +// ContentBlockStartEvent represents the content_block_start SSE event +type ContentBlockStartEvent struct { + Type string `json:"type"` + Index int `json:"index"` + ContentBlock ContentBlock `json:"content_block"` +} + +// ContentBlockDeltaEvent represents the content_block_delta SSE event +type ContentBlockDeltaEvent struct { + Type string `json:"type"` + Index int `json:"index"` + Delta DeltaBlock `json:"delta"` +} + +// ContentBlockStopEvent represents the content_block_stop SSE event +type ContentBlockStopEvent struct { + Type string `json:"type"` + Index int `json:"index"` +} + +// MessageDeltaEvent represents the message_delta SSE event +type MessageDeltaEvent struct { + Type string `json:"type"` + Delta MessageDelta `json:"delta"` + Usage *DeltaUsage `json:"usage,omitempty"` +} + +// MessageDelta represents the delta in message_delta event +type MessageDelta struct { + StopReason string `json:"stop_reason,omitempty"` + StopSequence *string `json:"stop_sequence,omitempty"` +} + +// DeltaUsage represents usage in message_delta event +type DeltaUsage struct { + OutputTokens int `json:"output_tokens"` +} + +// ContentBlock represents a content block in the response +type ContentBlock struct { + Type string `json:"type"` // "text", "thinking", "tool_use" + Text string `json:"text,omitempty"` // for type "text" + Thinking string `json:"thinking,omitempty"` // for type "thinking" + Signature string `json:"signature,omitempty"` // for type "thinking" + ID string `json:"id,omitempty"` // for type "tool_use" + Name string `json:"name,omitempty"` // for type "tool_use" + Input interface{} `json:"input,omitempty"` // for type "tool_use" +} + +// DeltaBlock represents a delta block in streaming +type DeltaBlock struct { + Type string `json:"type"` // "text_delta", "thinking_delta", "input_json_delta" + Text string `json:"text,omitempty"` // for type "text_delta" + Thinking string `json:"thinking,omitempty"` // for type "thinking_delta" + PartialJSON string `json:"partial_json,omitempty"` // for type "input_json_delta" +} + +// UsageInfo represents token usage information +type UsageInfo struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` +} + +// NonStreamResponse represents the full non-streaming response from Anthropic API +type NonStreamResponse struct { + ID string `json:"id"` + Type string `json:"type"` + Role string `json:"role"` + Content []ContentBlock `json:"content"` + Model string `json:"model"` + StopReason string `json:"stop_reason"` + StopSequence *string `json:"stop_sequence"` + Usage *UsageInfo `json:"usage,omitempty"` +} + +// APIError represents an error response from Anthropic API +type APIError struct { + Type string `json:"type"` + Error struct { + Type string `json:"type"` + Message string `json:"message"` + } `json:"error"` +} + +// streamAccumulator accumulates streaming response data +type streamAccumulator struct { + id string + model string + role string + content string + thinkingContent string + thinkingSignature string + toolCalls map[int]*accumulatedToolCall + stopReason string + usage *message.UsageInfo + + // Current content block tracking + currentBlockIndex int + currentBlockType string +} + +// accumulatedToolCall accumulates a single tool call from streaming +type accumulatedToolCall struct { + id string + name string + inputJSON string +} + +// messageTracker tracks message lifecycle for stream events +type messageTracker struct { + active bool + messageID string + messageType message.StreamChunkType + startTime int64 + chunkCount int + toolCallInfo *message.EventToolCallInfo + idGenerator *message.IDGenerator +} diff --git a/agent/llm/providers/factory.go b/agent/llm/providers/factory.go index 3fa1165c..ede08740 100644 --- a/agent/llm/providers/factory.go +++ b/agent/llm/providers/factory.go @@ -4,7 +4,9 @@ import ( "fmt" "github.com/yaoapp/gou/connector" + gouAnthropicConn "github.com/yaoapp/gou/connector/anthropic" "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/llm/providers/anthropic" "github.com/yaoapp/yao/agent/llm/providers/openai" "github.com/yaoapp/yao/agent/output/message" ) @@ -40,10 +42,15 @@ func SelectProvider(conn connector.Connector, options *context.CompletionOptions // - Reasoning (o1, GPT-4o thinking, etc.) return openai.New(conn, options.Capabilities), nil - case "claude": - // TODO: Implement Claude provider - // For now, use OpenAI provider (may have compatibility issues) - return openai.New(conn, options.Capabilities), nil + case "anthropic": + // Anthropic Messages API (Claude, Kimi Code, etc.) + // Check if connector has native Anthropic capabilities + settings := conn.Setting() + if caps, ok := settings["capabilities"].(*gouAnthropicConn.Capabilities); ok { + return anthropic.NewFromAnthropicCaps(conn, caps), nil + } + // Fallback: use OpenAI capabilities (converted from connector settings) + return anthropic.New(conn, options.Capabilities), nil default: // Default to OpenAI-compatible provider @@ -53,21 +60,24 @@ func SelectProvider(conn connector.Connector, options *context.CompletionOptions // DetectAPIFormat detects the API format from connector func DetectAPIFormat(conn connector.Connector) string { - // Check connector type + // Check connector type directly + if conn.Is(connector.ANTHROPIC) { + return "anthropic" + } + if conn.Is(connector.OPENAI) { return "openai" } - // Check connector settings for host URL + // Check connector settings for host URL patterns as fallback settings := conn.Setting() if settings != nil { if host, ok := settings["host"].(string); ok { - // Detect by host URL patterns - if contains(host, "anthropic.com") || contains(host, "claude") { - return "claude" + if contains(host, "anthropic.com") || contains(host, "api.kimi.com/coding") { + return "anthropic" } if contains(host, "deepseek.com") { - return "openai" // DeepSeek uses OpenAI-compatible API + return "openai" } } } diff --git a/agent/sandbox/claude/command.go b/agent/sandbox/claude/command.go index 252c5472..b09d204f 100644 --- a/agent/sandbox/claude/command.go +++ b/agent/sandbox/claude/command.go @@ -338,9 +338,27 @@ func buildEnvironment(opts *Options, systemPrompt string) map[string]string { // Explicitly set XAUTHORITY to the correct path. env["XAUTHORITY"] = "/home/sandbox/.Xauthority" - // claude-proxy runs on localhost:3456, Claude CLI connects to it - env["ANTHROPIC_BASE_URL"] = "http://127.0.0.1:3456" - env["ANTHROPIC_API_KEY"] = "dummy" // Proxy doesn't verify this + if opts.ConnectorType == "anthropic" { + // Anthropic mode: Claude CLI connects directly to the Anthropic-compatible backend + // No proxy needed — the backend already speaks Anthropic Messages API + env["ANTHROPIC_BASE_URL"] = opts.ConnectorHost + env["ANTHROPIC_API_KEY"] = opts.ConnectorKey + } else { + // OpenAI mode (default): Claude CLI connects to claude-proxy on localhost:3456 + // The proxy translates Anthropic Messages API → OpenAI Chat Completions API + env["ANTHROPIC_BASE_URL"] = "http://127.0.0.1:3456" + env["ANTHROPIC_API_KEY"] = "dummy" // Proxy doesn't verify this + } + + // Set model environment variables from connector + // Claude CLI uses these to select the model for all roles + if opts.Model != "" { + env["ANTHROPIC_MODEL"] = opts.Model + env["ANTHROPIC_DEFAULT_OPUS_MODEL"] = opts.Model + env["ANTHROPIC_DEFAULT_SONNET_MODEL"] = opts.Model + env["ANTHROPIC_DEFAULT_HAIKU_MODEL"] = opts.Model + env["CLAUDE_CODE_SUBAGENT_MODEL"] = opts.Model + } // Pass secrets as environment variables for Claude CLI to use // These are configured in package.yao sandbox.secrets (e.g., LLM_API_KEY, GITHUB_TOKEN) diff --git a/agent/sandbox/claude/executor.go b/agent/sandbox/claude/executor.go index 1b03aaa5..1cf859b9 100644 --- a/agent/sandbox/claude/executor.go +++ b/agent/sandbox/claude/executor.go @@ -38,6 +38,7 @@ type Options struct { ConnectorHost string ConnectorKey string Model string + ConnectorType string // Connector API type: "openai" or "anthropic" ConnectorOptions map[string]interface{} // Extra connector options (e.g., thinking, max_tokens) Secrets map[string]string // Secrets to pass to container (e.g., GITHUB_TOKEN) } @@ -330,6 +331,12 @@ func (e *Executor) startClaudeProxy(ctx context.Context) error { return nil } + // Skip proxy for Anthropic connectors — Claude CLI connects directly + // The backend already speaks Anthropic Messages API, no conversion needed + if e.opts.ConnectorType == "anthropic" { + return nil + } + // Build proxy config configJSON, err := BuildProxyConfig(e.opts) if err != nil { diff --git a/agent/sandbox/executor.go b/agent/sandbox/executor.go index 89ba56a6..4fcbf821 100644 --- a/agent/sandbox/executor.go +++ b/agent/sandbox/executor.go @@ -41,6 +41,7 @@ func New(manager *infraSandbox.Manager, opts *Options) (Executor, error) { ConnectorHost: opts.ConnectorHost, ConnectorKey: opts.ConnectorKey, Model: opts.Model, + ConnectorType: opts.ConnectorType, // "openai" or "anthropic" ConnectorOptions: opts.ConnectorOptions, // Extra options like thinking, max_tokens Secrets: opts.Secrets, // Secrets for container env vars } diff --git a/agent/sandbox/types.go b/agent/sandbox/types.go index 209944a2..eddce3db 100644 --- a/agent/sandbox/types.go +++ b/agent/sandbox/types.go @@ -94,6 +94,10 @@ type Options struct { ConnectorKey string `json:"-"` Model string `json:"-"` + // ConnectorType - connector API type: "openai" or "anthropic" + // Determines whether to use claude-proxy (openai) or direct connection (anthropic) + ConnectorType string `json:"-"` + // ConnectorOptions - extra options from connector config (e.g., thinking, max_tokens, temperature) // These are backend-specific parameters passed to the proxy ConnectorOptions map[string]interface{} `json:"-"` diff --git a/openapi/llm/llm.go b/openapi/llm/llm.go index 30a8374a..b162018c 100644 --- a/openapi/llm/llm.go +++ b/openapi/llm/llm.go @@ -59,13 +59,13 @@ func listProviders(c *gin.Context) { // Get user-defined model capabilities once at the start of request modelCapabilities := getModelCapabilities() - // Get all OpenAI-compatible LLM connectors from AIConnectors - // Note: All openai type connectors are automatically added to AIConnectors during loading + // Get all LLM connectors from AIConnectors + // Note: All AI type connectors (openai, anthropic, fastembed) are automatically added to AIConnectors during loading // See gou/connector/connector.go LoadSource() for details for _, opt := range connector.AIConnectors { connType := getConnectorType(opt.Value) - // Only include OpenAI-compatible LLM connectors - if connType == "openai" { + // Include OpenAI-compatible and Anthropic LLM connectors + if connType == "openai" || connType == "anthropic" { conn, ok := connector.Connectors[opt.Value] if !ok { continue @@ -99,11 +99,14 @@ func getConnectorType(id string) string { return "unknown" } - // Only return openai type (OpenAI-compatible format) if conn.Is(connector.OPENAI) { return "openai" } + if conn.Is(connector.ANTHROPIC) { + return "anthropic" + } + return "unknown" }