Skip to main content
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.

How It Works

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

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

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

from praisonaiagents.server import AgentServer

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

API Reference

ServerConfig

@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

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

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

server = AgentServer()

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

server.start()

Client Connection (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

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

Best Practices

Replace cors_origins=["*"] with explicit front-end origins before exposing the server publicly.
Use ServerConfig(auth_token=...) when the server binds outside loopback so /publish cannot be abused.
Stream long agent runs on /events; submit work via REST and broadcast milestones to connected clients.
Watch server.client_count and tune max_connections before production traffic spikes.

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.