灵隙(synapse) v1.0.0 版本

This commit is contained in:
lychang
2026-06-28 03:24:26 +08:00
parent 88b606b9ca
commit 812f68d5aa
48 changed files with 5566 additions and 1499 deletions

View File

@@ -0,0 +1,621 @@
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
}