diff --git a/standalone/README.md b/standalone/README.md index 9bf25eb..bd58840 100644 --- a/standalone/README.md +++ b/standalone/README.md @@ -29,6 +29,15 @@ java -jar standalone/build/libs/standalone-all.jar - `GET /health` — liveness - `POST /agentik/conversations` — создать беседу - `GET /agentik/conversations/{id}/events` — SSE-стрим ответов +- `POST /a2a/` — A2A JSON-RPC (`message/send`, `tasks/get`, `tasks/cancel`) +- `GET /a2a/.well-known/agent-card.json` — AgentCard + +### A2A + +Адаптер `A2aBridge` гоняет A2A-контекст на диалог :proto: `contextId` мапится на +`Conversation` (пустой/неизвестный `contextId` → новый диалог). Ответ — склеенный +текст хода; id внутреннего диалога возвращается в `metadata.agentikConversationId` +ответа. Задачи живут в in-memory `TaskStore` (не переживают рестарт процесса). ## Переменные окружения diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aBridge.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aBridge.kt new file mode 100644 index 0000000..33ccb6f --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aBridge.kt @@ -0,0 +1,98 @@ +package pw.binom.agentik.standalone + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.coroutineScope +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonObject +import mu.KotlinLogging +import pw.binom.a2a.model.Message +import pw.binom.a2a.model.Role +import pw.binom.a2a.model.TextPart +import pw.binom.a2a.server.AgentHandler +import pw.binom.agentik.proto.Agent +import pw.binom.agentik.proto.Content +import pw.binom.agentik.proto.Conversation +import pw.binom.agentik.proto.Event +import java.util.concurrent.ConcurrentHashMap + +private val log = KotlinLogging.logger {} + +/** + * Адаптер :proto [Agent] к A2A [AgentHandler] (pw.binom.a2a:server). + * + * Контекст A2A (contextId) мапится на диалог :proto: + * - неизвестный/пустой contextId -> createConversation(temp=false); + * - известный contextId -> getConversation(id); если диалог пропал (удаляли/рестартились) + * -> новый диалог, контекст пересоздаётся. + * - id внутреннего диалога отдаётся клиенту в `metadata.agentikConversationId` ответа. + * + * Ответ A2A = склеенные [Event.AppendText] нашего хода. Подписку на [Conversation.events] + * открываем ДО [Conversation.send] (иначе события начала хода могут быть упущены), + * завершение хода ждём по [Event.End] / [Event.Interrupted] / [Event.Error]. + * + * Ограничение v1: tool-события и картинки в A2A-ответ не транслируются; + * при нескольких ходов в очереди за контекстом текст предыдущего хода + * может попасть в ответ. + */ +class A2aBridge(private val agent: Agent) : AgentHandler { + + private val contextToConversation = ConcurrentHashMap() + + override suspend fun handle(request: Message, contextId: String?): Message = coroutineScope { + val text = request.parts + .filterIsInstance() + .joinToString("\n") { it.text } + val conv = resolveConversation(contextId) + + val since = conv.updatedAt + val reply = StringBuilder() + val turnDone = CompletableDeferred() + val subscription = async { + conv.events(since).collect { e -> + when (e) { + is Event.AppendText -> reply.append(e.body) + is Event.End, is Event.Interrupted -> turnDone.complete(Unit) + is Event.Error -> + if (!turnDone.completeExceptionally( + IllegalStateException("agent turn failed: ${e.message}") + ) + ) {} + else -> {} + } + } + } + conv.send(listOf(Content.Text(text))) + try { + turnDone.await() + } finally { + subscription.cancel() + } + log.info { "a2a context=$contextId conv=${conv.id} reply=${reply.length} chars" } + Message( + role = Role.AGENT, + parts = listOf(TextPart(reply.toString().ifEmpty { "(empty response)" })), + metadata = buildJsonObject { + put("agentikConversationId", JsonPrimitive(conv.id)) + }, + ) + } + + private suspend fun resolveConversation(contextId: String?): Conversation { + if (contextId.isNullOrBlank()) { + val conv = agent.createConversation(temp = false) + contextToConversation[conv.id] = conv.id + log.info { "a2a: new context -> conv=${conv.id}" } + return conv + } + val existing = contextToConversation[contextId] + if (existing != null) { + agent.getConversation(existing)?.let { return it } + log.info { "a2a: context=$contextId conv=$existing lost, creating new" } + } + val conv = agent.createConversation(temp = false) + contextToConversation[contextId] = conv.id + log.info { "a2a: context=$contextId -> conv=${conv.id}" } + return conv + } +} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt index 179aed5..1e501a9 100644 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt @@ -10,6 +10,7 @@ import io.ktor.server.routing.get import io.ktor.server.routing.routing import kotlinx.coroutines.Dispatchers import kotlinx.io.files.Path +import pw.binom.a2a.server.a2aAgent import pw.binom.agentik.memory.MemoryReviewer import pw.binom.agentik.memory.MemorySystem import pw.binom.agentik.memory.md.openMdMemorySystem @@ -41,6 +42,8 @@ import java.io.File * GET /agentik/conversations/{id}/messages -> [Message] * GET /agentik/conversations/{id}/events -> text/event-stream (SSE) * GET /agentik/events -> text/event-stream (SSE, Agent-level) + * POST /a2a/ -> A2A JSON-RPC (message/send, tasks/get, tasks/cancel) + * GET /a2a/.well-known/agent-card.json -> AgentCard * GET /health -> "ok" * * Вся конфигурация — [AgentikConfig.fromEnv] (см. [AgentikConfig]). Источники: @@ -215,6 +218,7 @@ fun main() { routing { get("/health") { call.respondText("ok") } agentikAgent(agent, path = "/agentik") + a2aAgent(agentName = "agentik", handler = A2aBridge(agent), path = "/a2a") if (config.debugEndpoints) { debugRoutes( agent = agent, @@ -231,6 +235,8 @@ fun main() { println(" GET /health") println(" POST /agentik/conversations -> 201") println(" GET /agentik/conversations/{id}/events -> SSE") + println(" POST /a2a/ -> A2A JSON-RPC (message/send, tasks/get, tasks/cancel)") + println(" GET /a2a/.well-known/agent-card.json -> AgentCard") println(" storage: ${config.dbPath}") println(" llm: ${config.llm.backend} ${config.llm.modelInfo()}") println(" mcp: ${mcpRegistry.allTools.size} tools from ${mcpRegistry.connectedServerCount} servers")