Praison AI

A2U API

Agent-to-User event streaming API endpoints

A2U API

The A2U (Agent-to-User) API provides Server-Sent Events (SSE) streaming for real-time agent-to-user communication.

Overview

A2U enables real-time event streaming from agents to users, including:

  • Agent status updates
  • Thinking/processing indicators
  • Tool call notifications
  • Response streaming
  • Error notifications

When to Use

  • Real-time UI: Update UI as agent processes
  • Progress tracking: Show agent thinking/tool usage
  • Streaming responses: Display responses as they’re generated

Base URL + Playground

bash
# Start A2U server
praisonai serve a2u --host 127.0.0.1 --port 8083

Base URL: http://127.0.0.1:8083

Endpoints

GET /a2u/info

Get A2U server information and available event types.

curl
curl http://127.0.0.1:8083/a2u/info
Python
import requests

response = requests.get("http://127.0.0.1:8083/a2u/info")
info = response.json()
print(f"Event types: {info['event_types']}")

Response:

json
{
  "name": "A2U Event Stream",
  "version": "1.0.0",
  "streams": ["events"],
  "event_types": [
    "agent.started",
    "agent.thinking",
    "agent.tool_call",
    "agent.response",
    "agent.completed",
    "agent.error"
  ]
}

POST /a2u/subscribe

Create a subscription to an event stream.

ParamField string

Stream name to subscribe to

ParamField array

Optional list of event types to filter

curl
curl -X POST http://127.0.0.1:8083/a2u/subscribe \
  -H "Content-Type: application/json" \
  -d '{"stream": "events", "filters": ["agent.response", "agent.completed"]}'
Python
import requests

response = requests.post(
    "http://127.0.0.1:8083/a2u/subscribe",
    json={"stream": "events", "filters": ["agent.response"]}
)
subscription = response.json()
print(f"Stream URL: {subscription['stream_url']}")

Response:

json
{
  "subscription_id": "sub-abc123def456",
  "stream_name": "events",
  "stream_url": "http://127.0.0.1:8083/a2u/events/sub/sub-abc123def456",
  "created_at": "2024-01-15T10:30:00Z"
}

GET /a2u/events/{stream_name}

Subscribe and stream events via SSE.

stream_name string

Stream name to subscribe to

curl
curl -N http://127.0.0.1:8083/a2u/events/events
Python
import requests

response = requests.get(
    "http://127.0.0.1:8083/a2u/events/events",
    stream=True
)
for line in response.iter_lines():
    if line:
        print(line.decode())
JavaScript
const eventSource = new EventSource("http://127.0.0.1:8083/a2u/events/events");

eventSource.addEventListener("agent.response", (event) => {
  const data = JSON.parse(event.data);
  console.log("Response:", data.response);
});

eventSource.addEventListener("agent.completed", (event) => {
  console.log("Agent completed");
  eventSource.close();
});

SSE Events:

event: agent.started
data: {"agent_id": "agent-1", "agent_name": "Assistant"}
id: evt-001

event: agent.thinking
data: {"agent_id": "agent-1", "message": "Processing query..."}
id: evt-002

event: agent.tool_call
data: {"agent_id": "agent-1", "tool_name": "web_search", "arguments": {"query": "AI"}}
id: evt-003

event: agent.response
data: {"agent_id": "agent-1", "response": "Here are the results..."}
id: evt-004

event: agent.completed
data: {"agent_id": "agent-1", "result": "Task completed"}
id: evt-005

POST /a2u/unsubscribe

Unsubscribe from an event stream.

ParamField string

Subscription ID to unsubscribe

bash
curl -X POST http://127.0.0.1:8083/a2u/unsubscribe \
  -H "Content-Type: application/json" \
  -d '{"subscription_id": "sub-abc123def456"}'

Response:

json
{
  "status": "unsubscribed"
}

GET /a2u/health

A2U subsystem health check.

bash
curl http://127.0.0.1:8083/a2u/health \
  -H "Authorization: Bearer $A2U_AUTH_TOKEN"

Response:

json
{
  "status": "healthy",
  "active_subscriptions": 3,
  "active_streams": 1
}

Event Types

EventDescription
agent.startedAgent started processing
agent.thinkingAgent is thinking/processing
agent.tool_callAgent called a tool
agent.responseAgent generated a response
agent.completedAgent completed task
agent.errorAgent encountered an error

Errors

StatusDescription
200Success
400Invalid request
401Missing/invalid bearer token (when A2U_AUTH_TOKEN is set)
404Subscription not found
429Subscription cap reached — {"error": "A2U subscription limit reached"} (PRAISONAI_A2U_MAX_SUBS, default 1024)
500Server error
503Non-loopback bind without A2U_AUTH_TOKEN — {"error": "A2U_AUTH_TOKEN not configured; refusing non-loopback traffic"}

The 429 applies to POST /a2u/subscribe and GET /a2u/events/{stream_name} when the subscription cap is reached; the 503 applies to any /a2u/* route on a non-loopback bind with no token. See Server: A2U → Security & Limits.

CLI Equivalent

bash
# Start A2U server
praisonai serve a2u --port 8083

Python SDK

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

# Emit events from your agent code
emit_agent_started("agent-1", "Assistant")
emit_agent_response("agent-1", "Hello, how can I help?")
emit_agent_completed("agent-1", result="Done")

Notes

  • Events are delivered via Server-Sent Events (SSE)
  • Subscriptions auto-cleanup on disconnect
  • Use filters to reduce event volume
  • Events include timestamps and unique IDs

Related