Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.praison.ai/llms.txt

Use this file to discover all available pages before exploring further.

​
Agent Server Module

The Agent Server module provides an HTTP server with Server-Sent Events (SSE) for real-time agent communication.

​
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

from praisonaiagents.server import AgentServer, ServerConfig

# Create server with custom config
config = ServerConfig(
    host="0.0.0.0",
    port=8080,
    cors_origins=["http://localhost:3000"]
)

server = AgentServer(config=config)

# Start server (non-blocking)
server.start()

# Broadcast events to connected clients
server.broadcast("message", {"text": "Hello clients!"})

# Stop server
server.stop()

​
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!"}}'