AG-UI Protocol Python SDK

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from ag_ui.core import RunAgentInput, TextMessageContent, RunFinished, AgentEvent
from ag_ui.encoder import AgentEventEncoder
import uvicorn
import asyncio
import json
import os

app = FastAPI(title="AG-UI Endpoint Example")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Initial greeting event
        yield AgentEventEncoder.encode(TextMessageContent(text="Hello from AG-UI!"))
        
        # Echo the last text message from the input
        if input_data.messages:
            last_message = input_data.messages[-1]
            # Ensure the last message is a TextMessageContent for echoing
            if isinstance(last_message, TextMessageContent):
                yield AgentEventEncoder.encode(TextMessageContent(text=f"You said: {last_message.text}"))
            else:
                yield AgentEventEncoder.encode(TextMessageContent(text="Received a non-text message type."))
        else:
            yield AgentEventEncoder.encode(TextMessageContent(text="No messages provided in input."))

        # Simulate some asynchronous work
        await asyncio.sleep(0.5)
        yield AgentEventEncoder.encode(TextMessageContent(text="Performing some agent actions..."))
        await asyncio.sleep(0.5)

        # End the run with a RunFinished event
        yield AgentEventEncoder.encode(RunFinished())

    # Use StreamingResponse for Server-Sent Events (SSE)
    return StreamingResponse(event_generator(), media_type="text/event-stream")

if __name__ == "__main__":
    # To run this, you'll need fastapi and uvicorn installed:
    # pip install "fastapi[all]" uvicorn
    # Then, run this script directly:
    # python your_script_name.py
    # Or, if saved as main.py:
    # uvicorn main:app --host 0.0.0.0 --port 8000
    uvicorn.run(app, host="0.0.0.0", port=8000)