Repository: livekit/agents
File: livekit-agents/livekit/agents/worker.py (Python)

Code before:
            @server.rtc_session(agent_name="survey_agent")
            async def my_agent(job_ctx: JobContext): ...

            server.rtc_session(my_agent, agent_name="survey_agent")
        """

        def decorator(
            f: Callable[[JobContext], Awaitable[None]],
        ) -> Callable[[JobContext], Awaitable[None]]:
            if self._entrypoint_fnc is not None:
                raise RuntimeError(
                    "The AgentServer currently only supports registering only one rtc_session"
                )
            self._entrypoint_fnc = f
            self._request_fnc = on_request
            self._session_end_fnc = on_session_end
            self._agent_name = agent_name
            self._server_type = type
            return f

        if func is not None:
            return decorator(func)
        return decorator

    @overload
    def setup(
        self,
        func: Callable[[JobProcess], Any],
    ) -> Callable[[JobProcess], Any]: ...

    @overload
    def setup(
        self,
    ) -> Callable[[Callable[[JobProcess], Any]], Callable[[JobProcess], Any]]: ...

    def setup(
        self,
        func: Callable[[JobProcess], Any] | None = None,
    ) -> (
        Callable[[JobProcess], Any]
        | Callable[[Callable[[JobProcess], Any]], Callable[[JobProcess], Any]]
    ):
        """
        Decorator or direct registrar for the setup/prewarm function.

        Usage:
            @server.setup()
            def setup_process(job_proc: JobProcess): ...

            server.setup(setup_process)
        """

Review comment (suggestion):
any reason to not use decorators for setup_fnc? I know a lot of examples are using 
```python
server = AgentServer()


@server.setup()
def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()

```

Code after:
    ):
        """
        Decorator or direct registrar for the RTC session entrypoint.

        Usage:
            @server.rtc_session(agent_name="survey_agent")
            async def my_agent(job_ctx: JobContext): ...

            server.rtc_session(my_agent, agent_name="survey_agent")
        """

        def decorator(
            f: Callable[[JobContext], Awaitable[None]],
        ) -> Callable[[JobContext], Awaitable[None]]:
            if self._entrypoint_fnc is not None:
                raise RuntimeError(
                    "The AgentServer currently only supports registering only one rtc_session"
                )
            self._entrypoint_fnc = f
            self._request_fnc = on_request
            self._session_end_fnc = on_session_end
            self._agent_name = agent_name
            self._server_type = type
            return f

        if func is not None:
            return decorator(func)

        return decorator

    @property
    def worker_info(self) -> WorkerInfo:
        return WorkerInfo(
            http_port=self._http_server.port if self._http_server else 0,
            cloud_agents=bool(self._worker_token),
        )

    async def run(self, *, devmode: bool = False, unregistered: bool = False) -> None:
        """This method starts the worker's internal event loop, initializes any required
        executors, HTTP servers, and process pools, and optionally registers the worker
        with the LiveKit server.

        Args:
            devmode (bool, optional):
                If True, the worker runs in development mode.
                This affects certain environment-dependent defaults, such as the
                number of idle processes, logging verbosity, and load thresholds,
                making it easier to test and debug without production constraints.

            unregistered (bool, optional):
                If True, the worker will start without registering itself with the