Praison AI

Server: AGUI

Deploy agents as AG-UI protocol servers for CopilotKit integration

Deploy agents as AG-UI servers for CopilotKit and other AG-UI compatible frontends.

CLI

bash
pip install "praisonaiagents[os]"
export OPENAI_API_KEY="your-key"

# Start AGUI/A2U server
praisonai serve a2u --port 8002
OptionDefaultDescription
--host127.0.0.1Host to bind to
--port8002Server port
--fileagents.yamlAgents YAML file

Python

python
from praisonaiagents import Agent
from praisonaiagents import AGUI
from fastapi import FastAPI
import uvicorn

agent = Agent(
    name="Assistant",
    role="Helpful AI Assistant",
    goal="Help users with their questions",
    llm="gpt-4o-mini"
)

agui = AGUI(agent=agent)

app = FastAPI()
app.include_router(agui.get_router())

uvicorn.run(app, host="0.0.0.0", port=8000)

Expected Output:

INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000

Verify:

bash
curl http://localhost:8000/status

Response:

json
{"status": "available"}

Verify Routes

After include_router, app.routes won’t list /agui and /status as flat paths — Starlette nests them under an internal _IncludedRouter. The endpoints work at runtime; verify them with TestClient.

Diagram

Mount & Verify

include_router

GET /status

POST /agui

FastAPI app

_IncludedRouter wrapper

AGUI router: /agui, /status

TestClient

python
from fastapi import FastAPI
from fastapi.testclient import TestClient
from praisonaiagents import Agent, AGUI

agent = Agent(instructions="You are a helpful assistant")
agui = AGUI(agent=agent)

app = FastAPI()
app.include_router(agui.get_router())

client = TestClient(app)

# Health check
assert client.get("/status").status_code == 200

# AG-UI endpoint reachable (will stream SSE on real input)
assert client.post("/agui", json={"message": "hi"}).status_code != 404

agents.yaml

AGUI server is Python-only. No agents.yaml support.

AGUI Endpoints

EndpointMethodDescription
/aguiPOSTRun agent with SSE streaming
/statusGETCheck agent availability

AGUI Parameters

ParameterTypeDefaultDescription
agentAgent–PraisonAI Agent instance
agentsAgents–Multi-agent workflow
namestragent.nameAGUI endpoint name
descriptionstragent.roleAgent description
prefixstr""URL prefix

Concurrency & Thread Isolation

One AGUI(agent=agent) instance safely serves many concurrent clients — each run’s streamed tokens stay in its own SSE stream.

A shared Agent still means a shared chat_history. The request’s threadId does not create a separate conversation history. For per-thread memory, give each thread its own Agent.

Diagram

Agent per thread

POST /alice/agui

POST /bob/agui

Client A

AGUI(agent_alice)

Client B

AGUI(agent_bob)

Agent A
own history

Agent B
own history

Shared Agent (default)

POST /agui

POST /agui

Client A
thread-A

AGUI(agent)

Client B
thread-B

One Agent
one chat_history

BehaviourShared AgentAgent per thread
Streamed tokens isolated per runYesYes
tool_call_* events isolated per runYesYes
Custom praisonai.* events isolated per runYesYes
Conversation history isolated per threadIdNoYes

Agent per thread — mount one AGUI router per thread (or tenant) with prefix:

python
from praisonaiagents import Agent, AGUI
from fastapi import FastAPI

app = FastAPI()

for thread in ["alice", "bob"]:
    agent = Agent(instructions="You are a helpful assistant")
    app.include_router(AGUI(agent=agent, prefix=f"/{thread}").get_router())

# Client A → POST /alice/agui, Client B → POST /bob/agui

CopilotKit Integration

jsx
import { CopilotKit } from "@copilotkit/react-core";

function App() {
  return (
    <CopilotKit runtimeUrl="http://localhost:8000/agui">
      <YourApp />
    </CopilotKit>
  );
}

Send AGUI Request

bash
curl -X POST http://localhost:8000/agui \
  -H "Content-Type: application/json" \
  -d '{
    "threadId": "thread-123",
    "runId": "run-456",
    "state": {
      "messages": [
        {"id": "msg-1", "role": "user", "content": "Hello!"}
      ]
    }
  }'

Response (SSE Stream):

event: run_started
data: {"type": "run_started", "threadId": "thread-123", "runId": "run-456"}

event: text_message_start
data: {"type": "text_message_start", "messageId": "msg-2"}

event: text_message_content
data: {"type": "text_message_content", "messageId": "msg-2", "delta": "Hello! How can I help?"}

event: text_message_end
data: {"type": "text_message_end", "messageId": "msg-2"}

event: run_finished
data: {"type": "run_finished", "threadId": "thread-123", "runId": "run-456"}

Event Types

EventDescription
run_startedAgent run has started
run_finishedAgent run completed
run_errorTerminal error — includes mid-stream provider failures
text_message_startNew text message started
text_message_contentText content delta
text_message_endText message completed
tool_call_startTool call started
tool_call_argsTool call arguments
tool_call_endTool call completed
custom (praisonai.*)PraisonAI-specific events for recoverable failures and progress — see below

A terminal error reaches the client as run_error, now including mid-stream provider failures that were previously silently dropped. Recoverable conditions (retry, model fallback, streaming fallback) arrive as custom praisonai.* events while the run continues.

PraisonAI custom events

AG-UI has no first-class frames for recoverable failures or user-visible progress. PraisonAI carries these as custom events with a praisonai.* name. AG-UI clients are expected to ignore custom event names they do not recognise, so these are backwards-compatible.

Custom event nameWhen it firesvalue shape
praisonai.retryLLM call is being retried after a transient failure{ message, metadata }
praisonai.model_fallbackProvider swapped to a fallback model{ message, metadata }
praisonai.stream_unavailableStreaming path failed; falling back to non-streaming{ message, metadata }
praisonai.tool_progressA running tool emitted progress{ content, tool_call, metadata }
praisonai.todo_updatedThe agent’s todo list changed{ content, tool_call, metadata }

See the AGUI API reference for the SSE payload and a JavaScript handler.

JavaScript Client

javascript
const response = await fetch('http://localhost:8000/agui', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    threadId: 'thread-123',
    runId: 'run-456',
    state: {
      messages: [{ id: 'msg-1', role: 'user', content: 'Hello!' }]
    }
  })
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  const text = decoder.decode(value);
  const lines = text.split('\n');

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const event = JSON.parse(line.slice(6));
      console.log('Event:', event);
    }
  }
}

Troubleshooting

IssueFix
Port in uselsof -i :8000
Missing depspip install "praisonaiagents[os]"
No API keyexport OPENAI_API_KEY="your-key"
CORS errorsAdd CORS middleware to FastAPI app
/agui and /status missing from app.routesExpected — Starlette nests included routers under _IncludedRouter. Verify with TestClient (see Verify Routes) or inspect agui.get_router().routes directly. Endpoints work at runtime.

Related