Agent Server Module

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

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

Endpoint	Method	Description
/health	GET	Health check
/info	GET	Server information
/events	GET	SSE event stream
/publish	POST	Publish 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

Restrict CORS in production

Replace cors_origins=["*"] with explicit front-end origins before exposing the server publicly.

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.