ag-ui-protocol

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)