Praison AI

Agent Server Module

HTTP server with SSE event streaming for real-time agent communication

python
from praisonaiagents import Agent

agent = Agent(name="server-agent", instructions="Serve agent requests over HTTP.")
agent.start("Start the agent server on port 8080.")

Expose agents over HTTP with Server-Sent Events so clients receive live updates without polling.

The user connects a web client; the agent server streams SSE events while the agent processes each HTTP request.

Diagram

REST

SSE /events

🌐 Client

🛰️ AgentServer

🤖 Agent run

How It Works

A client subscribes to the event stream, and the server pushes agent updates live as work progresses.

Diagram
AgentAgentServerClientUserAgentAgentServerClientUserOpen connectionGET /events (SSE)Run taskProgress eventsStream updatesLive results

Features

  • REST API – HTTP endpoints for agent operations
  • SSE Streaming – Real-time event streaming to clients
  • CORS Support – Configurable cross-origin settings
  • Multi-client – Handle multiple concurrent connections

Quick Start

  1. Start a server
    python
    from praisonaiagents.server import AgentServer, ServerConfig

config = ServerConfig(host="0.0.0.0", port=8080, cors_origins=["http://localhost:3000"])
server = AgentServer(config=config)
server.start()
server.broadcast("message", {"text": "Hello clients!"})
  2. Use as a context manager
    python
    from praisonaiagents.server import AgentServer

with AgentServer(port=8080) as server:
    server.broadcast("status", {"ready": True})
# Server stops automatically on exit

API Reference

ServerConfig

python
@dataclass
class ServerConfig:
    host: str = "127.0.0.1"
    port: int = 8765
    cors_origins: List[str] = []
    auth_token: Optional[str] = None
    max_connections: int = 100

AgentServer

python
class AgentServer:
    def start(self, blocking: bool = False) -> None:
        """Start the server."""

    def stop(self) -> None:
        """Stop the server and disconnect clients."""

    def broadcast(self, event_type: str, data: Dict) -> None:
        """Broadcast event to all connected clients."""

    @property
    def is_running(self) -> bool:
        """Check if server is running."""

    @property
    def client_count(self) -> int:
        """Get number of connected clients."""

HTTP Endpoints

EndpointMethodDescription
/healthGETHealth check
/infoGETServer information
/eventsGETSSE event stream
/publishPOSTPublish event to clients

Examples

Context Manager

python
from praisonaiagents.server import AgentServer

with AgentServer(port=8080) as server:
    # Server is running
    server.broadcast("status", {"ready": True})

    # Do work...

# Server automatically stopped

Event Handler

python
server = AgentServer()

@server.on_event("message")
def handle_message(data):
    print(f"Received: {data}")

server.start()

Client Connection (JavaScript)

javascript
const eventSource = new EventSource('http://localhost:8765/events');

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('Received:', data);
};

eventSource.addEventListener('message', (event) => {
    console.log('Message event:', JSON.parse(event.data));
});

Publishing Events

bash
# Publish via HTTP
curl -X POST http://localhost:8765/publish \
  -H "Content-Type: application/json" \
  -d '{"type": "notification", "data": {"text": "Hello!"}}'

Best Practices

Restrict CORS in production

Replace cors_origins=["*"] with explicit front-end origins before exposing the server publicly. AgentServer never allows credentialed CORS (cookies), whatever the origin list. For AgentOS rules, see CORS and credentials.

Set auth_token for remote clients

Use ServerConfig(auth_token=...) when the server binds outside loopback so /publish cannot be abused.

Prefer SSE for progress, REST for commands

Stream long agent runs on /events; submit work via REST and broadcast milestones to connected clients.

Monitor client_count during load tests

Watch server.client_count and tune max_connections before production traffic spikes.

Related

Async Jobs

Submit long-running work and poll or stream results via the jobs API.

 Agent API Launch

Launch agents as HTTP services with minimal setup.