622 lines
18 KiB
Go
622 lines
18 KiB
Go
package externalapi
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"mime/multipart"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"synapse/pkg/types"
|
||
)
|
||
|
||
// =============================================================================
|
||
// External API Layer - 非Agent调用者API
|
||
// =============================================================================
|
||
//
|
||
// 这一层为非Agent调用者（如普通服务、前端、第三方应用）提供了更简单的RESTful API，
|
||
// 隐藏了Synapse协议的复杂性。
|
||
//
|
||
// 与Synapse协议的区别：
|
||
// - 使用API Key认证，而非HMAC签名
|
||
// - 简化的请求/响应格式
|
||
// - 支持回调和轮询两种模式
|
||
// - 无需注册为Agent即可调用
|
||
// =============================================================================
|
||
|
||
// =============================================================================
|
||
// API Handler
|
||
// =============================================================================
|
||
|
||
// ExternalAPIHandler 处理外部API请求
|
||
type ExternalAPIHandler struct {
|
||
// TaskExecutor 任务执行器
|
||
TaskExecutor TaskExecutor
|
||
|
||
// APIKey API密钥（如果为空则不验证）
|
||
APIKey string
|
||
|
||
// MaxFileSize 最大文件大小
|
||
MaxFileSize int64
|
||
}
|
||
|
||
// TaskExecutor 任务执行器接口
|
||
type TaskExecutor interface {
|
||
// SubmitTask 提交任务
|
||
SubmitTask(ctx context.Context, taskType string, payload map[string]any, externalTaskID string, priority string, callbackURL string, callbackToken string) (string, string, error)
|
||
|
||
// GetTaskStatus 获取任务状态
|
||
GetTaskStatus(ctx context.Context, taskID string) (*types.TaskInfo, error)
|
||
|
||
// ListTasks 列出任务
|
||
ListTasks(ctx context.Context, status string, limit, offset int) ([]types.TaskInfo, int, error)
|
||
|
||
// CancelTask 取消任务
|
||
CancelTask(ctx context.Context, taskID string) error
|
||
}
|
||
|
||
// NewExternalAPIHandler 创建外部API处理器
|
||
func NewExternalAPIHandler(executor TaskExecutor, apiKey string) *ExternalAPIHandler {
|
||
return &ExternalAPIHandler{
|
||
TaskExecutor: executor,
|
||
APIKey: apiKey,
|
||
MaxFileSize: 100 << 20, // 100MB
|
||
}
|
||
}
|
||
|
||
// RegisterRoutes 注册路由
|
||
func (h *ExternalAPIHandler) RegisterRoutes(mux *http.ServeMux, prefix string) {
|
||
// 任务相关
|
||
mux.HandleFunc(prefix+"/task", h.withAuth(h.handleTaskCreate))
|
||
mux.HandleFunc(prefix+"/task/", h.withAuth(h.handleTaskDetail))
|
||
mux.HandleFunc(prefix+"/tasks", h.withAuth(h.handleTaskList))
|
||
|
||
// 多部分表单提交
|
||
mux.HandleFunc(prefix+"/task/multipart", h.withAuth(h.handleTaskMultipart))
|
||
|
||
// 能力查询
|
||
mux.HandleFunc(prefix+"/capabilities", h.withAuth(h.handleCapabilities))
|
||
}
|
||
|
||
// =============================================================================
|
||
// 认证中间件
|
||
// =============================================================================
|
||
|
||
func (h *ExternalAPIHandler) withAuth(next http.HandlerFunc) http.HandlerFunc {
|
||
return func(w http.ResponseWriter, r *http.Request) {
|
||
// 如果没有配置API Key，跳过验证
|
||
if h.APIKey != "" {
|
||
key := r.Header.Get("X-API-Key")
|
||
if key == "" {
|
||
key = r.URL.Query().Get("api_key")
|
||
}
|
||
if key != h.APIKey {
|
||
h.writeError(w, http.StatusUnauthorized, "unauthorized", "invalid or missing API key")
|
||
return
|
||
}
|
||
}
|
||
|
||
// CORS
|
||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, X-API-Key")
|
||
|
||
if r.Method == "OPTIONS" {
|
||
w.WriteHeader(http.StatusOK)
|
||
return
|
||
}
|
||
|
||
next(w, r)
|
||
}
|
||
}
|
||
|
||
// =============================================================================
|
||
// 任务创建
|
||
// =============================================================================
|
||
|
||
// ExternalTaskRequest 外部任务请求
|
||
type ExternalTaskRequest struct {
|
||
TaskType string `json:"task_type"`
|
||
Payload map[string]any `json:"payload"`
|
||
ExternalTaskID string `json:"external_task_id,omitempty"`
|
||
Priority string `json:"priority,omitempty"`
|
||
CallbackURL string `json:"callback_url,omitempty"`
|
||
CallbackToken string `json:"callback_token,omitempty"`
|
||
}
|
||
|
||
func (h *ExternalAPIHandler) handleTaskCreate(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only POST is supported")
|
||
return
|
||
}
|
||
|
||
body, err := io.ReadAll(r.Body)
|
||
if err != nil {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "failed to read body")
|
||
return
|
||
}
|
||
|
||
var req ExternalTaskRequest
|
||
if err := json.Unmarshal(body, &req); err != nil {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "invalid JSON format")
|
||
return
|
||
}
|
||
|
||
// 验证必填字段
|
||
if req.TaskType == "" {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "task_type is required")
|
||
return
|
||
}
|
||
if req.Payload == nil {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "payload is required")
|
||
return
|
||
}
|
||
|
||
// 默认优先级
|
||
if req.Priority == "" {
|
||
req.Priority = "normal"
|
||
}
|
||
|
||
// 提交任务
|
||
taskID, status, err := h.TaskExecutor.SubmitTask(
|
||
r.Context(),
|
||
req.TaskType,
|
||
req.Payload,
|
||
req.ExternalTaskID,
|
||
req.Priority,
|
||
req.CallbackURL,
|
||
req.CallbackToken,
|
||
)
|
||
if err != nil {
|
||
log.Printf("[ExternalAPI] 任务提交失败: %v", err)
|
||
h.writeError(w, http.StatusInternalServerError, "task_submission_failed", err.Error())
|
||
return
|
||
}
|
||
|
||
log.Printf("[ExternalAPI] 任务提交成功: type=%s, task_id=%s", req.TaskType, taskID)
|
||
|
||
h.writeJSON(w, http.StatusAccepted, map[string]any{
|
||
"success": true,
|
||
"task_id": taskID,
|
||
"status": status,
|
||
"message": "任务已接收，正在处理中",
|
||
})
|
||
}
|
||
|
||
// =============================================================================
|
||
// Multipart任务创建（带文件）
|
||
// =============================================================================
|
||
|
||
func (h *ExternalAPIHandler) handleTaskMultipart(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only POST is supported")
|
||
return
|
||
}
|
||
|
||
// 限制请求大小
|
||
r.Body = http.MaxBytesReader(w, r.Body, h.MaxFileSize)
|
||
|
||
if err := r.ParseMultipartForm(h.MaxFileSize); err != nil {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "failed to parse multipart form: "+err.Error())
|
||
return
|
||
}
|
||
|
||
// 获取请求JSON
|
||
requestStr := r.FormValue("request")
|
||
if requestStr == "" {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "missing 'request' field")
|
||
return
|
||
}
|
||
|
||
var req ExternalTaskRequest
|
||
if err := json.Unmarshal([]byte(requestStr), &req); err != nil {
|
||
h.writeError(w, http.StatusBadRequest, "bad_request", "invalid request JSON")
|
||
return
|
||
}
|
||
|
||
// 提取文件
|
||
files := make(map[string][]byte)
|
||
if r.MultipartForm != nil {
|
||
for fieldName, fileHeaders := range r.MultipartForm.File {
|
||
if fieldName == "request" {
|
||
continue
|
||
}
|
||
for _, fh := range fileHeaders {
|
||
file, err := fh.Open()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
data, err := io.ReadAll(file)
|
||
file.Close()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
files[fieldName] = data
|
||
}
|
||
}
|
||
}
|
||
|
||
// 将文件信息加入payload
|
||
if len(files) > 0 {
|
||
if req.Payload == nil {
|
||
req.Payload = make(map[string]any)
|
||
}
|
||
req.Payload["__files"] = files
|
||
}
|
||
|
||
// 提交任务
|
||
taskID, status, err := h.TaskExecutor.SubmitTask(
|
||
r.Context(),
|
||
req.TaskType,
|
||
req.Payload,
|
||
req.ExternalTaskID,
|
||
req.Priority,
|
||
req.CallbackURL,
|
||
req.CallbackToken,
|
||
)
|
||
if err != nil {
|
||
h.writeError(w, http.StatusInternalServerError, "task_submission_failed", err.Error())
|
||
return
|
||
}
|
||
|
||
h.writeJSON(w, http.StatusAccepted, map[string]any{
|
||
"success": true,
|
||
"task_id": taskID,
|
||
"status": status,
|
||
"message": "任务已接收，正在处理中",
|
||
"files_uploaded": len(files),
|
||
})
|
||
}
|
||
|
||
// =============================================================================
|
||
// 任务详情/状态
|
||
// =============================================================================
|
||
|
||
func (h *ExternalAPIHandler) handleTaskDetail(w http.ResponseWriter, r *http.Request) {
|
||
path := strings.TrimPrefix(r.URL.Path, "/api/v1/synapse/task/")
|
||
parts := strings.SplitN(path, "/", 2)
|
||
taskID := parts[0]
|
||
|
||
if taskID == "" {
|
||
h.writeError(w, http.StatusNotFound, "not_found", "task_id is required")
|
||
return
|
||
}
|
||
|
||
// /task/{id}/status
|
||
if len(parts) == 2 {
|
||
switch parts[1] {
|
||
case "status":
|
||
h.handleTaskStatus(w, r, taskID)
|
||
return
|
||
case "cancel":
|
||
h.handleTaskCancel(w, r, taskID)
|
||
return
|
||
default:
|
||
h.writeError(w, http.StatusNotFound, "not_found", "unknown subpath")
|
||
return
|
||
}
|
||
}
|
||
|
||
// /task/{id} (GET)
|
||
if r.Method != http.MethodGet {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only GET is supported")
|
||
return
|
||
}
|
||
|
||
task, err := h.TaskExecutor.GetTaskStatus(r.Context(), taskID)
|
||
if err != nil {
|
||
h.writeError(w, http.StatusNotFound, "task_not_found", err.Error())
|
||
return
|
||
}
|
||
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"task": task,
|
||
})
|
||
}
|
||
|
||
func (h *ExternalAPIHandler) handleTaskStatus(w http.ResponseWriter, r *http.Request, taskID string) {
|
||
if r.Method != http.MethodGet {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only GET is supported")
|
||
return
|
||
}
|
||
|
||
task, err := h.TaskExecutor.GetTaskStatus(r.Context(), taskID)
|
||
if err != nil {
|
||
h.writeError(w, http.StatusNotFound, "task_not_found", err.Error())
|
||
return
|
||
}
|
||
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"task_id": task.ID,
|
||
"status": task.Status,
|
||
"progress": task.Progress,
|
||
"message": task.Message,
|
||
})
|
||
}
|
||
|
||
func (h *ExternalAPIHandler) handleTaskCancel(w http.ResponseWriter, r *http.Request, taskID string) {
|
||
if r.Method != http.MethodPost {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only POST is supported")
|
||
return
|
||
}
|
||
|
||
if err := h.TaskExecutor.CancelTask(r.Context(), taskID); err != nil {
|
||
h.writeError(w, http.StatusInternalServerError, "cancel_failed", err.Error())
|
||
return
|
||
}
|
||
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"task_id": taskID,
|
||
"status": "cancelled",
|
||
"message": "任务已取消",
|
||
})
|
||
}
|
||
|
||
// =============================================================================
|
||
// 任务列表
|
||
// =============================================================================
|
||
|
||
func (h *ExternalAPIHandler) handleTaskList(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodGet {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only GET is supported")
|
||
return
|
||
}
|
||
|
||
status := r.URL.Query().Get("status")
|
||
limitStr := r.URL.Query().Get("limit")
|
||
offsetStr := r.URL.Query().Get("offset")
|
||
|
||
limit := 20
|
||
offset := 0
|
||
if limitStr != "" {
|
||
if n, err := strconv.Atoi(limitStr); err == nil && n > 0 && n <= 100 {
|
||
limit = n
|
||
}
|
||
}
|
||
if offsetStr != "" {
|
||
if n, err := strconv.Atoi(offsetStr); err == nil && n >= 0 {
|
||
offset = n
|
||
}
|
||
}
|
||
|
||
tasks, total, err := h.TaskExecutor.ListTasks(r.Context(), status, limit, offset)
|
||
if err != nil {
|
||
h.writeError(w, http.StatusInternalServerError, "list_failed", err.Error())
|
||
return
|
||
}
|
||
|
||
hasMore := offset+limit < total
|
||
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"tasks": tasks,
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
"has_more": hasMore,
|
||
})
|
||
}
|
||
|
||
// =============================================================================
|
||
// 能力查询
|
||
// =============================================================================
|
||
|
||
// CapabilitiesProvider 能力提供者接口
|
||
type CapabilitiesProvider interface {
|
||
GetCapabilities(ctx context.Context) (*CapabilitiesInfo, error)
|
||
}
|
||
|
||
// CapabilitiesInfo 能力信息
|
||
type CapabilitiesInfo struct {
|
||
AgentID string `json:"agent_id"`
|
||
AgentName string `json:"agent_name"`
|
||
SystemID string `json:"system_id"`
|
||
ProtocolVersion string `json:"protocol_version"`
|
||
SupportedTaskTypes []SupportedTaskType `json:"supported_task_types"`
|
||
SupportedContentTypes []string `json:"supported_content_types"`
|
||
SupportedTransportModes []string `json:"supported_transport_modes"`
|
||
MaxFileSize int64 `json:"max_file_size"`
|
||
MaxConcurrentTasks int `json:"max_concurrent_tasks"`
|
||
MaxTaskDuration int `json:"max_task_duration_seconds"`
|
||
SupportProgressUpdates bool `json:"support_progress_updates"`
|
||
SupportStreaming bool `json:"support_streaming"`
|
||
SupportHiL bool `json:"support_hil"`
|
||
SupportGroups bool `json:"support_groups"`
|
||
}
|
||
|
||
// SupportedTaskType 支持的任务类型
|
||
type SupportedTaskType struct {
|
||
Type string `json:"type"`
|
||
Name string `json:"name"`
|
||
Description string `json:"description"`
|
||
EstimatedTime string `json:"estimated_time"`
|
||
InputSchema string `json:"input_schema,omitempty"`
|
||
OutputSchema string `json:"output_schema,omitempty"`
|
||
Capabilities []string `json:"capabilities,omitempty"`
|
||
}
|
||
|
||
// CapabilitiesProviderField 能力提供者字段（可选注入）
|
||
var CapabilitiesProviderField CapabilitiesProvider
|
||
|
||
func (h *ExternalAPIHandler) handleCapabilities(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodGet {
|
||
h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only GET is supported")
|
||
return
|
||
}
|
||
|
||
// 如果有注入的能力提供者，使用它
|
||
if CapabilitiesProviderField != nil {
|
||
caps, err := CapabilitiesProviderField.GetCapabilities(r.Context())
|
||
if err != nil {
|
||
h.writeError(w, http.StatusInternalServerError, "capabilities_error", err.Error())
|
||
return
|
||
}
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"capabilities": caps,
|
||
})
|
||
return
|
||
}
|
||
|
||
// 默认能力信息
|
||
caps := &CapabilitiesInfo{
|
||
AgentID: "synapse-agent",
|
||
AgentName: "Synapse Agent",
|
||
SystemID: "synapse",
|
||
ProtocolVersion: "1.0.0",
|
||
SupportedTaskTypes: []SupportedTaskType{},
|
||
SupportedContentTypes: []string{"application/json", "text/plain", "application/octet-stream"},
|
||
SupportedTransportModes: []string{"callback", "polling"},
|
||
MaxFileSize: h.MaxFileSize,
|
||
MaxConcurrentTasks: 10,
|
||
MaxTaskDuration: 300,
|
||
SupportProgressUpdates: true,
|
||
SupportStreaming: false,
|
||
SupportHiL: true,
|
||
SupportGroups: true,
|
||
}
|
||
|
||
h.writeJSON(w, http.StatusOK, map[string]any{
|
||
"success": true,
|
||
"capabilities": caps,
|
||
})
|
||
}
|
||
|
||
// =============================================================================
|
||
// 回调发送器
|
||
// =============================================================================
|
||
|
||
// CallbackSender 发送回调通知
|
||
type CallbackSender struct {
|
||
HTTPClient *http.Client
|
||
}
|
||
|
||
// NewCallbackSender 创建回调发送器
|
||
func NewCallbackSender() *CallbackSender {
|
||
return &CallbackSender{
|
||
HTTPClient: &http.Client{Timeout: 10 * time.Second},
|
||
}
|
||
}
|
||
|
||
// CallbackPayload 回调负载
|
||
type CallbackPayload struct {
|
||
MessageID string `json:"message_id"`
|
||
MessageType string `json:"message_type"` // task_status, task_complete, task_failed
|
||
TaskID string `json:"task_id"`
|
||
Status string `json:"status"`
|
||
Progress int `json:"progress"`
|
||
Result map[string]any `json:"result,omitempty"`
|
||
Error *CallbackError `json:"error,omitempty"`
|
||
Timestamp string `json:"timestamp"`
|
||
}
|
||
|
||
// CallbackError 回调错误
|
||
type CallbackError struct {
|
||
Code string `json:"code"`
|
||
Message string `json:"message"`
|
||
}
|
||
|
||
// SendCallback 发送回调
|
||
func (s *CallbackSender) SendCallback(ctx context.Context, callbackURL, callbackToken string, payload *CallbackPayload) error {
|
||
if callbackURL == "" {
|
||
return nil
|
||
}
|
||
|
||
body, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
req, err := http.NewRequestWithContext(ctx, "POST", callbackURL, bytes.NewReader(body))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
req.Header.Set("Content-Type", "application/json")
|
||
if callbackToken != "" {
|
||
req.Header.Set("X-Callback-Token", callbackToken)
|
||
}
|
||
|
||
resp, err := s.HTTPClient.Do(req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode >= 400 {
|
||
respBody, _ := io.ReadAll(resp.Body)
|
||
return fmt.Errorf("callback returned HTTP %d: %s", resp.StatusCode, string(respBody))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// =============================================================================
|
||
// 辅助函数
|
||
// =============================================================================
|
||
|
||
func (h *ExternalAPIHandler) writeJSON(w http.ResponseWriter, status int, data map[string]any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(status)
|
||
json.NewEncoder(w).Encode(data)
|
||
}
|
||
|
||
func (h *ExternalAPIHandler) writeError(w http.ResponseWriter, status int, code, message string) {
|
||
h.writeJSON(w, status, map[string]any{
|
||
"success": false,
|
||
"error": code,
|
||
"message": message,
|
||
})
|
||
}
|
||
|
||
// FileInfo 上传的文件信息
|
||
type FileInfo struct {
|
||
Fieldname string
|
||
Filename string
|
||
Size int64
|
||
Data []byte
|
||
Header multipart.FileHeader
|
||
}
|
||
|
||
// ExtractFiles 从multipart请求中提取文件
|
||
func ExtractFiles(r *http.Request) ([]FileInfo, error) {
|
||
if r.MultipartForm == nil {
|
||
if err := r.ParseMultipartForm(32 << 20); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
var files []FileInfo
|
||
for fieldName, fileHeaders := range r.MultipartForm.File {
|
||
for _, fh := range fileHeaders {
|
||
file, err := fh.Open()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
data, err := io.ReadAll(file)
|
||
file.Close()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
files = append(files, FileInfo{
|
||
Fieldname: fieldName,
|
||
Filename: fh.Filename,
|
||
Size: fh.Size,
|
||
Data: data,
|
||
Header: *fh,
|
||
})
|
||
}
|
||
}
|
||
|
||
return files, nil
|
||
}
|