Add a2a module: agent card, tasks, streaming, input-required and agent-to-agent calls on the A2A Java SDK, served over the JDK HTTP server
Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01G8ikz8xdWuTP5yun8DZ1hk
This commit is contained in:
Claude
@@ -0,0 +1,156 @@
|
||||
package com.ankurm.agents.a2a;
|
||||
|
||||
import io.a2a.A2A;
|
||||
import io.a2a.client.Client;
|
||||
import io.a2a.client.ClientEvent;
|
||||
import io.a2a.client.MessageEvent;
|
||||
import io.a2a.client.TaskEvent;
|
||||
import io.a2a.client.TaskUpdateEvent;
|
||||
import io.a2a.client.config.ClientConfig;
|
||||
import io.a2a.client.transport.jsonrpc.JSONRPCTransport;
|
||||
import io.a2a.client.transport.jsonrpc.JSONRPCTransportConfigBuilder;
|
||||
import io.a2a.spec.AgentCard;
|
||||
import io.a2a.spec.Artifact;
|
||||
import io.a2a.spec.Message;
|
||||
import io.a2a.spec.Part;
|
||||
import io.a2a.spec.Task;
|
||||
import io.a2a.spec.TaskArtifactUpdateEvent;
|
||||
import io.a2a.spec.TaskState;
|
||||
import io.a2a.spec.TaskStatusUpdateEvent;
|
||||
import io.a2a.spec.TextPart;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/** One call to a remote A2A agent: fetch its card, send a message, collect what comes back until the task settles. */
|
||||
public final class A2aCaller {
|
||||
|
||||
/** What came back: every event in order, and the last task snapshot seen. */
|
||||
public record Reply(List<ClientEvent> events, Task task, Message message) {
|
||||
|
||||
public String text() {
|
||||
if (task != null && task.artifacts() != null && !task.artifacts().isEmpty()) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (Artifact a : task.artifacts()) {
|
||||
for (Part<?> p : a.parts()) {
|
||||
if (p instanceof TextPart t) sb.append(t.text());
|
||||
}
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
if (task != null && task.status().message() != null) {
|
||||
return textOf(task.status().message());
|
||||
}
|
||||
return message == null ? "" : textOf(message);
|
||||
}
|
||||
}
|
||||
|
||||
private A2aCaller() {
|
||||
}
|
||||
|
||||
public static AgentCard fetchCard(String agentUrl) throws Exception {
|
||||
return A2A.getAgentCard(agentUrl);
|
||||
}
|
||||
|
||||
public static Reply send(String agentUrl, String text, String continueTaskId, boolean streaming) throws Exception {
|
||||
AgentCard card = fetchCard(agentUrl);
|
||||
List<ClientEvent> events = Collections.synchronizedList(new ArrayList<>());
|
||||
AtomicReference<Task> task = new AtomicReference<>();
|
||||
AtomicReference<Message> message = new AtomicReference<>();
|
||||
AtomicReference<Throwable> error = new AtomicReference<>();
|
||||
CountDownLatch settled = new CountDownLatch(1);
|
||||
|
||||
Client client = Client.builder(card)
|
||||
.withTransport(JSONRPCTransport.class, new JSONRPCTransportConfigBuilder())
|
||||
.clientConfig(ClientConfig.builder().setStreaming(streaming).build())
|
||||
.addConsumer((event, c) -> {
|
||||
events.add(event);
|
||||
if (event instanceof TaskEvent te) {
|
||||
task.set(te.getTask());
|
||||
if (stops(te.getTask().status().state())) settled.countDown();
|
||||
} else if (event instanceof TaskUpdateEvent tu) {
|
||||
task.set(tu.getTask());
|
||||
if (stops(tu.getTask().status().state())) settled.countDown();
|
||||
} else if (event instanceof MessageEvent me) {
|
||||
message.set(me.getMessage());
|
||||
settled.countDown();
|
||||
}
|
||||
})
|
||||
.streamingErrorHandler(t -> {
|
||||
error.set(t);
|
||||
settled.countDown();
|
||||
})
|
||||
.build();
|
||||
try {
|
||||
Message.Builder m = Message.builder().role(Message.Role.ROLE_USER)
|
||||
.parts(new TextPart(text)).messageId(UUID.randomUUID().toString());
|
||||
if (continueTaskId != null) {
|
||||
m.taskId(continueTaskId);
|
||||
}
|
||||
client.sendMessage(m.build());
|
||||
if (!settled.await(20, TimeUnit.SECONDS)) {
|
||||
throw new IllegalStateException("Timed out waiting for " + agentUrl);
|
||||
}
|
||||
if (error.get() != null) {
|
||||
throw new IllegalStateException("Streaming failed", error.get());
|
||||
}
|
||||
return new Reply(List.copyOf(events), task.get(), message.get());
|
||||
} finally {
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
|
||||
/** Asks the remote agent for a task it already knows about. */
|
||||
public static Task getTask(String agentUrl, String taskId) throws Exception {
|
||||
Client client = Client.builder(fetchCard(agentUrl))
|
||||
.withTransport(JSONRPCTransport.class, new JSONRPCTransportConfigBuilder())
|
||||
.build();
|
||||
try {
|
||||
return client.getTask(new io.a2a.spec.TaskQueryParams(taskId), null);
|
||||
} finally {
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean stops(TaskState s) {
|
||||
return s.isFinal() || s == TaskState.TASK_STATE_INPUT_REQUIRED || s == TaskState.TASK_STATE_AUTH_REQUIRED;
|
||||
}
|
||||
|
||||
static String textOf(Message m) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (Part<?> p : m.parts()) {
|
||||
if (p instanceof TextPart t) sb.append(t.text());
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
/** A one-line description of an event, for transcripts. */
|
||||
public static String describe(ClientEvent e) {
|
||||
if (e instanceof TaskEvent te) {
|
||||
return "TaskEvent state=" + short_(te.getTask().status().state()) + " artifacts=" + te.getTask().artifacts().size();
|
||||
}
|
||||
if (e instanceof MessageEvent me) {
|
||||
return "MessageEvent text=\"" + textOf(me.getMessage()) + "\"";
|
||||
}
|
||||
if (e instanceof TaskUpdateEvent tu) {
|
||||
if (tu.getUpdateEvent() instanceof TaskStatusUpdateEvent s) {
|
||||
return "TaskStatusUpdate state=" + short_(s.status().state()) + " final=" + s.isFinal();
|
||||
}
|
||||
if (tu.getUpdateEvent() instanceof TaskArtifactUpdateEvent a) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (Part<?> p : a.artifact().parts()) if (p instanceof TextPart t) sb.append(t.text());
|
||||
return "TaskArtifactUpdate append=" + a.append() + " lastChunk=" + a.lastChunk() + " text=\"" + sb + "\"";
|
||||
}
|
||||
}
|
||||
return e.getClass().getSimpleName();
|
||||
}
|
||||
|
||||
static String short_(TaskState s) {
|
||||
return s.name().replace("TASK_STATE_", "");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,224 @@
|
||||
package com.ankurm.agents.a2a;
|
||||
|
||||
import com.sun.net.httpserver.HttpExchange;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import io.a2a.grpc.utils.JSONRPCUtils;
|
||||
import io.a2a.grpc.utils.ProtoUtils;
|
||||
import io.a2a.jsonrpc.common.json.JsonUtil;
|
||||
import io.a2a.jsonrpc.common.wrappers.*;
|
||||
import io.a2a.server.ServerCallContext;
|
||||
import io.a2a.server.agentexecution.AgentExecutor;
|
||||
import io.a2a.server.auth.UnauthenticatedUser;
|
||||
import io.a2a.server.events.InMemoryQueueManager;
|
||||
import io.a2a.server.events.MainEventBus;
|
||||
import io.a2a.server.events.MainEventBusProcessor;
|
||||
import io.a2a.server.requesthandlers.DefaultRequestHandler;
|
||||
import io.a2a.server.tasks.BasePushNotificationSender;
|
||||
import io.a2a.server.tasks.InMemoryPushNotificationConfigStore;
|
||||
import io.a2a.server.tasks.InMemoryTaskStore;
|
||||
import io.a2a.server.util.sse.SseFormatter;
|
||||
import io.a2a.spec.A2AError;
|
||||
import io.a2a.spec.AgentCard;
|
||||
import io.a2a.spec.InternalError;
|
||||
import io.a2a.spec.TransportProtocol;
|
||||
import io.a2a.transport.jsonrpc.handler.JSONRPCHandler;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.HashMap;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Flow;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* Serves one A2A agent over JSON-RPC using only the JDK's built-in HTTP server. The SDK ships the protocol
|
||||
* logic (tasks, queues, streaming) and a Quarkus binding; this class is the small binding that replaces Quarkus.
|
||||
* The request routing follows the SDK's own Quarkus route class.
|
||||
*/
|
||||
public final class A2aHttpServer implements AutoCloseable {
|
||||
|
||||
private final HttpServer http;
|
||||
private final JSONRPCHandler rpc;
|
||||
private final AgentCard card;
|
||||
private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
private final MainEventBusProcessor processor;
|
||||
|
||||
private A2aHttpServer(HttpServer http, JSONRPCHandler rpc, AgentCard card, MainEventBusProcessor processor) {
|
||||
this.http = http;
|
||||
this.processor = processor;
|
||||
this.rpc = rpc;
|
||||
this.card = card;
|
||||
}
|
||||
|
||||
/** Starts on a free port. The card is built after the port is known, so it can carry the real URL. */
|
||||
public static A2aHttpServer start(Function<String, AgentCard> cardForUrl, AgentExecutor agent) throws IOException {
|
||||
HttpServer http = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
String url = "http://127.0.0.1:" + http.getAddress().getPort();
|
||||
AgentCard card = cardForUrl.apply(url);
|
||||
|
||||
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
MainEventBus bus = new MainEventBus();
|
||||
InMemoryTaskStore taskStore = new InMemoryTaskStore();
|
||||
InMemoryQueueManager queues = new InMemoryQueueManager(taskStore, bus);
|
||||
InMemoryPushNotificationConfigStore pushConfigs = new InMemoryPushNotificationConfigStore();
|
||||
MainEventBusProcessor processor = new MainEventBusProcessor(bus, taskStore, new BasePushNotificationSender(pushConfigs), queues);
|
||||
callPackagePrivate(processor, "start");
|
||||
DefaultRequestHandler handler = DefaultRequestHandler.create(agent, taskStore, queues, pushConfigs, processor, executor, executor);
|
||||
JSONRPCHandler rpc = new JSONRPCHandler(card, handler, executor);
|
||||
|
||||
A2aHttpServer server = new A2aHttpServer(http, rpc, card, processor);
|
||||
http.createContext("/", server::route);
|
||||
http.setExecutor(server.executor);
|
||||
http.start();
|
||||
return server;
|
||||
}
|
||||
|
||||
public String url() {
|
||||
return "http://127.0.0.1:" + http.getAddress().getPort();
|
||||
}
|
||||
|
||||
public AgentCard card() {
|
||||
return card;
|
||||
}
|
||||
|
||||
private void route(HttpExchange x) throws IOException {
|
||||
try (x) {
|
||||
String path = x.getRequestURI().getPath();
|
||||
if (x.getRequestMethod().equals("GET") && path.equals("/.well-known/agent-card.json")) {
|
||||
reply(x, JsonUtil.toJson(card));
|
||||
} else if (x.getRequestMethod().equals("POST") && path.equals("/")) {
|
||||
jsonRpc(x, new String(x.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
|
||||
} else {
|
||||
x.sendResponseHeaders(404, -1);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new IOException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private void jsonRpc(HttpExchange x, String body) throws Exception {
|
||||
ServerCallContext context = new ServerCallContext(UnauthenticatedUser.INSTANCE,
|
||||
new HashMap<>(java.util.Map.of(ServerCallContext.TRANSPORT_KEY, TransportProtocol.JSONRPC)), Set.of());
|
||||
A2AResponse<?> response = null;
|
||||
Flow.Publisher<? extends A2AResponse<?>> stream = null;
|
||||
try {
|
||||
A2ARequest<?> request = JSONRPCUtils.parseRequestBody(body, "");
|
||||
if (request instanceof SendMessageRequest r) {
|
||||
response = rpc.onMessageSend(r, context);
|
||||
} else if (request instanceof GetTaskRequest r) {
|
||||
response = rpc.onGetTask(r, context);
|
||||
} else if (request instanceof ListTasksRequest r) {
|
||||
response = rpc.onListTasks(r, context);
|
||||
} else if (request instanceof CancelTaskRequest r) {
|
||||
response = rpc.onCancelTask(r, context);
|
||||
} else if (request instanceof SendStreamingMessageRequest r) {
|
||||
stream = rpc.onMessageSendStream(r, context);
|
||||
} else if (request instanceof SubscribeToTaskRequest r) {
|
||||
stream = rpc.onSubscribeToTask(r, context);
|
||||
} else {
|
||||
response = new A2AErrorResponse(request.getId(), new io.a2a.spec.UnsupportedOperationError());
|
||||
}
|
||||
} catch (A2AError e) {
|
||||
response = new A2AErrorResponse(e);
|
||||
} catch (Exception e) {
|
||||
response = new A2AErrorResponse(new InternalError(e.getMessage()));
|
||||
}
|
||||
if (stream != null) {
|
||||
sse(x, stream);
|
||||
} else {
|
||||
reply(x, serialize(response));
|
||||
}
|
||||
}
|
||||
|
||||
private static String serialize(A2AResponse<?> response) {
|
||||
if (response instanceof A2AErrorResponse error) {
|
||||
return JSONRPCUtils.toJsonRPCErrorResponse(error.getId(), error.getError());
|
||||
}
|
||||
if (response.getError() != null) {
|
||||
return JSONRPCUtils.toJsonRPCErrorResponse(response.getId(), response.getError());
|
||||
}
|
||||
com.google.protobuf.MessageOrBuilder proto;
|
||||
if (response instanceof GetTaskResponse r) {
|
||||
proto = ProtoUtils.ToProto.task(r.getResult());
|
||||
} else if (response instanceof CancelTaskResponse r) {
|
||||
proto = ProtoUtils.ToProto.task(r.getResult());
|
||||
} else if (response instanceof SendMessageResponse r) {
|
||||
proto = ProtoUtils.ToProto.taskOrMessage(r.getResult());
|
||||
} else if (response instanceof ListTasksResponse r) {
|
||||
proto = ProtoUtils.ToProto.listTasksResult(r.getResult());
|
||||
} else {
|
||||
throw new IllegalArgumentException("Unsupported response: " + response.getClass().getName());
|
||||
}
|
||||
return JSONRPCUtils.toJsonRPCResultResponse(response.getId(), proto);
|
||||
}
|
||||
|
||||
private void sse(HttpExchange x, Flow.Publisher<? extends A2AResponse<?>> publisher) throws IOException, InterruptedException {
|
||||
x.getResponseHeaders().add("Content-Type", "text/event-stream");
|
||||
x.sendResponseHeaders(200, 0);
|
||||
OutputStream out = x.getResponseBody();
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
AtomicLong id = new AtomicLong();
|
||||
publisher.subscribe(new Flow.Subscriber<A2AResponse<?>>() {
|
||||
@Override
|
||||
public void onSubscribe(Flow.Subscription s) {
|
||||
s.request(Long.MAX_VALUE);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(A2AResponse<?> item) {
|
||||
try {
|
||||
out.write(SseFormatter.formatResponseAsSSE(item, id.getAndIncrement()).getBytes(StandardCharsets.UTF_8));
|
||||
out.flush();
|
||||
} catch (IOException e) {
|
||||
done.countDown();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
done.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onComplete() {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
done.await();
|
||||
}
|
||||
|
||||
private static void reply(HttpExchange x, String json) throws IOException {
|
||||
byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
|
||||
x.getResponseHeaders().add("Content-Type", "application/json");
|
||||
x.sendResponseHeaders(200, bytes.length);
|
||||
x.getResponseBody().write(bytes);
|
||||
}
|
||||
|
||||
/**
|
||||
* The SDK starts its event-bus thread from a CDI lifecycle callback, and the public ensureStarted() does nothing
|
||||
* outside CDI. Without this thread every event is accepted and none is delivered, and the client gets
|
||||
* "Could not find a Task/Message". start() and stop() are package-private, so this calls them by reflection.
|
||||
*/
|
||||
private static void callPackagePrivate(MainEventBusProcessor processor, String method) {
|
||||
try {
|
||||
var m = MainEventBusProcessor.class.getDeclaredMethod(method);
|
||||
m.setAccessible(true);
|
||||
m.invoke(processor);
|
||||
} catch (ReflectiveOperationException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
callPackagePrivate(processor, "stop");
|
||||
http.stop(0);
|
||||
executor.shutdownNow();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package com.ankurm.agents.a2a;
|
||||
|
||||
import io.a2a.server.agentexecution.AgentExecutor;
|
||||
import io.a2a.server.agentexecution.RequestContext;
|
||||
import io.a2a.server.tasks.AgentEmitter;
|
||||
import io.a2a.spec.A2AError;
|
||||
import io.a2a.spec.AgentCapabilities;
|
||||
import io.a2a.spec.AgentCard;
|
||||
import io.a2a.spec.AgentInterface;
|
||||
import io.a2a.spec.AgentSkill;
|
||||
import io.a2a.spec.Part;
|
||||
import io.a2a.spec.TaskState;
|
||||
import io.a2a.spec.TextPart;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
/**
|
||||
* Agent B: turns a comma-separated list of terms into a short brief. It does not know the definitions. For each
|
||||
* term it asks Agent A (the glossary) over A2A, so one agent is the server for its caller and the client of another.
|
||||
*/
|
||||
public final class BriefAgent implements AgentExecutor {
|
||||
|
||||
private final String glossaryUrl;
|
||||
/** What this agent sent to the glossary and what came back, for the transcript. */
|
||||
public final List<String> callLog = new CopyOnWriteArrayList<>();
|
||||
|
||||
public BriefAgent(String glossaryUrl) {
|
||||
this.glossaryUrl = glossaryUrl;
|
||||
}
|
||||
|
||||
public static AgentCard card(String url) {
|
||||
return AgentCard.builder()
|
||||
.name("Brief Agent")
|
||||
.description("Writes a short brief from a list of Java terms by asking the Glossary Agent")
|
||||
.version("1.0.0")
|
||||
.supportedInterfaces(List.of(new AgentInterface("JSONRPC", url)))
|
||||
.capabilities(AgentCapabilities.builder().streaming(true).build())
|
||||
.defaultInputModes(List.of("text/plain"))
|
||||
.defaultOutputModes(List.of("text/plain"))
|
||||
.skills(List.of(AgentSkill.builder()
|
||||
.id("brief").name("Brief on several terms").description("Comma-separated terms in, bullet list out")
|
||||
.tags(List.of("java", "summary")).examples(List.of("record, sealed class")).build()))
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(RequestContext context, AgentEmitter emitter) throws A2AError {
|
||||
emitter.submit();
|
||||
emitter.startWork();
|
||||
List<String> lines = new ArrayList<>();
|
||||
for (String term : context.getUserInput(" ").split(",")) {
|
||||
term = term.strip();
|
||||
try {
|
||||
A2aCaller.Reply reply = A2aCaller.send(glossaryUrl, term, null, false);
|
||||
TaskState state = reply.task().status().state();
|
||||
callLog.add("SendMessage \"" + term + "\" -> " + A2aCaller.short_(state) + " \"" + reply.text() + "\"");
|
||||
if (state != TaskState.TASK_STATE_COMPLETED) {
|
||||
emitter.fail(emitter.newAgentMessage(
|
||||
List.<Part<?>>of(new TextPart("Glossary Agent could not help: " + reply.text())), Map.of()));
|
||||
return;
|
||||
}
|
||||
lines.add("- " + term + ": " + reply.text());
|
||||
} catch (Exception e) {
|
||||
emitter.fail(emitter.newAgentMessage(List.<Part<?>>of(new TextPart("Glossary Agent unreachable: " + e.getMessage())), Map.of()));
|
||||
return;
|
||||
}
|
||||
}
|
||||
emitter.addArtifact(List.<Part<?>>of(new TextPart(String.join("\n", lines))), "brief", "brief", Map.of(), false, true);
|
||||
emitter.complete();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(RequestContext context, AgentEmitter emitter) throws A2AError {
|
||||
emitter.cancel();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package com.ankurm.agents.a2a;
|
||||
|
||||
import io.a2a.server.agentexecution.AgentExecutor;
|
||||
import io.a2a.server.agentexecution.RequestContext;
|
||||
import io.a2a.server.tasks.AgentEmitter;
|
||||
import io.a2a.spec.A2AError;
|
||||
import io.a2a.spec.AgentCapabilities;
|
||||
import io.a2a.spec.AgentCard;
|
||||
import io.a2a.spec.AgentInterface;
|
||||
import io.a2a.spec.AgentSkill;
|
||||
import io.a2a.spec.Part;
|
||||
import io.a2a.spec.TextPart;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Agent A: defines Java terms. It does no model call, because the point is the protocol. A real agent
|
||||
* would put an LLM or a database behind execute().
|
||||
*/
|
||||
public final class GlossaryAgent implements AgentExecutor {
|
||||
|
||||
static final Map<String, String> TERMS = Map.of(
|
||||
"record", "A record is a final class that is a transparent carrier for its components.",
|
||||
"sealed class", "A sealed class restricts which other classes may extend it.",
|
||||
"virtual thread", "A virtual thread is a lightweight thread scheduled by the JVM, not the OS.");
|
||||
|
||||
public static AgentCard card(String url) {
|
||||
return AgentCard.builder()
|
||||
.name("Glossary Agent")
|
||||
.description("Defines Java terms")
|
||||
.version("1.0.0")
|
||||
.supportedInterfaces(List.of(new AgentInterface("JSONRPC", url)))
|
||||
.capabilities(AgentCapabilities.builder().streaming(true).build())
|
||||
.defaultInputModes(List.of("text/plain"))
|
||||
.defaultOutputModes(List.of("text/plain"))
|
||||
.skills(List.of(AgentSkill.builder()
|
||||
.id("define").name("Define a term").description("Returns a one-sentence definition")
|
||||
.tags(List.of("java", "glossary")).examples(List.of("record")).build()))
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(RequestContext context, AgentEmitter emitter) throws A2AError {
|
||||
if (context.getTask() == null) {
|
||||
emitter.submit();
|
||||
}
|
||||
String term = context.getUserInput(" ").strip().toLowerCase();
|
||||
if (term.isEmpty()) {
|
||||
emitter.requiresInput(emitter.newAgentMessage(List.<Part<?>>of(new TextPart("Which term should I define?")), Map.of()));
|
||||
return;
|
||||
}
|
||||
emitter.startWork();
|
||||
String definition = TERMS.get(term);
|
||||
if (definition == null) {
|
||||
emitter.fail(emitter.newAgentMessage(List.<Part<?>>of(new TextPart("I do not know the term '" + term + "'.")), Map.of()));
|
||||
return;
|
||||
}
|
||||
// Two chunks of one artifact, so a streaming client sees the answer arrive in pieces.
|
||||
int cut = definition.indexOf(' ', definition.length() / 2);
|
||||
emitter.addArtifact(List.<Part<?>>of(new TextPart(definition.substring(0, cut))), "definition", "definition", Map.of(), false, false);
|
||||
emitter.addArtifact(List.<Part<?>>of(new TextPart(definition.substring(cut))), "definition", "definition", Map.of(), true, true);
|
||||
emitter.complete();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(RequestContext context, AgentEmitter emitter) throws A2AError {
|
||||
emitter.cancel();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.ankurm.agents.a2a;
|
||||
|
||||
import io.a2a.server.TransportMetadata;
|
||||
import io.a2a.spec.TransportProtocol;
|
||||
|
||||
/**
|
||||
* Tells the SDK that this application serves JSON-RPC. The SDK checks an agent card against the transports it
|
||||
* can discover, and the Quarkus module normally registers one; since this demo has no Quarkus, it registers its own.
|
||||
*/
|
||||
public class JdkJsonRpcTransportMetadata implements TransportMetadata {
|
||||
|
||||
@Override
|
||||
public String getTransportProtocol() {
|
||||
return TransportProtocol.JSONRPC.asString();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user