From 1159dcdbf30d3ffab39e18fa57fe46d9678cefde Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 19 Aug 2026 05:36:51 -0400 Subject: [PATCH] feat(a2a): add agent interoperability protocol support Add an A2A JSON-RPC endpoint with authenticated message/send, message/stream, tasks/get, and tasks/cancel handling. Expose anonymous minimal Agent Cards while keeping the enabled-agent skills list behind existing Bearer authentication. Bridge inbound calls into the existing agent runtime, add an in-memory task store with duplicate task rejection, JSON-RPC idempotency snapshots, terminal TTL cleanup, and SSE status/artifact event streaming with heartbeat comments. Add the call_a2a_agent tool and peer adapter with Agent Card discovery fallback, blocking task polling through tasks/get, event-boundary SSE parsing, response caps, timeout limits, redirect refusal, and private-network SSRF protection. Wire mateclaw.a2a configuration, document deployment settings in English and Chinese, mirror bundled docs, and cover task storage, JSON-RPC validation, card privacy, lifecycle/cancel behavior, SSE parsing, and outbound guardrails with focused tests. --- .../java/vip/mate/config/SecurityConfig.java | 2 + .../interop/a2a/A2aAgentCardController.java | 32 ++ .../mate/interop/a2a/A2aAgentCardService.java | 93 +++++ .../interop/a2a/A2aAutoConfiguration.java | 9 + .../vip/mate/interop/a2a/A2aCallTool.java | 67 ++++ .../mate/interop/a2a/A2aExecutionBridge.java | 9 + .../mate/interop/a2a/A2aExecutionRequest.java | 12 + .../interop/a2a/A2aJsonRpcController.java | 321 ++++++++++++++++++ .../vip/mate/interop/a2a/A2aPeerAdapter.java | 298 ++++++++++++++++ .../a2a/A2aPeerAdapterConfiguration.java | 21 ++ .../vip/mate/interop/a2a/A2aProperties.java | 27 ++ .../java/vip/mate/interop/a2a/A2aTask.java | 75 ++++ .../vip/mate/interop/a2a/A2aTaskStore.java | 98 ++++++ .../a2a/A2aTaskStoreConfiguration.java | 15 + .../vip/mate/interop/a2a/A2aTaskSweeper.java | 26 ++ .../a2a/DefaultA2aExecutionBridge.java | 32 ++ .../java/vip/mate/interop/a2a/SseFrames.java | 51 +++ .../src/main/resources/application.yml | 13 + .../src/main/resources/docs/en/a2a.md | 39 +++ .../src/main/resources/docs/zh/a2a.md | 39 +++ .../interop/a2a/A2aJsonRpcControllerTest.java | 142 ++++++++ .../mate/interop/a2a/A2aPeerAdapterTest.java | 100 ++++++ .../mate/interop/a2a/A2aTaskStoreTest.java | 81 +++++ .../vip/mate/interop/a2a/SseFramesTest.java | 60 ++++ 24 files changed, 1662 insertions(+) create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardController.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardService.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAutoConfiguration.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aCallTool.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionBridge.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionRequest.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aJsonRpcController.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapter.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapterConfiguration.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aProperties.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTask.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStore.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStoreConfiguration.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskSweeper.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/DefaultA2aExecutionBridge.java create mode 100644 mateclaw-server/src/main/java/vip/mate/interop/a2a/SseFrames.java create mode 100644 mateclaw-server/src/main/resources/docs/en/a2a.md create mode 100644 mateclaw-server/src/main/resources/docs/zh/a2a.md create mode 100644 mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aJsonRpcControllerTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aPeerAdapterTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aTaskStoreTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/interop/a2a/SseFramesTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/config/SecurityConfig.java b/mateclaw-server/src/main/java/vip/mate/config/SecurityConfig.java index 826e7592..cea7dfb2 100644 --- a/mateclaw-server/src/main/java/vip/mate/config/SecurityConfig.java +++ b/mateclaw-server/src/main/java/vip/mate/config/SecurityConfig.java @@ -100,6 +100,8 @@ public class SecurityConfig { // KB Open API: authenticated by KbOpenApiAuthFilter (API key), // not JWT — must be permitAll so the filter is the sole gatekeeper (R1). "/api/v1/open/kb/**", + "/api/a2a/card", + "/.well-known/agent-card.json", "/api/v1/talk/ws", // Desktop local-tool tunnel — the handshake interceptor // authenticates the ?token= query param itself, so the diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardController.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardController.java new file mode 100644 index 00000000..5e17681a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardController.java @@ -0,0 +1,32 @@ +package vip.mate.interop.a2a; + +import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; +import org.springframework.security.core.Authentication; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RestController; + +import java.util.Map; + +@RestController +@RequiredArgsConstructor +public class A2aAgentCardController { + + private final A2aAgentCardService cardService; + + @GetMapping("/api/a2a/card") + public Map card(HttpServletRequest request, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, + Authentication authentication) { + if (authentication == null) { + return cardService.publicCard(request); + } + return cardService.authenticatedCard(request, workspaceId); + } + + @GetMapping("/.well-known/agent-card.json") + public Map wellKnown(HttpServletRequest request) { + return cardService.publicCard(request); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardService.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardService.java new file mode 100644 index 00000000..a959cd6b --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAgentCardService.java @@ -0,0 +1,93 @@ +package vip.mate.interop.a2a; + +import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.support.ServletUriComponentsBuilder; +import vip.mate.agent.AgentService; +import vip.mate.agent.model.AgentEntity; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class A2aAgentCardService { + + private final A2aProperties properties; + private final AgentService agentService; + + public Map publicCard(HttpServletRequest request) { + Map card = baseCard(request); + card.put("supportsAuthenticatedExtendedCard", true); + card.remove("skills"); + return card; + } + + public Map authenticatedCard(HttpServletRequest request, Long workspaceId) { + long wsId = workspaceId == null ? 1L : workspaceId; + Map card = baseCard(request); + List> skills = new ArrayList<>(); + for (AgentEntity agent : agentService.listAgentsByWorkspace(wsId, true)) { + Map skill = new LinkedHashMap<>(); + skill.put("id", String.valueOf(agent.getId())); + skill.put("name", agent.getName()); + skill.put("description", agent.getDescription() == null ? "" : agent.getDescription()); + skill.put("tags", tags(agent.getTags())); + skills.add(skill); + } + card.put("skills", skills); + return card; + } + + private Map baseCard(HttpServletRequest request) { + String rpcUrl = externalBaseUrl(request).replaceAll("/+$", "") + "/api/a2a"; + Map card = new LinkedHashMap<>(); + card.put("name", "MateClaw"); + card.put("description", "A multi-agent runtime exposed through A2A JSON-RPC."); + card.put("url", rpcUrl); + card.put("version", "1.0.0"); + card.put("protocolVersion", "1.0"); + card.put("supportedInterfaces", List.of(Map.of( + "url", rpcUrl, + "protocolBinding", "JSONRPC", + "protocolVersion", "1.0" + ))); + card.put("capabilities", Map.of( + "streaming", true, + "pushNotifications", false, + "stateTransitionHistory", false + )); + card.put("defaultInputModes", List.of("text/plain")); + card.put("defaultOutputModes", List.of("text/plain")); + card.put("skills", List.of()); + return card; + } + + private String externalBaseUrl(HttpServletRequest request) { + if (properties.getBaseUrl() != null && !properties.getBaseUrl().isBlank()) { + return properties.getBaseUrl().trim(); + } + return ServletUriComponentsBuilder.fromRequestUri(request) + .replacePath(null) + .replaceQuery(null) + .build() + .toUriString(); + } + + private static List tags(String tags) { + if (tags == null || tags.isBlank()) { + return List.of(); + } + List out = new ArrayList<>(); + for (String tag : tags.split(",")) { + String trimmed = tag.trim(); + if (!trimmed.isBlank()) { + out.add(trimmed); + } + } + return out; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAutoConfiguration.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAutoConfiguration.java new file mode 100644 index 00000000..fedc909c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aAutoConfiguration.java @@ -0,0 +1,9 @@ +package vip.mate.interop.a2a; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +@Configuration +@EnableConfigurationProperties(A2aProperties.class) +public class A2aAutoConfiguration { +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aCallTool.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aCallTool.java new file mode 100644 index 00000000..68f7b752 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aCallTool.java @@ -0,0 +1,67 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.RequiredArgsConstructor; +import org.springframework.ai.tool.annotation.Tool; +import org.springframework.ai.tool.annotation.ToolParam; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.Map; + +@Component +@RequiredArgsConstructor +public class A2aCallTool { + + private final A2aPeerAdapter peerAdapter; + private final ObjectMapper objectMapper; + + @Tool(name = "call_a2a_agent", description = "Call another A2A-compatible agent. The config JSON must include url and may include headers.") + public String callA2aAgent( + @ToolParam(description = "Message to send to the peer agent") String message, + @ToolParam(description = "Optional peer conversation context id", required = false) String contextId, + @ToolParam(description = "Optional peer skill id", required = false) String skillId, + @ToolParam(description = "JSON object: {\"url\":\"https://peer/api/a2a\",\"headers\":{\"Authorization\":\"Bearer ...\"},\"stream\":false}") String config + ) { + try { + Map cfg = parseConfig(config); + String url = String.valueOf(cfg.getOrDefault("url", "")).trim(); + if (url.isBlank()) { + return "Error: config.url is required."; + } + Map headers = headers(cfg.get("headers")); + boolean stream = Boolean.TRUE.equals(cfg.get("stream")); + A2aPeerAdapter.PeerResult result = stream + ? peerAdapter.stream(url, message, contextId, skillId, headers) + : peerAdapter.sendBlocking(url, message, contextId, skillId, headers); + if (stream && !result.frames().isEmpty()) { + return objectMapper.writeValueAsString(result.frames()); + } + return result.body(); + } catch (Exception e) { + return "Error: " + e.getMessage(); + } + } + + private Map parseConfig(String config) throws Exception { + if (config == null || config.isBlank()) { + return Map.of(); + } + return objectMapper.readValue(config, new TypeReference>() { + }); + } + + private static Map headers(Object value) { + if (!(value instanceof Map raw)) { + return Map.of(); + } + Map out = new LinkedHashMap<>(); + for (Map.Entry entry : raw.entrySet()) { + if (entry.getKey() != null && entry.getValue() != null) { + out.put(String.valueOf(entry.getKey()), String.valueOf(entry.getValue())); + } + } + return out; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionBridge.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionBridge.java new file mode 100644 index 00000000..7f3e8f24 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionBridge.java @@ -0,0 +1,9 @@ +package vip.mate.interop.a2a; + +public interface A2aExecutionBridge { + + ExecutionResult executeBlocking(A2aExecutionRequest request); + + record ExecutionResult(String text, boolean terminal) { + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionRequest.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionRequest.java new file mode 100644 index 00000000..ebcd53fd --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aExecutionRequest.java @@ -0,0 +1,12 @@ +package vip.mate.interop.a2a; + +public record A2aExecutionRequest( + String taskId, + String contextId, + String message, + Long agentId, + Long workspaceId, + String username, + Long userId +) { +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aJsonRpcController.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aJsonRpcController.java new file mode 100644 index 00000000..1ccfa7ac --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aJsonRpcController.java @@ -0,0 +1,321 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.RequiredArgsConstructor; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.security.core.Authentication; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +@RestController +@RequestMapping("/api/a2a") +@RequiredArgsConstructor +public class A2aJsonRpcController { + + private static final int ERR_INVALID_REQUEST = -32600; + private static final int ERR_METHOD_NOT_FOUND = -32601; + private static final int ERR_INVALID_PARAMS = -32602; + private static final int ERR_TASK_NOT_FOUND = -32001; + private static final int ERR_DUPLICATE_TASK = -32009; + + private final ObjectMapper objectMapper; + private final A2aProperties properties; + private final A2aTaskStore store; + private final A2aExecutionBridge bridge; + private final ExecutorService streamExecutor = Executors.newVirtualThreadPerTaskExecutor(); + + @PostMapping + public ResponseEntity handle(@RequestBody JsonNode body, Authentication authentication) { + if (!properties.isEnabled()) { + return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("error", "A2A is disabled")); + } + if (authentication == null) { + return ResponseEntity.status(HttpStatus.UNAUTHORIZED).body(error(null, -32050, "unauthorized")); + } + Object rpcId = rpcId(body == null ? null : body.get("id")); + if (rpcId == InvalidRpcId.INSTANCE) { + return ResponseEntity.ok(error(null, ERR_INVALID_REQUEST, "JSON-RPC id must be a string, number, or null")); + } + if (body == null || !body.isObject() || !"2.0".equals(text(body.get("jsonrpc")))) { + return ResponseEntity.ok(error(rpcId, ERR_INVALID_REQUEST, "invalid JSON-RPC request")); + } + String method = text(body.get("method")); + JsonNode params = body.get("params"); + String tenant = tenant(params); + String rpcKey = rpcId == null ? null : String.valueOf(rpcId); + if (rpcKey != null) { + var existing = store.rpcSnapshot(tenant, rpcKey); + if (existing.isPresent()) { + return ResponseEntity.ok(result(rpcId, existing.get())); + } + } + + try { + return switch (method) { + case "message/send" -> ResponseEntity.ok(handleSend(rpcId, tenant, params, authentication)); + case "message/stream" -> stream(rpcId, tenant, params, authentication); + case "tasks/get" -> ResponseEntity.ok(handleGet(rpcId, tenant, params)); + case "tasks/cancel" -> ResponseEntity.ok(handleCancel(rpcId, tenant, params)); + default -> ResponseEntity.ok(error(rpcId, ERR_METHOD_NOT_FOUND, "method not found")); + }; + } catch (IllegalStateException e) { + return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS) + .body(error(rpcId, -32051, e.getMessage())); + } + } + + private Map handleSend(Object rpcId, String tenant, JsonNode params, Authentication auth) { + try { + A2aExecutionRequest request = executionRequest(tenant, params, auth); + A2aTask submitted = A2aTask.submitted(request.taskId(), request.contextId(), tenant); + if (!store.putIfAbsent(tenant, submitted)) { + return error(rpcId, ERR_DUPLICATE_TASK, "task id already exists"); + } + A2aTask working = store.update(tenant, request.taskId(), + task -> task.withStatus("working", null, false)).orElse(submitted); + boolean blocking = !params.has("configuration") + || !params.get("configuration").has("blocking") + || params.get("configuration").get("blocking").asBoolean(true); + A2aTask responseTask = blocking ? executeWithTimeout(tenant, request, working) : working; + Map snapshot = responseTask.toMap(); + store.rememberRpcSnapshot(tenant, rpcId == null ? null : String.valueOf(rpcId), snapshot); + return result(rpcId, snapshot); + } catch (IllegalArgumentException e) { + return error(rpcId, ERR_INVALID_PARAMS, e.getMessage()); + } + } + + private ResponseEntity stream(Object rpcId, String tenant, JsonNode params, Authentication auth) { + SseEmitter emitter = new SseEmitter(0L); + AtomicBoolean done = new AtomicBoolean(false); + streamExecutor.execute(() -> heartbeat(emitter, done)); + streamExecutor.execute(() -> { + try { + A2aExecutionRequest request = executionRequest(tenant, params, auth); + A2aTask submitted = A2aTask.submitted(request.taskId(), request.contextId(), tenant); + if (!store.putIfAbsent(tenant, submitted)) { + send(emitter, "error", error(rpcId, ERR_DUPLICATE_TASK, "task id already exists")); + emitter.complete(); + return; + } + send(emitter, "task", submitted.toMap()); + A2aTask working = store.update(tenant, request.taskId(), + task -> task.withStatus("working", null, false)).orElse(submitted); + send(emitter, "status-update", working.toMap()); + A2aExecutionBridge.ExecutionResult out = bridge.executeBlocking(request); + A2aTask withArtifact = store.update(tenant, request.taskId(), + task -> task.withArtifact(out.text(), true)).orElse(working); + send(emitter, "artifact-update", withArtifact.artifacts().getLast()); + A2aTask completed = store.update(tenant, request.taskId(), + task -> task.withStatus(out.terminal() ? "completed" : "working", out.text(), out.terminal())) + .orElse(withArtifact); + send(emitter, "status-update", completed.toMap()); + emitter.complete(); + } catch (Exception e) { + try { + send(emitter, "error", error(rpcId, ERR_INVALID_PARAMS, e.getMessage())); + } catch (IOException ignored) { + // The client may already have disconnected. + } + emitter.completeWithError(e); + } finally { + done.set(true); + } + }); + return ResponseEntity.ok() + .contentType(MediaType.TEXT_EVENT_STREAM) + .body(emitter); + } + + private Map handleGet(Object rpcId, String tenant, JsonNode params) { + String taskId = taskId(params); + return store.get(tenant, taskId) + .>map(task -> result(rpcId, task.toMap())) + .orElseGet(() -> error(rpcId, ERR_TASK_NOT_FOUND, "task not found")); + } + + private Map handleCancel(Object rpcId, String tenant, JsonNode params) { + String taskId = taskId(params); + return store.update(tenant, taskId, task -> task.terminal() + ? task + : task.withStatus("canceled", "Task canceled by caller.", true)) + .>map(task -> result(rpcId, task.toMap())) + .orElseGet(() -> error(rpcId, ERR_TASK_NOT_FOUND, "task not found")); + } + + private A2aTask executeWithTimeout(String tenant, A2aExecutionRequest request, A2aTask current) { + CompletableFuture future = + CompletableFuture.supplyAsync(() -> bridge.executeBlocking(request)); + try { + A2aExecutionBridge.ExecutionResult out = future.get(properties.getCallTimeoutMs(), TimeUnit.MILLISECONDS); + A2aTask withArtifact = store.update(tenant, request.taskId(), + task -> task.withArtifact(out.text(), true)).orElse(current); + return store.update(tenant, request.taskId(), + task -> task.withStatus(out.terminal() ? "completed" : "working", out.text(), out.terminal())) + .orElse(withArtifact); + } catch (TimeoutException e) { + return current; + } catch (Exception e) { + return store.update(tenant, request.taskId(), + task -> task.withStatus("failed", e.getMessage(), true)).orElse(current); + } + } + + private A2aExecutionRequest executionRequest(String tenant, JsonNode params, Authentication auth) { + if (params == null || !params.isObject()) { + throw new IllegalArgumentException("params object is required"); + } + JsonNode message = params.get("message"); + if (message == null || !message.isObject()) { + throw new IllegalArgumentException("message object is required"); + } + String taskId = firstText(message.get("taskId"), params.get("id")); + if (taskId.isBlank()) { + taskId = "task-" + UUID.randomUUID(); + } + String contextId = firstText(message.get("contextId"), params.get("contextId")); + if (contextId.isBlank()) { + contextId = taskId; + } + String text = extractText(message.get("parts")); + if (text.isBlank()) { + throw new IllegalArgumentException("message text is required"); + } + Long agentId = agentId(message.get("metadata")); + Long workspaceId = longOrDefault(params.get("workspaceId"), 1L); + Long userId = auth.getDetails() instanceof Number n ? n.longValue() : null; + return new A2aExecutionRequest(taskId, contextId, text, agentId, workspaceId, auth.getName(), userId); + } + + private static Long agentId(JsonNode metadata) { + String skillId = metadata == null ? "" : text(metadata.get("skillId")); + if (skillId.isBlank()) { + throw new IllegalArgumentException("message.metadata.skillId is required"); + } + try { + return Long.parseLong(skillId); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("message.metadata.skillId must be a numeric agent id"); + } + } + + private static String extractText(JsonNode parts) { + if (parts == null || !parts.isArray()) { + return ""; + } + List texts = new ArrayList<>(); + for (JsonNode part : parts) { + String text = text(part.get("text")); + if (!text.isBlank()) { + texts.add(text); + } + } + return String.join("\n", texts); + } + + private static String taskId(JsonNode params) { + String id = firstText(params == null ? null : params.get("id"), + params == null ? null : params.get("taskId")); + if (id.isBlank()) { + throw new IllegalArgumentException("task id is required"); + } + return id; + } + + private static String tenant(JsonNode params) { + return text(params == null ? null : params.get("tenant")); + } + + private static Long longOrDefault(JsonNode node, Long fallback) { + if (node == null || node.isNull()) { + return fallback; + } + if (node.isNumber()) { + return node.longValue(); + } + if (node.isTextual() && !node.asText().isBlank()) { + return Long.parseLong(node.asText()); + } + return fallback; + } + + private static Object rpcId(JsonNode id) { + if (id == null || id.isNull()) { + return null; + } + if (id.isTextual()) { + return id.asText(); + } + if (id.isNumber()) { + return id.numberValue(); + } + return InvalidRpcId.INSTANCE; + } + + private static String firstText(JsonNode first, JsonNode second) { + String value = text(first); + return value.isBlank() ? text(second) : value; + } + + private static String text(JsonNode node) { + return node == null || node.isNull() ? "" : node.asText(""); + } + + private static Map result(Object id, Object result) { + Map out = new LinkedHashMap<>(); + out.put("jsonrpc", "2.0"); + out.put("id", id); + out.put("result", result); + return out; + } + + private static Map error(Object id, int code, String message) { + Map out = new LinkedHashMap<>(); + out.put("jsonrpc", "2.0"); + out.put("id", id); + out.put("error", Map.of("code", code, "message", message == null ? "" : message)); + return out; + } + + private void send(SseEmitter emitter, String event, Object data) throws IOException { + emitter.send(SseEmitter.event() + .name(event) + .data(objectMapper.writeValueAsString(data))); + } + + private void heartbeat(SseEmitter emitter, AtomicBoolean done) { + while (!done.get()) { + try { + Thread.sleep(15_000L); + if (!done.get()) { + emitter.send(SseEmitter.event().comment("heartbeat")); + } + } catch (Exception e) { + done.set(true); + } + } + } + + private enum InvalidRpcId { + INSTANCE + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapter.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapter.java new file mode 100644 index 00000000..37b35227 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapter.java @@ -0,0 +1,298 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.JsonNode; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +public class A2aPeerAdapter { + + private final ObjectMapper objectMapper; + private final Policy policy; + private final HttpClient httpClient; + + public A2aPeerAdapter(ObjectMapper objectMapper, Policy policy) { + this.objectMapper = objectMapper; + this.policy = policy == null ? Policy.defaults() : policy; + this.httpClient = HttpClient.newBuilder() + .connectTimeout(this.policy.timeout()) + .followRedirects(HttpClient.Redirect.NEVER) + .build(); + } + + public PeerResult sendBlocking(String url, String message, String contextId, String skillId, + Map headers) throws IOException, InterruptedException { + URI uri = resolveRpcUri(url, headers == null ? Map.of() : headers); + Map body = rpcBody("message/send", message, contextId, skillId); + CappedBody response = post(uri, body, headers == null ? Map.of() : headers); + CappedBody finalBody = pollIfRunning(uri, response, headers == null ? Map.of() : headers); + return new PeerResult(finalBody.body(), List.of(), response.truncated() || finalBody.truncated()); + } + + public PeerResult stream(String url, String message, String contextId, String skillId, + Map headers) throws IOException, InterruptedException { + URI uri = resolveRpcUri(url, headers == null ? Map.of() : headers); + Map body = rpcBody("message/stream", message, contextId, skillId); + CappedBody response = post(uri, body, headers == null ? Map.of() : headers); + return new PeerResult(response.body(), SseFrames.parse(response.body()), response.truncated()); + } + + private CappedBody pollIfRunning(URI rpcUri, CappedBody initial, Map headers) + throws IOException, InterruptedException { + JsonNode task = taskNode(initial.body()); + String taskId = task == null ? "" : text(task.get("id")); + if (taskId.isBlank() || isTerminalState(task)) { + return initial; + } + long deadline = System.nanoTime() + policy.timeout().toNanos(); + CappedBody latest = initial; + while (System.nanoTime() < deadline) { + Thread.sleep(Math.min(1_000L, Math.max(100L, policy.timeout().toMillis()))); + latest = post(rpcUri, taskGetBody(taskId), headers); + task = taskNode(latest.body()); + if (task == null || isTerminalState(task)) { + return latest; + } + } + return latest; + } + + private Map taskGetBody(String taskId) { + return Map.of( + "jsonrpc", "2.0", + "id", "rpc-" + UUID.randomUUID(), + "method", "tasks/get", + "params", Map.of("id", taskId) + ); + } + + private JsonNode taskNode(String body) { + try { + JsonNode root = objectMapper.readTree(body); + JsonNode result = root.get("result"); + if (result == null || result.isNull()) { + return null; + } + if (result.has("task")) { + return result.get("task"); + } + return result; + } catch (Exception e) { + return null; + } + } + + private static boolean isTerminalState(JsonNode task) { + JsonNode status = task.get("status"); + String state = status == null ? "" : text(status.get("state")); + return "completed".equalsIgnoreCase(state) + || "canceled".equalsIgnoreCase(state) + || "cancelled".equalsIgnoreCase(state) + || "failed".equalsIgnoreCase(state) + || "TASK_STATE_COMPLETED".equals(state) + || "TASK_STATE_CANCELED".equals(state) + || "TASK_STATE_FAILED".equals(state); + } + + private URI resolveRpcUri(String endpoint, Map headers) { + URI configured = safeUri(endpoint); + for (URI candidate : cardCandidates(configured)) { + try { + CappedBody card = get(candidate, headers); + String url = rpcUrlFromCard(card.body()); + if (!url.isBlank()) { + return safeUri(url); + } + } catch (Exception ignored) { + // Discovery is best-effort; the configured endpoint remains valid. + } + } + return configured; + } + + private List cardCandidates(URI endpoint) { + List out = new ArrayList<>(); + String raw = endpoint.toString(); + if (raw.endsWith(".json")) { + out.add(endpoint); + } + out.add(endpoint.resolve(trimTrailingSlash(endpoint.getPath()) + "/card")); + out.add(endpoint.resolve("/.well-known/agent-card.json")); + return out; + } + + private String rpcUrlFromCard(String body) throws IOException { + JsonNode root = objectMapper.readTree(body); + JsonNode interfaces = root.get("supportedInterfaces"); + if (interfaces != null && interfaces.isArray()) { + for (JsonNode iface : interfaces) { + if ("JSONRPC".equalsIgnoreCase(text(iface.get("protocolBinding")))) { + String url = text(iface.get("url")); + if (!url.isBlank()) { + return url; + } + } + } + } + return text(root.get("url")); + } + + private CappedBody get(URI uri, Map headers) throws IOException, InterruptedException { + HttpRequest.Builder builder = HttpRequest.newBuilder(safeUri(uri.toString())) + .timeout(policy.timeout()) + .GET(); + for (Map.Entry header : headers.entrySet()) { + if (header.getKey() != null && header.getValue() != null) { + builder.header(header.getKey(), header.getValue()); + } + } + HttpResponse response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofByteArray()); + if (response.statusCode() >= 300 && response.statusCode() < 400) { + throw new IOException("redirects are not allowed"); + } + if (response.statusCode() < 200 || response.statusCode() >= 300) { + throw new IOException("peer returned HTTP " + response.statusCode()); + } + return cap(response.body()); + } + + private Map rpcBody(String method, String message, String contextId, String skillId) { + Map metadata = new LinkedHashMap<>(); + if (skillId != null && !skillId.isBlank()) { + metadata.put("skillId", skillId); + } + Map msg = new LinkedHashMap<>(); + msg.put("messageId", "msg-" + UUID.randomUUID()); + if (contextId != null && !contextId.isBlank()) { + msg.put("contextId", contextId); + } + msg.put("parts", List.of(Map.of("kind", "text", "text", message == null ? "" : message))); + msg.put("metadata", metadata); + return Map.of( + "jsonrpc", "2.0", + "id", "rpc-" + UUID.randomUUID(), + "method", method, + "params", Map.of("message", msg, "configuration", Map.of("blocking", true)) + ); + } + + private CappedBody post(URI uri, Map body, Map headers) + throws IOException, InterruptedException { + HttpRequest.Builder builder = HttpRequest.newBuilder(uri) + .timeout(policy.timeout()) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(body))); + for (Map.Entry header : headers.entrySet()) { + if (header.getKey() != null && header.getValue() != null) { + builder.header(header.getKey(), header.getValue()); + } + } + HttpResponse response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofByteArray()); + if (response.statusCode() >= 300 && response.statusCode() < 400) { + throw new IOException("redirects are not allowed"); + } + if (response.statusCode() < 200 || response.statusCode() >= 300) { + throw new IOException("peer returned HTTP " + response.statusCode()); + } + return cap(response.body()); + } + + private CappedBody cap(byte[] body) { + byte[] bytes = body == null ? new byte[0] : body; + boolean truncated = bytes.length > policy.maxResponseBytes(); + int length = truncated ? policy.maxResponseBytes() : bytes.length; + return new CappedBody(new String(bytes, 0, length, StandardCharsets.UTF_8), truncated); + } + + private URI safeUri(String url) { + URI uri = URI.create(url); + String scheme = uri.getScheme(); + if (!"http".equalsIgnoreCase(scheme) && !"https".equalsIgnoreCase(scheme)) { + throw new IllegalArgumentException("Only HTTP and HTTPS A2A URLs are supported"); + } + if (uri.getHost() == null || uri.getHost().isBlank()) { + throw new IllegalArgumentException("A2A URL host is required"); + } + if (!policy.allowPrivateNetwork()) { + rejectPrivateAddress(uri.getHost()); + } + return uri; + } + + private static void rejectPrivateAddress(String host) { + try { + for (InetAddress address : InetAddress.getAllByName(host)) { + byte[] raw = address.getAddress(); + if (address.isAnyLocalAddress() + || address.isLoopbackAddress() + || address.isLinkLocalAddress() + || address.isSiteLocalAddress() + || isReserved(raw)) { + throw new IllegalArgumentException("A2A URL resolves to a private or reserved address"); + } + } + } catch (IOException e) { + throw new IllegalArgumentException("A2A URL host could not be resolved", e); + } + } + + private static boolean isReserved(byte[] raw) { + if (raw.length == 4) { + int first = raw[0] & 0xff; + int second = raw[1] & 0xff; + return first == 0 + || first == 10 + || first == 127 + || first == 169 && second == 254 + || first == 172 && second >= 16 && second <= 31 + || first == 192 && second == 168 + || first >= 224; + } + if (raw.length == 16) { + int first = raw[0] & 0xff; + return first == 0 + || first == 0xfc + || first == 0xfd + || first == 0xfe; + } + return true; + } + + private static String trimTrailingSlash(String path) { + if (path == null || path.isBlank()) { + return ""; + } + return path.replaceAll("/+$", ""); + } + + private static String text(JsonNode node) { + return node == null || node.isNull() ? "" : node.asText(""); + } + + public record Policy(Duration timeout, int maxResponseBytes, boolean allowPrivateNetwork) { + public static Policy defaults() { + return new Policy(Duration.ofSeconds(120), 1_048_576, false); + } + } + + public record PeerResult(String body, List frames, boolean truncated) { + public PeerResult { + frames = frames == null ? List.of() : List.copyOf(frames); + } + } + + private record CappedBody(String body, boolean truncated) { + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapterConfiguration.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapterConfiguration.java new file mode 100644 index 00000000..ad310ebd --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aPeerAdapterConfiguration.java @@ -0,0 +1,21 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.time.Duration; + +@Configuration +public class A2aPeerAdapterConfiguration { + + @Bean + public A2aPeerAdapter a2aPeerAdapter(ObjectMapper objectMapper, A2aProperties properties) { + A2aPeerAdapter.Policy policy = new A2aPeerAdapter.Policy( + Duration.ofMillis(properties.getOutboundTimeoutMs()), + properties.getMaxResponseBytes(), + properties.isAllowPrivateOutbound() + ); + return new A2aPeerAdapter(objectMapper, policy); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aProperties.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aProperties.java new file mode 100644 index 00000000..4f4c3e04 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aProperties.java @@ -0,0 +1,27 @@ +package vip.mate.interop.a2a; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@Data +@ConfigurationProperties(prefix = "mateclaw.a2a") +public class A2aProperties { + + private boolean enabled = false; + + private String baseUrl; + + private long callTimeoutMs = 120_000L; + + private int maxTasks = 1_000; + + private long taskTtlSeconds = 3_600L; + + private int maxResponseBytes = 1_048_576; + + private long outboundTimeoutMs = 120_000L; + + private boolean allowPrivateOutbound = false; + + private long sweepIntervalMs = 60_000L; +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTask.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTask.java new file mode 100644 index 00000000..1a61e5d7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTask.java @@ -0,0 +1,75 @@ +package vip.mate.interop.a2a; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public record A2aTask( + String id, + String contextId, + String tenant, + String state, + String message, + List> artifacts, + boolean terminal, + Instant createdAt, + Instant updatedAt +) { + + private static final List TERMINAL_STATES = List.of("completed", "canceled", "failed"); + + public A2aTask { + artifacts = artifacts == null ? List.of() : List.copyOf(artifacts); + createdAt = createdAt == null ? Instant.now() : createdAt; + updatedAt = updatedAt == null ? createdAt : updatedAt; + } + + public static A2aTask submitted(String id, String contextId, String tenant) { + Instant now = Instant.now(); + return new A2aTask(id, contextId, tenant, "submitted", null, List.of(), false, now, now); + } + + public A2aTask withStatus(String state, String message, boolean terminal) { + boolean finalState = terminal || TERMINAL_STATES.contains(state); + return new A2aTask(id, contextId, tenant, state, message, artifacts, finalState, createdAt, Instant.now()); + } + + public A2aTask withUpdatedAt(Instant updatedAt) { + return new A2aTask(id, contextId, tenant, state, message, artifacts, terminal, createdAt, updatedAt); + } + + public A2aTask withArtifact(String text, boolean append) { + Map artifact = Map.of( + "artifactId", "artifact-" + (artifacts.size() + 1), + "parts", List.of(Map.of("kind", "text", "text", text != null ? text : "")) + ); + List> next = append + ? new ArrayList<>(artifacts) + : new ArrayList<>(); + next.add(artifact); + return new A2aTask(id, contextId, tenant, state, message, next, terminal, createdAt, Instant.now()); + } + + public Map toMap() { + return Map.of( + "id", id, + "contextId", contextId, + "status", Map.of( + "state", state, + "message", messageAsA2aMessage(message) + ), + "artifacts", artifacts + ); + } + + private static Object messageAsA2aMessage(String text) { + if (text == null || text.isBlank()) { + return Map.of(); + } + return Map.of( + "role", "agent", + "parts", List.of(Map.of("kind", "text", "text", text)) + ); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStore.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStore.java new file mode 100644 index 00000000..d2236a33 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStore.java @@ -0,0 +1,98 @@ +package vip.mate.interop.a2a; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Iterator; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.UnaryOperator; + +public class A2aTaskStore { + + private final int maxTasks; + private final Duration ttl; + private final Clock clock; + private final ConcurrentMap tasks = new ConcurrentHashMap<>(); + private final ConcurrentMap> rpcSnapshots = new ConcurrentHashMap<>(); + + public A2aTaskStore(int maxTasks, Duration ttl) { + this(maxTasks, ttl, Clock.systemUTC()); + } + + public A2aTaskStore(int maxTasks, Duration ttl, Clock clock) { + this.maxTasks = Math.max(1, maxTasks); + this.ttl = ttl == null ? Duration.ofHours(1) : ttl; + this.clock = clock == null ? Clock.systemUTC() : clock; + } + + public boolean putIfAbsent(String tenant, A2aTask task) { + if (task == null || task.id() == null || task.id().isBlank()) { + return false; + } + if (tasks.size() >= maxTasks) { + sweepExpired(); + if (tasks.size() >= maxTasks) { + throw new IllegalStateException("too many A2A tasks"); + } + } + return tasks.putIfAbsent(taskKey(tenant, task.id()), task.withUpdatedAt(clock.instant())) == null; + } + + public Optional get(String tenant, String taskId) { + if (taskId == null || taskId.isBlank()) { + return Optional.empty(); + } + return Optional.ofNullable(tasks.get(taskKey(tenant, taskId))); + } + + public Optional update(String tenant, String taskId, UnaryOperator updater) { + String key = taskKey(tenant, taskId); + A2aTask updated = tasks.computeIfPresent(key, + (ignored, current) -> updater.apply(current).withUpdatedAt(clock.instant())); + return Optional.ofNullable(updated); + } + + public boolean rememberRpcSnapshot(String tenant, String rpcId, Map snapshot) { + if (rpcId == null || snapshot == null) { + return true; + } + return rpcSnapshots.putIfAbsent(rpcKey(tenant, rpcId), Map.copyOf(snapshot)) == null; + } + + public Optional> rpcSnapshot(String tenant, String rpcId) { + if (rpcId == null) { + return Optional.empty(); + } + return Optional.ofNullable(rpcSnapshots.get(rpcKey(tenant, rpcId))); + } + + public int sweepExpired() { + Instant cutoff = clock.instant().minus(ttl); + int removed = 0; + Iterator> it = tasks.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry entry = it.next(); + A2aTask task = entry.getValue(); + if (task.terminal() && task.updatedAt().isBefore(cutoff)) { + it.remove(); + removed++; + } + } + return removed; + } + + private static String taskKey(String tenant, String taskId) { + return normalizeTenant(tenant) + "|" + taskId; + } + + private static String rpcKey(String tenant, String rpcId) { + return normalizeTenant(tenant) + "|rpc|" + rpcId; + } + + private static String normalizeTenant(String tenant) { + return tenant == null || tenant.isBlank() ? "default" : tenant; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStoreConfiguration.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStoreConfiguration.java new file mode 100644 index 00000000..d97ed6fd --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskStoreConfiguration.java @@ -0,0 +1,15 @@ +package vip.mate.interop.a2a; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.time.Duration; + +@Configuration +public class A2aTaskStoreConfiguration { + + @Bean + public A2aTaskStore a2aTaskStore(A2aProperties properties) { + return new A2aTaskStore(properties.getMaxTasks(), Duration.ofSeconds(properties.getTaskTtlSeconds())); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskSweeper.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskSweeper.java new file mode 100644 index 00000000..8eb635e6 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/A2aTaskSweeper.java @@ -0,0 +1,26 @@ +package vip.mate.interop.a2a; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class A2aTaskSweeper { + + private final A2aProperties properties; + private final A2aTaskStore store; + + @Scheduled(fixedDelayString = "${mateclaw.a2a.sweep-interval-ms:60000}") + public void sweep() { + if (!properties.isEnabled()) { + return; + } + int removed = store.sweepExpired(); + if (removed > 0) { + log.debug("A2A task sweep removed {} expired task(s)", removed); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/DefaultA2aExecutionBridge.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/DefaultA2aExecutionBridge.java new file mode 100644 index 00000000..35c7c299 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/DefaultA2aExecutionBridge.java @@ -0,0 +1,32 @@ +package vip.mate.interop.a2a; + +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import vip.mate.agent.AgentService; +import vip.mate.agent.context.ChatOrigin; + +@Service +@RequiredArgsConstructor +public class DefaultA2aExecutionBridge implements A2aExecutionBridge { + + private final AgentService agentService; + + @Override + public ExecutionResult executeBlocking(A2aExecutionRequest request) { + ChatOrigin origin = ChatOrigin.web( + request.contextId(), + request.username(), + request.workspaceId(), + null, + null, + request.userId() + ); + AgentService.ChatResult result = agentService.chatWithUsage( + request.agentId(), + request.message(), + request.contextId(), + origin + ); + return new ExecutionResult(result.content(), true); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/interop/a2a/SseFrames.java b/mateclaw-server/src/main/java/vip/mate/interop/a2a/SseFrames.java new file mode 100644 index 00000000..4985c3b6 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/interop/a2a/SseFrames.java @@ -0,0 +1,51 @@ +package vip.mate.interop.a2a; + +import java.util.ArrayList; +import java.util.List; + +public final class SseFrames { + + private SseFrames() { + } + + public record Frame(String event, String data) { + } + + public static List parse(String input) { + List frames = new ArrayList<>(); + if (input == null || input.isEmpty()) { + return frames; + } + String event = "message"; + List dataLines = new ArrayList<>(); + String[] lines = input.split("\\R", -1); + for (String line : lines) { + if (line.isEmpty()) { + flush(frames, event, dataLines); + event = "message"; + dataLines = new ArrayList<>(); + continue; + } + if (line.startsWith(":")) { + continue; + } + if (line.startsWith("event:")) { + event = line.substring("event:".length()).trim(); + continue; + } + if (line.startsWith("data:")) { + String value = line.substring("data:".length()); + dataLines.add(value.startsWith(" ") ? value.substring(1) : value); + } + } + flush(frames, event, dataLines); + return frames; + } + + private static void flush(List frames, String event, List dataLines) { + if (!dataLines.isEmpty()) { + frames.add(new Frame(event == null || event.isBlank() ? "message" : event, + String.join("\n", dataLines))); + } + } +} diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index 504a6604..ea168761 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -131,6 +131,19 @@ springdoc: # MateClaw 自定义配置 mateclaw: + a2a: + enabled: ${MATECLAW_A2A_ENABLED:false} + # Public base URL for Agent Cards. Production deployments should set this + # explicitly so peers do not depend on proxy-derived request headers. + base-url: ${MATECLAW_A2A_BASE_URL:} + call-timeout-ms: ${MATECLAW_A2A_CALL_TIMEOUT_MS:120000} + max-tasks: ${MATECLAW_A2A_MAX_TASKS:1000} + task-ttl-seconds: ${MATECLAW_A2A_TASK_TTL_SECONDS:3600} + max-response-bytes: ${MATECLAW_A2A_MAX_RESPONSE_BYTES:1048576} + outbound-timeout-ms: ${MATECLAW_A2A_OUTBOUND_TIMEOUT_MS:120000} + allow-private-outbound: ${MATECLAW_A2A_ALLOW_PRIVATE_OUTBOUND:false} + sweep-interval-ms: ${MATECLAW_A2A_SWEEP_INTERVAL_MS:60000} + # DeepSeek Harness runtime. The executable and Cordis composition are kept # outside the Spring classpath and can be supplied by environment variables. agent: diff --git a/mateclaw-server/src/main/resources/docs/en/a2a.md b/mateclaw-server/src/main/resources/docs/en/a2a.md new file mode 100644 index 00000000..b3a32c0d --- /dev/null +++ b/mateclaw-server/src/main/resources/docs/en/a2a.md @@ -0,0 +1,39 @@ +# A2A Protocol + +MateClaw can expose enabled agents through an A2A JSON-RPC endpoint and can call other A2A peers from agent tools. + +## Configuration + +```yaml +mateclaw: + a2a: + enabled: true + base-url: https://your-public-host + call-timeout-ms: 120000 + max-tasks: 1000 + task-ttl-seconds: 3600 +``` + +`base-url` is required for production deployments. If it is empty, MateClaw derives the Agent Card URL from the incoming request, which depends on proxy headers being correct. + +## Inbound + +- `GET /.well-known/agent-card.json` and anonymous `GET /api/a2a/card` return the public minimal card without `skills`. +- Authenticated `GET /api/a2a/card` returns enabled agents in `skills[]`; use the agent id as `message.metadata.skillId`. +- `POST /api/a2a` requires an existing MateClaw Bearer token and supports `message/send`, `message/stream`, `tasks/get`, and `tasks/cancel`. + +## Outbound Tool + +Agents can call `call_a2a_agent` with: + +```json +{ + "url": "https://peer.example.com/api/a2a", + "headers": { + "Authorization": "Bearer token" + }, + "stream": false +} +``` + +Outbound calls reject private or reserved network targets by default, do not follow redirects, cap response size, and poll `tasks/get` when a blocking send returns an in-progress task. diff --git a/mateclaw-server/src/main/resources/docs/zh/a2a.md b/mateclaw-server/src/main/resources/docs/zh/a2a.md new file mode 100644 index 00000000..8c8e5a37 --- /dev/null +++ b/mateclaw-server/src/main/resources/docs/zh/a2a.md @@ -0,0 +1,39 @@ +# A2A 协议 + +MateClaw 可以把已启用智能体暴露为 A2A JSON-RPC 端点，也可以通过工具调用其他 A2A 对端。 + +## 配置 + +```yaml +mateclaw: + a2a: + enabled: true + base-url: https://your-public-host + call-timeout-ms: 120000 + max-tasks: 1000 + task-ttl-seconds: 3600 +``` + +生产环境必须配置 `base-url`。为空时，MateClaw 会按入站请求推导名片 URL，这依赖反向代理正确传递 Host 与协议头。 + +## 入站 + +- `GET /.well-known/agent-card.json` 和匿名 `GET /api/a2a/card` 返回最小公开名片，不包含 `skills`。 +- 带 Bearer 访问 `GET /api/a2a/card` 返回 enabled 智能体列表到 `skills[]`；调用时把智能体 id 放到 `message.metadata.skillId`。 +- `POST /api/a2a` 复用 MateClaw Bearer token 鉴权，支持 `message/send`、`message/stream`、`tasks/get`、`tasks/cancel`。 + +## 出站工具 + +智能体可调用 `call_a2a_agent`： + +```json +{ + "url": "https://peer.example.com/api/a2a", + "headers": { + "Authorization": "Bearer token" + }, + "stream": false +} +``` + +出站默认拒绝内网和保留地址，不跟随重定向，限制响应大小；阻塞发送返回在途任务时，会继续轮询 `tasks/get`。 diff --git a/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aJsonRpcControllerTest.java b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aJsonRpcControllerTest.java new file mode 100644 index 00000000..8a83efdf --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aJsonRpcControllerTest.java @@ -0,0 +1,142 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.util.List; +import java.util.Map; +import java.time.Duration; + +import static org.hamcrest.Matchers.not; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +class A2aJsonRpcControllerTest { + + private MockMvc mvc; + private A2aExecutionBridge bridge; + + @BeforeEach + void setUp() { + ObjectMapper objectMapper = new ObjectMapper(); + bridge = mock(A2aExecutionBridge.class); + A2aTaskStore store = new A2aTaskStore(100, Duration.ofMinutes(5)); + A2aProperties properties = new A2aProperties(); + properties.setEnabled(true); + properties.setCallTimeoutMs(1000); + A2aJsonRpcController rpc = new A2aJsonRpcController(objectMapper, properties, store, bridge); + A2aAgentCardService cardService = mock(A2aAgentCardService.class); + when(cardService.publicCard(any())).thenReturn(Map.of( + "name", "MateClaw", + "supportsAuthenticatedExtendedCard", true)); + when(cardService.authenticatedCard(any(), any())).thenReturn(Map.of( + "name", "MateClaw", + "skills", List.of(Map.of("id", "agent-1", "name", "Agent")))); + A2aAgentCardController card = new A2aAgentCardController(cardService); + mvc = MockMvcBuilders.standaloneSetup(rpc, card).build(); + } + + @Test + void rejectsBooleanJsonRpcId() throws Exception { + mvc.perform(post("/api/a2a") + .contentType("application/json") + .principal(auth()) + .content(""" + {"jsonrpc":"2.0","id":true,"method":"tasks/get","params":{"id":"task-1"}} + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.error.code").value(-32600)); + } + + @Test + void anonymousCardOmitsSkillsAndAdvertisesExtendedCard() throws Exception { + mvc.perform(get("/api/a2a/card")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.supportsAuthenticatedExtendedCard").value(true)) + .andExpect(jsonPath("$.skills").doesNotExist()); + } + + @Test + void authenticatedCardIncludesSkills() throws Exception { + mvc.perform(get("/api/a2a/card").principal(auth())) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.skills[0].id").value("agent-1")); + } + + @Test + void sendCreatesTaskAndDuplicateRpcIdReturnsSameSnapshot() throws Exception { + when(bridge.executeBlocking(any())).thenReturn(new A2aExecutionBridge.ExecutionResult("hello", true)); + String body = """ + {"jsonrpc":"2.0","id":"rpc-1","method":"message/send","params":{ + "message":{"messageId":"m1","taskId":"task-1","parts":[{"kind":"text","text":"hi"}], + "metadata":{"skillId":"1"}}, + "configuration":{"blocking":true} + }} + """; + + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()).content(body)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.result.id").value("task-1")) + .andExpect(jsonPath("$.result.status.state").value("completed")); + + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()).content(body)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.result.id").value("task-1")) + .andExpect(jsonPath("$.result.status.state").value("completed")); + } + + @Test + void duplicateTaskIdWithDifferentRpcIdIsRejected() throws Exception { + when(bridge.executeBlocking(any())).thenReturn(new A2aExecutionBridge.ExecutionResult("hello", true)); + String first = """ + {"jsonrpc":"2.0","id":"rpc-1","method":"message/send","params":{ + "message":{"messageId":"m1","taskId":"task-1","parts":[{"kind":"text","text":"hi"}], + "metadata":{"skillId":"1"}} + }} + """; + String second = first.replace("\"rpc-1\"", "\"rpc-2\""); + + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()).content(first)) + .andExpect(status().isOk()); + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()).content(second)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.error.code").value(-32009)); + } + + @Test + void cancelTransitionsActiveTaskToCanceled() throws Exception { + when(bridge.executeBlocking(any())).thenReturn(new A2aExecutionBridge.ExecutionResult("hello", false)); + String send = """ + {"jsonrpc":"2.0","id":"rpc-1","method":"message/send","params":{ + "message":{"messageId":"m1","taskId":"task-1","parts":[{"kind":"text","text":"hi"}], + "metadata":{"skillId":"1"}}, + "configuration":{"blocking":true} + }} + """; + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()).content(send)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.result.status.state", not("completed"))); + + mvc.perform(post("/api/a2a").contentType("application/json").principal(auth()) + .content(""" + {"jsonrpc":"2.0","id":"rpc-2","method":"tasks/cancel","params":{"id":"task-1"}} + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.result.status.state").value("canceled")); + } + + private static UsernamePasswordAuthenticationToken auth() { + UsernamePasswordAuthenticationToken token = new UsernamePasswordAuthenticationToken("alice", null, List.of()); + token.setDetails(1L); + return token; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aPeerAdapterTest.java b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aPeerAdapterTest.java new file mode 100644 index 00000000..559dafcd --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aPeerAdapterTest.java @@ -0,0 +1,100 @@ +package vip.mate.interop.a2a; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class A2aPeerAdapterTest { + + private HttpServer server; + + @AfterEach + void tearDown() { + if (server != null) { + server.stop(0); + } + } + + @Test + void blocksPrivateNetworkTargetsByDefault() { + A2aPeerAdapter adapter = new A2aPeerAdapter(new ObjectMapper(), A2aPeerAdapter.Policy.defaults()); + + assertThrows(IllegalArgumentException.class, () -> + adapter.sendBlocking("http://127.0.0.1:8642/api/a2a", "hi", null, null, Map.of())); + } + + @Test + void refusesRedirects() throws Exception { + int port = startServer(exchange -> { + exchange.getResponseHeaders().add("Location", "https://example.com/api/a2a"); + exchange.sendResponseHeaders(302, -1); + exchange.close(); + }); + A2aPeerAdapter adapter = new A2aPeerAdapter(new ObjectMapper(), + new A2aPeerAdapter.Policy(Duration.ofSeconds(2), 1024 * 1024, true)); + + assertThrows(IOException.class, () -> + adapter.sendBlocking("http://127.0.0.1:" + port + "/api/a2a", "hi", null, null, Map.of())); + } + + @Test + void parsesSseFramesUsingEventBoundaries() throws Exception { + int port = startServer(exchange -> { + byte[] body = """ + event: artifact-update + data: hello + data: world + + """.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + A2aPeerAdapter adapter = new A2aPeerAdapter(new ObjectMapper(), + new A2aPeerAdapter.Policy(Duration.ofSeconds(2), 1024 * 1024, true)); + + A2aPeerAdapter.PeerResult result = adapter.stream( + "http://127.0.0.1:" + port + "/api/a2a", "hi", null, null, Map.of()); + + assertEquals(1, result.frames().size()); + assertEquals("hello\nworld", result.frames().getFirst().data()); + } + + @Test + void truncatesOversizedResponses() throws Exception { + int port = startServer(exchange -> { + byte[] body = "0123456789".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + A2aPeerAdapter adapter = new A2aPeerAdapter(new ObjectMapper(), + new A2aPeerAdapter.Policy(Duration.ofSeconds(2), 5, true)); + + A2aPeerAdapter.PeerResult result = adapter.sendBlocking( + "http://127.0.0.1:" + port + "/api/a2a", "hi", null, null, Map.of()); + + assertTrue(result.truncated()); + assertEquals(5, result.body().length()); + } + + private int startServer(HttpHandler handler) throws Exception { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/api/a2a", handler); + server.start(); + return server.getAddress().getPort(); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aTaskStoreTest.java b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aTaskStoreTest.java new file mode 100644 index 00000000..3a3ea788 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/interop/a2a/A2aTaskStoreTest.java @@ -0,0 +1,81 @@ +package vip.mate.interop.a2a; + +import org.junit.jupiter.api.Test; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class A2aTaskStoreTest { + + @Test + void putIfAbsentRejectsDuplicateTaskIdInSameTenant() { + MutableClock clock = new MutableClock(); + A2aTaskStore store = new A2aTaskStore(10, Duration.ofMinutes(5), clock); + A2aTask first = A2aTask.submitted("task-1", "ctx-1", "tenant-a"); + A2aTask duplicate = A2aTask.submitted("task-1", "ctx-2", "tenant-a"); + + assertTrue(store.putIfAbsent("tenant-a", first)); + assertFalse(store.putIfAbsent("tenant-a", duplicate)); + + assertEquals("ctx-1", store.get("tenant-a", "task-1").orElseThrow().contextId()); + } + + @Test + void duplicateJsonRpcIdReturnsStoredSnapshot() { + MutableClock clock = new MutableClock(); + A2aTaskStore store = new A2aTaskStore(10, Duration.ofMinutes(5), clock); + Map snapshot = Map.of("taskId", "task-1", "status", "submitted"); + + assertTrue(store.rememberRpcSnapshot("tenant-a", "rpc-1", snapshot)); + assertFalse(store.rememberRpcSnapshot("tenant-a", "rpc-1", Map.of("taskId", "other"))); + + assertEquals(snapshot, store.rpcSnapshot("tenant-a", "rpc-1").orElseThrow()); + } + + @Test + void sweepRemovesExpiredTerminalTasksAndKeepsActiveTasks() { + MutableClock clock = new MutableClock(); + A2aTaskStore store = new A2aTaskStore(10, Duration.ofSeconds(30), clock); + A2aTask done = A2aTask.submitted("done", "ctx", "tenant").withStatus("completed", "ok", true); + A2aTask active = A2aTask.submitted("active", "ctx", "tenant").withStatus("working", null, false); + + assertTrue(store.putIfAbsent("tenant", done)); + assertTrue(store.putIfAbsent("tenant", active)); + clock.advance(Duration.ofSeconds(31)); + + assertEquals(1, store.sweepExpired()); + assertTrue(store.get("tenant", "done").isEmpty()); + assertTrue(store.get("tenant", "active").isPresent()); + } + + private static final class MutableClock extends Clock { + private Instant now = Instant.parse("2026-08-19T00:00:00Z"); + + void advance(Duration duration) { + now = now.plus(duration); + } + + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/interop/a2a/SseFramesTest.java b/mateclaw-server/src/test/java/vip/mate/interop/a2a/SseFramesTest.java new file mode 100644 index 00000000..47722dc1 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/interop/a2a/SseFramesTest.java @@ -0,0 +1,60 @@ +package vip.mate.interop.a2a; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SseFramesTest { + + @Test + void parsesSingleDataFrameAtBlankLine() { + List frames = SseFrames.parse("event: artifact-update\ndata: {\"x\":1}\n\n"); + + assertEquals(1, frames.size()); + assertEquals("artifact-update", frames.getFirst().event()); + assertEquals("{\"x\":1}", frames.getFirst().data()); + } + + @Test + void joinsMultiLineDataWithNewlines() { + List frames = SseFrames.parse("event: message\ndata: first\ndata: second\n\n"); + + assertEquals("first\nsecond", frames.getFirst().data()); + } + + @Test + void ignoresCommentHeartbeatFrames() { + List frames = SseFrames.parse(": heartbeat\n\n"); + + assertTrue(frames.isEmpty()); + } + + @Test + void flushesTrailingFrameWithoutFinalBlankLine() { + List frames = SseFrames.parse("data: tail"); + + assertEquals(1, frames.size()); + assertEquals("message", frames.getFirst().event()); + assertEquals("tail", frames.getFirst().data()); + } + + @Test + void preservesEventBoundaryAcrossMultipleFrames() { + List frames = SseFrames.parse(""" + event: status-update + data: working + + event: artifact-update + data: one + data: two + + """); + + assertEquals(2, frames.size()); + assertEquals("working", frames.get(0).data()); + assertEquals("one\ntwo", frames.get(1).data()); + } +}