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

Event Types

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"})

Emitting events from tool callbacks

Sync emitters (emit_agent_started, emit_agent_thinking, emit_agent_tool_call, emit_agent_response, emit_agent_completed, emit_agent_error) are safe to call from any thread, including a tool callback firing on a background thread. The event bus schedules delivery on the SSE consumer’s own loop via call_soon_threadsafe, so events published from a different thread reach the stream instead of being silently dropped.

Events published in the small window between POST /a2u/subscribe and the client’s first GET /a2u/events/sub/{id} are buffered in a bounded pre-bind deque and drained into the queue on first consume. The cap is the same bound as the runtime queue (PRAISONAI_A2U_QUEUE_MAX), so a subscription that never consumes cannot grow without limit.

If the queue is full (slow or stalled consumer), events are still dropped with a warning log — only cross-loop delivery and the subscribe→consume gap are fixed.

publish_sync and tracked delivery

A2UEventBus.publish_sync(event) returns immediately with a tracked asyncio.Task. The module keeps a WeakSet of live tasks so the event loop can’t garbage-collect one mid-flight, and each task’s done-callback surfaces exceptions that were previously swallowed silently.

Use A2UEventBus.last_publish_task() to await the most recent scheduled publish — for example in tests, or when you need the real delivered subscriber count. The task’s result is the delivered count.

publish_sync checks asyncio.get_running_loop(). Inside a running loop it schedules the tracked task and returns the number of targeted subscribers; last_publish_task() then exposes the delivered count. Outside a running loop it blocks and returns the actual delivered count via the async bridge — but last_publish_task() stays None, so only await it from an async context. The related push/client.py future creation and acp/server.py asyncio.to_thread calls likewise use get_running_loop() instead of the deprecated asyncio.get_event_loop().

from praisonai.endpoints.a2u_server import A2UEventBus, A2UEvent

bus = A2UEventBus()

async def emit():
    bus.publish_sync(A2UEvent(event_type="job.completed", data={"job_id": "abc"}))
    delivered = await bus.last_publish_task()  # real subscriber count
    return delivered

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

Security & Limits

The A2U server fails closed on non-loopback binds without a token and caps subscriptions and per-subscription queues.

Fail-closed authentication

Loopback binds (127.0.0.1, ::1, localhost, empty) still allow unauthenticated traffic for local development. Every other bind without A2U_AUTH_TOKEN set returns 503:

{"error": "A2U_AUTH_TOKEN not configured; refusing non-loopback traffic"}

This matches the call-server security contract in Gateway Bind-Aware Auth.

Environment variables

Both _MAX_SUBS and _QUEUE_MAX fall back to their defaults on non-positive or unparseable env values, which prevents asyncio.Queue(maxsize<=0) from becoming silently unbounded.

Subscription and queue caps

When the subscription cap is reached, subscribe routes return 429:

{"error": "A2U subscription limit reached"}

When a subscription’s queue is full, publish_sync drops the event (put_nowait → asyncio.QueueFull) and logs:

A2U queue full for {sub_id} — dropping event {event_type}

HTTP responses

Troubleshooting

Related

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