package com.agent.management.external; import com.agent.management.config.ExternalApiAuthFilter; import com.agent.management.config.ExternalApiProperties; import com.agent.management.engine.AgentIgnoreFilter; import com.agent.management.engine.WorkflowEngine; import com.agent.management.engine.WorkflowRunDirManager; import com.agent.management.engine.WorkflowRunEvent; import com.agent.management.external.dto.ApiError; import com.agent.management.external.dto.ExternalDtos.CreateRunRequest; import com.agent.management.external.dto.ExternalDtos.CreateRunResponse; import com.agent.management.external.dto.ExternalDtos.FieldDef; import com.agent.management.external.dto.ExternalDtos.Links; import com.agent.management.external.dto.ExternalDtos.LogEntry; import com.agent.management.external.dto.ExternalDtos.NodeLogsResponse; import com.agent.management.external.dto.ExternalDtos.NodeResult; import com.agent.management.external.dto.ExternalDtos.RunResultResponse; import com.agent.management.external.dto.ExternalDtos.StartRunResponse; import com.agent.management.external.dto.ExternalDtos.WorkflowListResponse; import com.agent.management.external.dto.ExternalDtos.WorkflowSummary; import com.agent.management.external.dto.ExternalDtos.WorkspaceInfo; import com.agent.management.model.entity.Workflow; import com.agent.management.model.entity.WorkflowRun; import com.agent.management.model.entity.WorkflowRunNode; import com.agent.management.repository.WorkflowRepository; import com.agent.management.repository.WorkflowRunNodeRepository; import com.agent.management.repository.WorkflowRunRepository; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.core.io.Resource; import org.springframework.core.io.UrlResource; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.util.*; import java.util.stream.Collectors; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; /** * 外部 API（/api/v1/**）入口：第三方系统通过此 Controller 调用平台工作流。 * *

URL 路径使用 kebab-case 唯一标识 {@code workflowName}（与 Skill 一致）。 * 旧调用方使用数字 id 的 URL 仍可通过 by-id 兜底逻辑访问（用户手动改名为非数字前都自动兼容）。 * * 流程： * - POST /workflows/{workflowName}/runs 创建运行 * - POST /workflows/{workflowName}/runs/{runId}/start 启动执行 * - GET /workflows/{workflowName}/runs/{runId}/stream 订阅 SSE 流 * - GET /workflows/{workflowName}/runs/{runId} 查询运行结果 * - GET /workflows/{workflowName}/runs/{runId}/workspace 下载工作空间 zip * - GET /workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs 查询节点日志 * - GET /workflows 列出当前 API Key 可调用的工作流 */ @Slf4j @RestController @RequestMapping("/api/v1") @RequiredArgsConstructor public class ExternalApiController { private final ExternalApiProperties properties; private final ExternalRunRegistry runRegistry; private final SseEventBus eventBus; private final WorkflowEngine workflowEngine; private final WorkflowRepository workflowRepository; private final WorkflowRunRepository workflowRunRepository; private final WorkflowRunNodeRepository workflowRunNodeRepository; private final WorkflowRunDirManager runDirManager; private final AgentIgnoreFilter agentIgnoreFilter; private static final ObjectMapper MAPPER = new ObjectMapper().registerModule(new JavaTimeModule()); private static final int DEFAULT_TTL_HOURS = 720; private static final long SSE_TIMEOUT_MS = 1_800_000L; // ============================================================ // POST /workflows/{workflowName}/runs — 创建运行 // ============================================================ @PostMapping(value = "/workflows/{workflowName}/runs", consumes = MediaType.APPLICATION_JSON_VALUE) public ResponseEntity createRunJson(@PathVariable String workflowName, @RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry, @RequestBody(required = false) CreateRunRequest req) { return doCreateRun(workflowName, req, null, null); } @PostMapping(value = "/workflows/{workflowName}/runs", consumes = MediaType.MULTIPART_FORM_DATA_VALUE) public ResponseEntity createRunMultipart(@PathVariable String workflowName, @RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry, @RequestParam(value = "payload", required = false) String payloadJson, @RequestParam(value = "files", required = false) MultipartFile[] files, @RequestHeader(value = "X-Relative-Path", required = false) List relativePaths) { CreateRunRequest req = parsePayload(payloadJson); return doCreateRun(workflowName, req, files, relativePaths); } private ResponseEntity doCreateRun(String workflowName, CreateRunRequest req, MultipartFile[] files, List relativePaths) { // 1. 工作流解析 Workflow wf = resolveWorkflow(workflowName); if (wf == null) { return error(404, "WORKFLOW_NOT_FOUND", "工作流不存在: " + workflowName); } if (wf.getGraphData() == null || wf.getGraphData().isBlank() || wf.getGraphData().equals("{\"nodes\":[],\"edges\":[]}")) { return error(422, "GRAPH_INVALID", "工作流图为空，无法执行"); } // 2. 准备 inputs Map inputs = new HashMap<>(); if (req != null && req.getVariables() != null) { inputs.putAll(req.getVariables()); } // 3. 生成 runId + 创建工作目录 + 写入上传文件 String runId; try { runId = generateRunId(); Path runDir = runDirManager.createRunDir(wf.getId(), wf.getName(), runId); List uploadedFiles = new ArrayList<>(); if (files != null) { for (int i = 0; i < files.length; i++) { MultipartFile file = files[i]; if (file == null || file.isEmpty()) continue; String rel = (relativePaths != null && i < relativePaths.size() && !relativePaths.get(i).isBlank()) ? relativePaths.get(i) : file.getOriginalFilename(); if (rel == null || rel.isBlank()) continue; Path dest = runDir.resolve(rel).normalize(); if (!dest.startsWith(runDir)) { return error(400, "VALIDATION_FAILED", "非法上传路径: " + rel); } Files.createDirectories(dest.getParent()); file.transferTo(dest.toFile()); uploadedFiles.add(runDir.relativize(dest).toString().replace('\\', '/')); } } if (!uploadedFiles.isEmpty()) { inputs.put("uploadedFiles", uploadedFiles); } } catch (IllegalArgumentException e) { return error(400, "VALIDATION_FAILED", e.getMessage()); } catch (IOException e) { log.error("[ExternalApi] 创建运行失败 workflowName={}", workflowName, e); return error(500, "INTERNAL_ERROR", "工作目录创建失败"); } // 4. 注册 CREATED 状态 int ttlHours = (req != null && req.getTtlHours() != null && req.getTtlHours() > 0) ? req.getTtlHours() : DEFAULT_TTL_HOURS; runRegistry.register(runId, wf.getId(), wf.getName(), inputs, ttlHours); eventBus.register(runId); // 5. async=false：立即启动并直接进入 SSE 流 boolean async = req == null || req.getAsync() == null || req.getAsync(); if (!async) { return startRun(wf.getName(), wf.getId(), runId, inputs); } // 6. async=true：返回 runId + links（URL 使用 workflowName） CreateRunResponse body = new CreateRunResponse(); body.setRunId(runId); body.setWorkflowId(wf.getId()); body.setWorkflowName(wf.getName()); body.setDisplayName(wf.getDisplayName()); body.setStatus("CREATED"); body.setCreatedAt(Instant.now()); Links links = new Links(); String base = "/api/v1/workflows/" + wf.getName() + "/runs/" + runId; links.setStart(base + "/start"); links.setStream(base + "/stream"); links.setResult(base); links.setWorkspace(base + "/workspace"); body.setLinks(links); return ResponseEntity.ok(ApiError.ok(body)); } // ============================================================ // POST /workflows/{workflowName}/runs/{runId}/start — 启动运行 // ============================================================ @PostMapping("/workflows/{workflowName}/runs/{runId}/start") public ResponseEntity startRun(@PathVariable String workflowName, @PathVariable String runId) { ExternalRunRegistry.CreatedRun created = runRegistry.consume(runId); if (created == null) { return error(404, "RUN_NOT_FOUND", "运行不存在或已启动: " + runId); } // 校验 pathVariable 与 created 一致（兼容数字字符串） if (!matchesWorkflow(created, workflowName)) { return error(400, "VALIDATION_FAILED", "runId 与 workflowName 不匹配"); } return startRun(created.workflowName(), created.workflowId(), runId, created.inputs()); } private ResponseEntity startRun(String workflowName, Long workflowId, String runId, Map inputs) { workflowEngine.executeForExternal(workflowId, runId, inputs); StartRunResponse body = new StartRunResponse(); body.setRunId(runId); body.setStatus("RUNNING"); body.setStreamUrl("/api/v1/workflows/" + workflowName + "/runs/" + runId + "/stream"); return ResponseEntity.ok(ApiError.ok(body)); } // ============================================================ // GET /workflows/{workflowName}/runs/{runId}/stream — SSE 流 // ============================================================ @GetMapping(value = "/workflows/{workflowName}/runs/{runId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public ResponseEntity streamRun(@PathVariable String workflowName, @PathVariable String runId, @RequestHeader(value = "Last-Event-ID", required = false) Long lastEventId) { if (!eventBus.exists(runId)) { WorkflowRun record = findRunRecord(runId); if (record == null) { return ResponseEntity.notFound().build(); } eventBus.register(runId); eventBus.publish(runId, record.getStatus().equals("FAILED") ? "workflow_error" : "workflow_complete", WorkflowRunEvent.workflowComplete(runId, parseOutputs(record))); } SseEmitter emitter = eventBus.subscribe(runId, lastEventId); if (emitter == null) { return ResponseEntity.notFound().build(); } return ResponseEntity.ok() .header("X-Accel-Buffering", "no") .header("Cache-Control", "no-cache") .body(emitter); } // ============================================================ // GET /workflows/{workflowName}/runs/{runId} — 查询运行结果 // ============================================================ @GetMapping("/workflows/{workflowName}/runs/{runId}") public ResponseEntity getRunResult(@PathVariable String workflowName, @PathVariable String runId) { Workflow wf = resolveWorkflow(workflowName); if (wf == null) { return error(404, "WORKFLOW_NOT_FOUND", "工作流不存在: " + workflowName); } RunResultResponse body = new RunResultResponse(); body.setRunId(runId); body.setWorkflowId(wf.getId()); body.setWorkflowName(wf.getName()); body.setDisplayName(wf.getDisplayName()); SseEventBus.RunStatus live = eventBus.getStatus(runId); if (live != null) { body.setStatus(live.status()); } WorkflowRun record = findRunRecord(runId); if (record == null) { ExternalRunRegistry.CreatedRun created = runRegistry.peek(runId); if (created != null) { body.setStatus("CREATED"); return ResponseEntity.ok(ApiError.ok(body)); } return error(404, "RUN_NOT_FOUND", "运行不存在: " + runId); } body.setStatus(record.getStatus()); if (record.getStartedAt() != null) { body.setStartedAt(record.getStartedAt().toInstant(ZoneOffset.UTC)); } if (record.getCompletedAt() != null) { body.setCompletedAt(record.getCompletedAt().toInstant(ZoneOffset.UTC)); } body.setError(record.getError()); body.setOutputs(parseOutputs(record)); List nodes = workflowRunNodeRepository.findByRunIdOrderBySortOrderAsc(record.getId()); List nodeDtos = nodes.stream().map(n -> { NodeResult dto = new NodeResult(); dto.setNodeId(n.getNodeId()); dto.setNodeType(n.getNodeType()); dto.setLabel(n.getLabel()); dto.setStatus(n.getStatus()); dto.setOutput(parseJsonToMap(n.getOutput())); dto.setError(n.getError()); dto.setLogsCount(countLogs(n.getLogs())); dto.setLogsUrl("/api/v1/workflows/" + wf.getName() + "/runs/" + runId + "/nodes/" + n.getNodeId() + "/logs"); return dto; }).collect(Collectors.toList()); body.setNodes(nodeDtos); Path runDir = runDirManager.getRunDir(wf.getId(), wf.getName(), runId); if (runDir != null) { try { final int[] fileCount = {0}; final long[] totalBytes = {0}; Files.walk(runDir) .filter(p -> !Files.isDirectory(p) && shouldIncludeInZip(runDir, p)) .forEach(p -> { fileCount[0]++; try { totalBytes[0] += Files.size(p); } catch (IOException ignore) {} }); WorkspaceInfo ws = new WorkspaceInfo(); ws.setFileCount(fileCount[0]); ws.setTotalBytes(totalBytes[0]); ws.setDownloadUrl("/api/v1/workflows/" + wf.getName() + "/runs/" + runId + "/workspace"); body.setWorkspace(ws); } catch (IOException e) { log.warn("[ExternalApi] 统计工作目录失败 runId={}: {}", runId, e.getMessage()); } } return ResponseEntity.ok(ApiError.ok(body)); } // ============================================================ // GET /workflows/{workflowName}/runs/{runId}/workspace // ============================================================ @GetMapping("/workflows/{workflowName}/runs/{runId}/workspace") public ResponseEntity downloadWorkspace(@PathVariable String workflowName, @PathVariable String runId) throws IOException { Workflow wf = resolveWorkflow(workflowName); if (wf == null) { return ResponseEntity.notFound().build(); } Path runDir = runDirManager.getRunDir(wf.getId(), wf.getName(), runId); if (runDir == null) { return ResponseEntity.notFound().build(); } Path zipPath = Files.createTempFile("external-run-" + runId, ".zip"); try (ZipOutputStream zos = new ZipOutputStream(Files.newOutputStream(zipPath))) { Files.walk(runDir) .filter(p -> !Files.isDirectory(p) && shouldIncludeInZip(runDir, p)) .forEach(p -> { ZipEntry entry = new ZipEntry(runDir.relativize(p).toString().replace('\\', '/')); try { zos.putNextEntry(entry); Files.copy(p, zos); zos.closeEntry(); } catch (IOException e) { log.warn("[ExternalApi] zip 条目写入失败: {}", p, e); } }); } Resource resource = new UrlResource(zipPath.toUri()); String filename = URLEncoder.encode("workflow-" + wf.getName() + "-run-" + runId + ".zip", StandardCharsets.UTF_8); return ResponseEntity.ok() .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename*=UTF-8''" + filename) .contentType(MediaType.APPLICATION_OCTET_STREAM) .body(resource); } // ============================================================ // GET /workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs // ============================================================ @GetMapping("/workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs") public ResponseEntity getNodeLogs(@PathVariable String workflowName, @PathVariable String runId, @PathVariable String nodeId) { WorkflowRun record = findRunRecord(runId); if (record == null) { return error(404, "RUN_NOT_FOUND", "运行不存在: " + runId); } List nodes = workflowRunNodeRepository.findByRunIdOrderBySortOrderAsc(record.getId()); WorkflowRunNode target = nodes.stream() .filter(n -> nodeId.equals(n.getNodeId())) .findFirst().orElse(null); if (target == null) { return error(404, "NODE_NOT_FOUND", "节点不存在: " + nodeId); } List logs = new ArrayList<>(); if (target.getLogs() != null && !target.getLogs().isBlank()) { try { List raw = MAPPER.readValue(target.getLogs(), new TypeReference>() {}); for (JsonNode n : raw) { LogEntry e = new LogEntry(); e.setType(n.path("type").asText("")); e.setMessage(n.path("message").asText("")); e.setDetail(n.path("detail").isNull() ? null : n.path("detail").asText("")); String ts = n.path("timestamp").asText(""); if (!ts.isEmpty()) { try { e.setTimestamp(Instant.parse(ts)); } catch (Exception ignore) {} } logs.add(e); } } catch (Exception e) { log.warn("[ExternalApi] 解析节点日志失败 nodeId={}: {}", nodeId, e.getMessage()); } } NodeLogsResponse body = new NodeLogsResponse(); body.setRunId(runId); body.setNodeId(nodeId); body.setLogs(logs); return ResponseEntity.ok(ApiError.ok(body)); } // ============================================================ // GET /workflows — 列出可调用工作流 // ============================================================ @GetMapping("/workflows") public ResponseEntity listWorkflows(@RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry) { List all = workflowRepository.findAllByOrderByUpdatedAtDesc(); List accessible = all.stream() .filter(w -> properties.canAccess(keyEntry, w.getName(), w.getId())) .filter(w -> w.getGraphData() != null && !w.getGraphData().isBlank()) .collect(Collectors.toList()); List items = new ArrayList<>(); for (Workflow w : accessible) { WorkflowSummary s = new WorkflowSummary(); s.setId(w.getId()); s.setName(w.getName()); s.setDisplayName(w.getDisplayName()); s.setDescription(w.getDescription()); try { JsonNode root = MAPPER.readTree(w.getGraphData()); s.setInputs(extractFields(root, "userInput")); s.setOutputs(extractFields(root, "output")); } catch (Exception e) { s.setInputs(List.of()); s.setOutputs(List.of()); } items.add(s); } WorkflowListResponse body = new WorkflowListResponse(); body.setTotal(items.size()); body.setItems(items); return ResponseEntity.ok(ApiError.ok(body)); } // ============================================================ // 工具方法 // ============================================================ /** * 解析工作流：优先 findByName，回退 findById（兼容数字 id URL）。 */ private Workflow resolveWorkflow(String nameOrId) { if (nameOrId == null || nameOrId.isEmpty()) return null; return workflowRepository.findByName(nameOrId) .orElseGet(() -> { if (nameOrId.matches("\\d+")) { return workflowRepository.findById(Long.valueOf(nameOrId)).orElse(null); } return null; }); } /** 判断 created run 是否与 pathVariable workflowName 匹配（name 或数字 id 兼容） */ private boolean matchesWorkflow(ExternalRunRegistry.CreatedRun created, String workflowName) { if (created == null || workflowName == null) return false; if (workflowName.equals(created.workflowName())) return true; if (workflowName.matches("\\d+") && Long.valueOf(workflowName).equals(created.workflowId())) return true; return false; } private ResponseEntity error(int httpStatus, String code, String message) { return ResponseEntity.status(httpStatus).body(ApiError.error(code, message)); } private String generateRunId() { return UUID.randomUUID().toString().substring(0, 8); } private CreateRunRequest parsePayload(String payloadJson) { if (payloadJson == null || payloadJson.isBlank()) return new CreateRunRequest(); try { return MAPPER.readValue(payloadJson, CreateRunRequest.class); } catch (Exception e) { log.warn("[ExternalApi] 解析 payload 失败，使用默认值: {}", e.getMessage()); return new CreateRunRequest(); } } private WorkflowRun findRunRecord(String runId) { return workflowRunRepository.findAllByOrderByStartedAtDesc().stream() .filter(r -> runId.equals(r.getRunId())) .findFirst().orElse(null); } @SuppressWarnings("unchecked") private Map parseOutputs(WorkflowRun record) { if (record.getOutputs() == null || record.getOutputs().isBlank()) return Map.of(); try { return MAPPER.readValue(record.getOutputs(), Map.class); } catch (Exception e) { return Map.of(); } } @SuppressWarnings("unchecked") private Map parseJsonToMap(String json) { if (json == null || json.isBlank()) return null; try { return MAPPER.readValue(json, Map.class); } catch (Exception e) { return null; } } private int countLogs(String logsJson) { if (logsJson == null || logsJson.isBlank()) return 0; try { return MAPPER.readTree(logsJson).size(); } catch (Exception e) { return 0; } } private List extractFields(JsonNode root, String nodeType) { List result = new ArrayList<>(); JsonNode nodes = root.path("nodes"); if (!nodes.isArray()) return result; for (JsonNode node : nodes) { if (!nodeType.equals(node.path("type").asText(""))) continue; JsonNode data = node.path("data"); JsonNode varArr = data.path("variables"); if (varArr.isArray()) { for (JsonNode v : varArr) { result.add(toFieldDef(v)); } } JsonNode fieldArr = data.path("fields"); if (fieldArr.isArray()) { for (JsonNode f : fieldArr) { result.add(toFieldDef(f)); } } } return result; } private FieldDef toFieldDef(JsonNode n) { FieldDef f = new FieldDef(); f.setName(n.path("name").asText("")); f.setType(n.path("type").asText("string")); f.setRequired(n.path("required").asBoolean(false)); f.setDescription(n.path("description").asText("")); return f; } private boolean shouldIncludeInZip(Path runDir, Path p) { Path rel = runDir.relativize(p); String relStr = rel.toString().replace('\\', '/'); if (agentIgnoreFilter.shouldIgnore(relStr, false)) return false; Path cur = rel; while (cur != null && cur.getNameCount() > 1) { cur = cur.getParent(); if (cur == null) break; if (agentIgnoreFilter.shouldIgnore(cur.toString().replace('\\', '/'), true)) return false; } return true; } }