Module agents.backend.protocol

Protocol definitions for VideoSDK Agent backend communication.

This module defines the message types and structures used for communication between VideoSDK agents and the backend registry server.

Classes

class AvailabilityRequest (type: str = 'availability_request',
job_id: str = '',
room_id: str = '',
agent_name: str = '')

Expand source code

@dataclass
class AvailabilityRequest:
    """Request from server asking if worker is available for a job."""

    type: str = "availability_request"
    job_id: str = ""
    room_id: str = ""
    agent_name: str = ""

Request from server asking if worker is available for a job.

Instance variables

var agent_name : str

var job_id : str

var room_id : str

var type : str

class AvailabilityResponse (type: str = 'availability_response',
job_id: str = '',
available: bool = False,
error: str | None = None,
token: str | None = None)

Expand source code

@dataclass
class AvailabilityResponse:
    """Response from worker indicating availability."""

    type: str = "availability_response"
    job_id: str = ""
    available: bool = False
    error: Optional[str] = None
    token: Optional[str] = None  # Worker's auth token when accepting job

    def dict(self) -> Dict[str, Any]:
        """Convert to dictionary, excluding None values."""
        result = {}
        for key, value in self.__dict__.items():
            if value is not None:
                result[key] = value
        return result

Response from worker indicating availability.

Instance variables

var available : bool

var error : str | None

var job_id : str

var token : str | None

var type : str

Methods

def dict(self) ‑> Dict[str, Any]

Expand source code

def dict(self) -> Dict[str, Any]:
    """Convert to dictionary, excluding None values."""
    result = {}
    for key, value in self.__dict__.items():
        if value is not None:
            result[key] = value
    return result

Convert to dictionary, excluding None values.

class JobAssignment (type: str = 'job_assignment',
job_id: str = '',
room_id: str = '',
agent_name: str = '',
token: str = '',
url: str = '',
room_options: Dict[str, Any] | None = None,
metadata: Dict[str, Any] | None = None,
wait: bool = False)

Expand source code

@dataclass
class JobAssignment:
    """Job assignment from server to worker."""

    type: str = "job_assignment"
    job_id: str = ""
    room_id: str = ""
    agent_name: str = ""
    token: str = ""
    url: str = ""
    room_options: Optional[Dict[str, Any]] = None
    metadata: Optional[Dict[str, Any]] = None
    wait: bool = False

    def dict(self) -> Dict[str, Any]:
        """Convert to dictionary, excluding None values."""
        result = {}
        for key, value in self.__dict__.items():
            if value is not None:
                result[key] = value
        return result

Job assignment from server to worker.

Instance variables

var agent_name : str

var job_id : str

var metadata : Dict[str, Any] | None

var room_id : str

var room_options : Dict[str, Any] | None

var token : str

var type : str

var url : str

var wait : bool

Methods

def dict(self) ‑> Dict[str, Any]

Expand source code

def dict(self) -> Dict[str, Any]:
    """Convert to dictionary, excluding None values."""
    result = {}
    for key, value in self.__dict__.items():
        if value is not None:
            result[key] = value
    return result

Convert to dictionary, excluding None values.

class JobStatus (*args, **kwds)

Expand source code

class JobStatus(str, Enum):
    """Job status enumeration."""

    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

Job status enumeration.

Ancestors

• builtins.str
• enum.Enum

Class variables

var CANCELLED

var COMPLETED

var FAILED

var PENDING

var RUNNING

class JobTermination (type: str = 'job_termination', job_id: str = '', reason: str | None = None)

Expand source code

@dataclass
class JobTermination:
    """Job termination request from server."""

    type: str = "job_termination"
    job_id: str = ""
    reason: Optional[str] = None

Job termination request from server.

Instance variables

var job_id : str

var reason : str | None

var type : str

class JobUpdate (type: str = 'job_update',
job_id: str = '',
status: str = '',
error: str | None = None,
participant_identity: str | None = None,
participant_name: str | None = None,
participant_metadata: str | None = None)

Expand source code

@dataclass
class JobUpdate:
    """Update from worker about job status."""

    type: str = "job_update"
    job_id: str = ""
    status: str = ""
    error: Optional[str] = None
    participant_identity: Optional[str] = None
    participant_name: Optional[str] = None
    participant_metadata: Optional[str] = None

    def dict(self) -> Dict[str, Any]:
        """Convert to dictionary, excluding None values."""
        result = {}
        for key, value in self.__dict__.items():
            if value is not None:
                result[key] = value
        return result

Update from worker about job status.

Instance variables

var error : str | None

var job_id : str

var participant_identity : str | None

var participant_metadata : str | None

var participant_name : str | None

var status : str

var type : str

Methods

def dict(self) ‑> Dict[str, Any]

Expand source code

def dict(self) -> Dict[str, Any]:
    """Convert to dictionary, excluding None values."""
    result = {}
    for key, value in self.__dict__.items():
        if value is not None:
            result[key] = value
    return result

Convert to dictionary, excluding None values.

class ServerMessage (type: str,
worker_id: str | None = None,
success: bool | None = None,
message: str | None = None,
job_id: str | None = None,
room_id: str | None = None,
agent_name: str | None = None,
token: str | None = None,
url: str | None = None)

Expand source code

