feat: AG-UI as in-process KMP API with Ktor client/server transports

Model the AG-UI protocol as an in-process API of Java-style interfaces and classes, mirroring the official AG-UI Java SDK, and build it as a Kotlin Multiplatform library with two Ktor transport modules. Modules: - api: BaseEvent sealed hierarchy (all AG-UI event types), messages, state, tool, context, RunAgentInput/Parameters, Agent/AgentSubscriber/ EventDispatcher, AguiJson (kotlinx.serialization, "type" discriminator), plus a SerializersModule registering every event subtype polymorphically. - client: HttpAgent over Ktor SSE (POST RunAgentInput -> stream BaseEvent), implementing the Agent interface. - server: Application.agui()/Route.aguiAgent() exposing an Agent behind a POST + SSE endpoint via SSEServerContent, plus a runnable sample. Tests: api serialization round-trips, client SSE decode against a real test-host server, server endpoint integration.
2026-09-09 22:11:08 +03:00
parent a5d6c38573
commit fed7f7ab71
30 changed files with 1619 additions and 0 deletions
@@ -0,0 +1,81 @@
+package com.agui.server
+
+import com.agui.api.agent.Agent
+import com.agui.api.event.BaseEvent
+import com.agui.api.event.RunErrorEvent
+import com.agui.api.run.RunAgentInput
+import com.agui.api.serialization.AguiJson
+import io.ktor.server.application.Application
+import io.ktor.server.application.ApplicationCall
+import io.ktor.server.application.call
+import io.ktor.server.request.receiveText
+import io.ktor.server.response.respond
+import io.ktor.server.routing.Route
+import io.ktor.server.routing.post
+import io.ktor.server.routing.routing
+import io.ktor.server.sse.SSEServerContent
+import io.ktor.server.sse.ServerSSESession
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.flow.collect
+import kotlinx.serialization.SerializationException
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+
+/**
+ * Exposes [agent] over the AG-UI HTTP + SSE transport on [path].
+ *
+ * The endpoint accepts a `POST` with a [RunAgentInput] JSON body and replies with
+ * `text/event-stream`: each server-sent event carries one [BaseEvent] as its `data` payload.
+ *
+ * If the agent's flow throws, a [RunErrorEvent] is emitted before the stream closes.
+ *
+ * This function installs nothing globally: no content negotiation, no plugins — just an
+ * endpoint inside the application's routing. Add other routes freely.
+ */
+public fun Application.agui(
+    agent: Agent,
+    path: String = "/agui",
+    json: Json = AguiJson,
+) {
+    routing {
+        aguiAgent(agent, path, json)
+    }
+}
+
+/**
+ * Route-level variant for composing into an existing routing block. Unlike
+ * [Application.agui] it does not create its own routing context.
+ */
+public fun Route.aguiAgent(
+    agent: Agent,
+    path: String = "/agui",
+    json: Json = AguiJson,
+) {
+    post(path) {
+        val input: RunAgentInput = decodeInput(call, json)
+        val handler: suspend (ServerSSESession) -> Unit = { session ->
+            try {
+                agent.run(input).collect { event ->
+                    session.send(json.encodeToString<BaseEvent>(event))
+                }
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Throwable) {
+                session.send(json.encodeToString<BaseEvent>(RunErrorEvent(e.message ?: "Agent error")))
+            }
+        }
+        call.respond(SSEServerContent(call, handler))
+    }
+}
+
+private suspend fun decodeInput(call: ApplicationCall, json: Json): RunAgentInput {
+    val text = call.receiveText()
+    if (text.isBlank()) {
+        return RunAgentInput(threadId = "", runId = "")
+    }
+    return try {
+        json.decodeFromString(RunAgentInput.serializer(), text)
+    } catch (e: SerializationException) {
+        throw IllegalArgumentException("Malformed AG-UI RunAgentInput: ${e.message}", e)
+    }
+}
@@ -0,0 +1,39 @@
+package com.agui.server.demo
+
+import com.agui.api.agent.Agent
+import com.agui.api.event.RunFinishedEvent
+import com.agui.api.event.RunStartedEvent
+import com.agui.api.event.StateSnapshotEvent
+import com.agui.api.event.TextMessageContentEvent
+import com.agui.api.event.TextMessageEndEvent
+import com.agui.api.event.TextMessageStartEvent
+import com.agui.server.agui
+import io.ktor.server.engine.embeddedServer
+import io.ktor.server.netty.Netty
+import kotlinx.coroutines.flow.flow
+
+/**
+ * A minimal AG-UI server that echoes the last user message back.
+ *
+ * Run it with:  gradlew :server:run   (or start the embedded server)
+ * then:  curl -N -X POST http://localhost:8080/agui -H "Content-Type: application/json" \
+ *        -d '{"threadId":"t","runId":"r","messages":[{"id":"m","role":"user","content":"hi"}]}'
+ */
+private val echoAgent: Agent = Agent { input ->
+    flow {
+        emit(RunStartedEvent(threadId = input.threadId, runId = input.runId))
+        emit(StateSnapshotEvent(snapshot = input.state))
+        val reply = (input.messages.lastOrNull()?.content ?: "world").ifBlank { "world" }
+        val messageId = "msg-${input.runId}"
+        emit(TextMessageStartEvent(messageId = messageId))
+        emit(TextMessageContentEvent(messageId = messageId, delta = "Hello, $reply!"))
+        emit(TextMessageEndEvent(messageId = messageId))
+        emit(RunFinishedEvent(threadId = input.threadId, runId = input.runId))
+    }
+}
+
+fun main() {
+    embeddedServer(Netty, port = 8080, host = "0.0.0.0") {
+        agui(echoAgent, path = "/agui")
+    }.start(wait = true)
+}
@@ -0,0 +1,81 @@
+package com.agui.server
+
+import com.agui.api.agent.Agent
+import com.agui.api.event.RunErrorEvent
+import com.agui.api.event.RunFinishedEvent
+import com.agui.api.event.RunStartedEvent
+import com.agui.api.event.StateSnapshotEvent
+import com.agui.api.event.TextMessageContentEvent
+import com.agui.api.event.TextMessageEndEvent
+import com.agui.api.event.TextMessageStartEvent
+import com.agui.api.message.Message
+import com.agui.api.message.MessageRole
+import com.agui.api.run.RunAgentInput
+import com.agui.api.state.mutableState
+import com.agui.client.HttpAgent
+import io.ktor.client.plugins.sse.SSE
+import io.ktor.server.application.application
+import io.ktor.server.testing.testApplication
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runTest
+import kotlinx.serialization.json.JsonPrimitive
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class AguiServerTest {
+
+    private val agent: Agent = Agent { input ->
+        flow {
+            emit(RunStartedEvent(threadId = input.threadId, runId = input.runId))
+            emit(StateSnapshotEvent(snapshot = input.state))
+            val messageId = "msg-${input.runId}"
+            emit(TextMessageStartEvent(messageId = messageId))
+            val last = input.messages.lastOrNull()
+            emit(TextMessageContentEvent(messageId = messageId, delta = "Hello ${last?.content ?: "world"}"))
+            emit(TextMessageEndEvent(messageId = messageId))
+            emit(RunFinishedEvent(threadId = input.threadId, runId = input.runId))
+        }
+    }
+
+    private val failingAgent: Agent = Agent { _ ->
+        flow { throw IllegalStateException("boom") }
+    }
+
+    @Test
+    fun `POST runs an agent and streams events via SSE`() = runTest {
+        testApplication {
+            application {
+                agui(agent)
+            }
+            val httpAgent = HttpAgent(client = createClient { install(SSE) }, endpointUrl = "/agui")
+            val input = RunAgentInput(
+                threadId = "t1",
+                runId = "r1",
+                messages = listOf(Message(id = "m0", role = MessageRole.USER, content = "hi")),
+                state = mutableState("counter" to JsonPrimitive(42)),
+            )
+            val events = httpAgent.run(input).toList()
+
+            assertEquals(RunStartedEvent(threadId = "t1", runId = "r1"), events.first())
+            assertEquals(RunFinishedEvent(threadId = "t1", runId = "r1"), events.last())
+            val snapshot = events.filterIsInstance<StateSnapshotEvent>().single()
+            assertEquals(mutableState("counter" to JsonPrimitive(42)), snapshot.snapshot)
+            assertTrue(events.any { it is TextMessageContentEvent && it.delta == "Hello hi" })
+        }
+    }
+
+    @Test
+    fun `an agent failure is signalled with a RUN_ERROR event`() = runTest {
+        testApplication {
+            application {
+                agui(failingAgent)
+            }
+            val httpAgent = HttpAgent(client = createClient { install(SSE) }, endpointUrl = "/agui")
+            val events = httpAgent.run(RunAgentInput(threadId = "t1", runId = "r1")).toList()
+            val error = events.single() as RunErrorEvent
+            assertEquals("boom", error.message)
+        }
+    }
+}