Server: A2U

Deploy agents with A2U (Agent-to-User) event streaming

Deploy agents with A2U (Agent-to-User) event streaming for real-time agent-to-user communication.

CLI

pip install "praisonai[serve]"
export OPENAI_API_KEY="your-key"

# Start A2U server
praisonai serve a2u --port 8083

Expected Output:

🚀 Starting A2U Event Stream Server...
   Host: 127.0.0.1
   Port: 8083
📡 Event stream: http://127.0.0.1:8083/a2u/events/events
ℹ️  Info: http://127.0.0.1:8083/a2u/info
✅ Server started at http://127.0.0.1:8083

Python

from fastapi import FastAPI
import uvicorn
from praisonai.endpoints.a2u_server import (
    create_a2u_routes,
    emit_agent_started,
    emit_agent_response,
    emit_agent_completed,
)

app = FastAPI(title="PraisonAI A2U Server")

# Add A2U routes
create_a2u_routes(app)

# Add discovery endpoint
@app.get("/__praisonai__/discovery")
async def discovery():
    return {
        "schema_version": "1.0.0",
        "server_name": "praisonai-a2u",
        "providers": [{"type": "a2u", "name": "A2U Event Stream"}],
        "endpoints": [{"name": "events", "provider_type": "a2u"}]
    }

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

Expected Output:

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

A2U Endpoints

Endpoint	Method	Description
/a2u/info	GET	Server info and capabilities
/a2u/events/<stream>	GET	SSE event stream
/a2u/subscribe	POST	Subscribe to events
/a2u/unsubscribe	POST	Unsubscribe from events

Event Types

Event Type	Description
agent.started	Agent started processing
agent.thinking	Agent is thinking/processing
agent.response	Agent generated response
agent.completed	Agent completed task
agent.error	Agent encountered error
tool.called	Tool was invoked
tool.result	Tool returned result

Subscribe to Events

# Get server info
curl http://localhost:8083/a2u/info

# Subscribe to events
curl -X POST http://localhost:8083/a2u/subscribe \
  -H "Content-Type: application/json" \
  -d '{"stream": "events", "filters": ["agent.response"]}'

# Stream events (SSE)
curl -N http://localhost:8083/a2u/events/events

Subscribe Response:

{
  "subscription_id": "sub_abc123",
  "stream": "events",
  "stream_url": "/a2u/events/events"
}

Emit Events from Agent

from praisonai.endpoints.a2u_server import (
    emit_agent_started,
    emit_agent_thinking,
    emit_agent_response,
    emit_agent_completed,
)

# In your agent code
agent_id = "agent-123"

emit_agent_started(agent_id, "MyAgent")
emit_agent_thinking(agent_id, "Processing your request...")
emit_agent_response(agent_id, "Here is my response")
emit_agent_completed(agent_id, {"status": "success"})

SSE Event Format

event: agent.started
data: {"agent_id": "agent-123", "agent_name": "MyAgent"}
id: evt_abc123

event: agent.response
data: {"agent_id": "agent-123", "content": "Hello!"}
id: evt_def456

Python Client

import requests
import sseclient

# Subscribe to events
url = "http://localhost:8083/a2u/events/events"
response = requests.get(url, stream=True)
client = sseclient.SSEClient(response)

for event in client.events():
    print(f"Event: {event.event}")
    print(f"Data: {event.data}")

CLI Options

Option	Default	Description
--port	8083	Server port
--host	127.0.0.1	Server host
--api-key	-	API key for authentication

Troubleshooting

Issue	Fix
Port in use	lsof -i :8083
Missing deps	pip install "praisonai[serve]"
No events	Check agent is emitting events
SSE not connecting	Check firewall, use host="0.0.0.0"

Related

• A2A Server - Agent-to-agent protocol
• Agents Server - Deploy as HTTP server
• Endpoints CLI - Client for all server types