Skip to content

Serving

any_agent.serving.ServerHandle dataclass

A handle for managing an async server instance.

This class provides a clean interface for managing the lifecycle of a server without requiring manual management of the underlying task and server objects.

Source code in src/any_agent/serving/server_handle.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass
class ServerHandle:
    """A handle for managing an async server instance.

    This class provides a clean interface for managing the lifecycle of a server
    without requiring manual management of the underlying task and server objects.
    """

    task: asyncio.Task[Any]
    server: UvicornServer

    async def shutdown(self, timeout_seconds: float = 10.0) -> None:
        """Gracefully shutdown the server with a timeout.

        Args:
            timeout_seconds: Maximum time to wait for graceful shutdown before forcing cancellation.

        """
        if not self.is_running():
            return  # Already shut down

        self.server.should_exit = True
        try:
            await asyncio.wait_for(self.task, timeout=timeout_seconds)
        except TimeoutError:
            logger.warning(
                "Server shutdown timed out after %ss, forcing cancellation",
                timeout_seconds,
            )
            self.task.cancel()
            try:
                await self.task
            except asyncio.CancelledError:
                pass
        except Exception as e:
            logger.error("Error during server shutdown: %s", e)
            # Still try to cancel the task to clean up
            if not self.task.done():
                self.task.cancel()
                try:
                    await self.task
                except asyncio.CancelledError:
                    pass

    def is_running(self) -> bool:
        """Check if the server is still running.

        Returns:
            True if the server task is still running, False otherwise.

        """
        return not self.task.done()

    @property
    def port(self) -> int:
        """Get the port the server is running on.

        If the server port was specified as 0, the port will be the one assigned by the OS.
        This helper method is useful to get the actual port that the server is running on.

        Returns:
            The port number the server is running on.

        """
        port = self.server.servers[0].sockets[0].getsockname()[1]
        assert port is not None
        assert isinstance(port, int)
        return port

port property

Get the port the server is running on.

If the server port was specified as 0, the port will be the one assigned by the OS. This helper method is useful to get the actual port that the server is running on.

Returns:

Type Description
int

The port number the server is running on.

is_running()

Check if the server is still running.

Returns:

Type Description
bool

True if the server task is still running, False otherwise.
Source code in src/any_agent/serving/server_handle.py 
56
57
58
59
60
61
62
63
def is_running(self) -> bool:
    """Check if the server is still running.

    Returns:
        True if the server task is still running, False otherwise.

    """
    return not self.task.done()

shutdown(timeout_seconds=10.0) async

Gracefully shutdown the server with a timeout.

Parameters:

Name Type Description Default
timeout_seconds float

Maximum time to wait for graceful shutdown before forcing cancellation.

 10.0
Source code in src/any_agent/serving/server_handle.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
async def shutdown(self, timeout_seconds: float = 10.0) -> None:
    """Gracefully shutdown the server with a timeout.

    Args:
        timeout_seconds: Maximum time to wait for graceful shutdown before forcing cancellation.

    """
    if not self.is_running():
        return  # Already shut down

    self.server.should_exit = True
    try:
        await asyncio.wait_for(self.task, timeout=timeout_seconds)
    except TimeoutError:
        logger.warning(
            "Server shutdown timed out after %ss, forcing cancellation",
            timeout_seconds,
        )
        self.task.cancel()
        try:
            await self.task
        except asyncio.CancelledError:
            pass
    except Exception as e:
        logger.error("Error during server shutdown: %s", e)
        # Still try to cancel the task to clean up
        if not self.task.done():
            self.task.cancel()
            try:
                await self.task
            except asyncio.CancelledError:
                pass