跳转至

AG-UI — Agent-User Interaction Protocol

Bi-directional protocol from CopilotKit. The agent streams events — TEXT_DELTA, TOOL_CALL, STATE_PATCH, LIFECYCLE — to the frontend.

Native server

Python
from apogee_ai_comunication import (
    LifecycleEvents, NativeAgUiServer, ServerConfig, AgentMessage, MessagePart, MessageRole,
)

async def handler(_message, emitter) -> None:
    for token in ["Hi", " ", "there"]:
        await emitter.emit(LifecycleEvents.text_delta(token))
    await emitter.emit(LifecycleEvents.tool_call("calc", {"a": 1}))

server = NativeAgUiServer(ServerConfig(), handler)
async for event in server.stream_events(
    AgentMessage(role=MessageRole.USER, parts=(MessagePart(kind="text", text="hi"),))
):
    print(event.kind.value, event.payload)

The default handler emits a TEXT_DELTA for the input text plus the LIFECYCLE start/finish frames. Use the NativeEventEmitter directly to mix event kinds before forwarding to a fastapi/Starlette SSE endpoint.

ag-ui-protocol adapter (extras=ag-ui)

Python
from apogee_ai_comunication.infrastructure.ag_ui.ag_ui_protocol_server import AgUiProtocolServer
server = AgUiProtocolServer(ServerConfig())