@dataclass
class ServerMessage:
    """Base message from server to worker."""

    type: str
    worker_id: Optional[str] = None
    success: Optional[bool] = None
    message: Optional[str] = None
    job_id: Optional[str] = None
    room_id: Optional[str] = None
    agent_name: Optional[str] = None
    token: Optional[str] = None
    url: Optional[str] = None

    def dict(self) -> Dict[str, Any]:
        """Convert to dictionary, excluding None values."""
        result = {}
        for key, value in self.__dict__.items():
            if value is not None:
                result[key] = value
        return result

Base message from server to worker.

Instance variables

var agent_name : str | None

var job_id : str | None

var message : str | None

var room_id : str | None

var success : bool | None

var token : str | None

var type : str

var url : str | None

var worker_id : str | None

Methods

def dict(self) ‑> Dict[str, Any]

Expand source code

def dict(self) -> Dict[str, Any]:
    """Convert to dictionary, excluding None values."""
    result = {}
    for key, value in self.__dict__.items():
        if value is not None:
            result[key] = value
    return result

Convert to dictionary, excluding None values.

class UpdateJobStatus (job_id: str,
status: JobStatus,
error: str | None = None,
participant_identity: str | None = None,
participant_name: str | None = None,
participant_metadata: str | None = None)

Expand source code

@dataclass
class UpdateJobStatus:
    """Update job status message."""

    job_id: str
    status: JobStatus
    error: Optional[str] = None
    participant_identity: Optional[str] = None
    participant_name: Optional[str] = None
    participant_metadata: Optional[str] = None

Update job status message.

Instance variables

var error : str | None

var job_id : str

var participant_identity : str | None

var participant_metadata : str | None

var participant_name : str | None

var status : JobStatus

class UpdateWorkerStatus (status: WorkerStatus,
load: float | None = None,
job_count: int | None = None,
error: str | None = None)

Expand source code

@dataclass
class UpdateWorkerStatus:
    """Update worker status message."""

    status: WorkerStatus
    load: Optional[float] = None
    job_count: Optional[int] = None
    error: Optional[str] = None

Update worker status message.

Instance variables

var error : str | None

var job_count : int | None

var load : float | None

var status : WorkerStatus

class WorkerMessage (type: str,
worker_id: str | None = None,
agent_name: str | None = None,
capabilities: List[str] | None = None,
status: str | None = None,
load: float | None = None,
job_count: int | None = None,
job_id: str | None = None,
available: bool | None = None,
participant_identity: str | None = None,
participant_name: str | None = None,
participant_metadata: str | None = None,
error: str | None = None,
registry_uuid: str | None = None,
max_capacity: int | None = None,
current_load: float | None = None,
token: str | None = None,
load_threshold: float | None = None,
max_processes: int | None = None)

Expand source code

@dataclass
class WorkerMessage:
    """Base message from worker to server."""

    type: str
    worker_id: Optional[str] = None
    agent_name: Optional[str] = None
    capabilities: Optional[List[str]] = None
    status: Optional[str] = None
    load: Optional[float] = None
    job_count: Optional[int] = None
    job_id: Optional[str] = None
    available: Optional[bool] = None
    participant_identity: Optional[str] = None
    participant_name: Optional[str] = None
    participant_metadata: Optional[str] = None
    error: Optional[str] = None
    registry_uuid: Optional[str] = None
    max_capacity: Optional[int] = None
    current_load: Optional[float] = None
    token: Optional[str] = None  # Authentication token for registry

    # Workload configuration from agent
    load_threshold: Optional[float] = None  # Agent's load threshold (e.g., 0.8)
    max_processes: Optional[int] = None  # Agent's max processes (e.g., 3)

    def dict(self) -> Dict[str, Any]:
        """Convert to dictionary, excluding None values."""
        result = {}
        for key, value in self.__dict__.items():
            if value is not None:
                result[key] = value
        return result

Base message from worker to server.

Instance variables

var agent_name : str | None

var available : bool | None

var capabilities : List[str] | None

var current_load : float | None

var error : str | None

var job_count : int | None

var job_id : str | None

var load : float | None

var load_threshold : float | None

var max_capacity : int | None

var max_processes : int | None

var participant_identity : str | None

var participant_metadata : str | None

var participant_name : str | None

var registry_uuid : str | None

var status : str | None

var token : str | None

var type : str

var worker_id : str | None

Methods

def dict(self) ‑> Dict[str, Any]

Expand source code

def dict(self) -> Dict[str, Any]:
    """Convert to dictionary, excluding None values."""
    result = {}
    for key, value in self.__dict__.items():
        if value is not None:
            result[key] = value
    return result

Convert to dictionary, excluding None values.

class WorkerPing (type: str = 'ping', timestamp: int | None = None)

Expand source code

@dataclass
class WorkerPing:
    """Ping message from worker."""

    type: str = "ping"
    timestamp: Optional[int] = None

Ping message from worker.

Instance variables

var timestamp : int | None

var type : str

class WorkerPong (type: str = 'pong', timestamp: int | None = None)

Expand source code

@dataclass
class WorkerPong:
    """Pong response from server."""

    type: str = "pong"
    timestamp: Optional[int] = None

Pong response from server.

Instance variables

var timestamp : int | None

var type : str

class WorkerStatus (*args, **kwds)

Expand source code

class WorkerStatus(str, Enum):
    """Worker status enumeration."""

    IDLE = "idle"
    BUSY = "busy"
    OFFLINE = "offline"
    ERROR = "error"

Worker status enumeration.

Ancestors

• builtins.str
• enum.Enum

Class variables

var BUSY

var ERROR

var IDLE

var OFFLINE