Coverage for src / lilbee / cli / commands / servers.py: 100%

1"""Serve (HTTP API) and mcp (stdio) server-boot commands.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from pathlib import Path 

7from typing import TYPE_CHECKING 

8 

9import typer 

10 

11from lilbee.cli.app import ( 

12 apply_overrides, 

13 console, 

14 data_dir_option, 

15 global_option, 

16) 

17from lilbee.core.config import cfg 

18 

19if TYPE_CHECKING: 

20 import uvicorn 

21 

22 

23def _port_file() -> Path: 

24 return cfg.data_dir / "server.port" 

25 

26 

27async def _run_server(server: uvicorn.Server, config: uvicorn.Config, host: str) -> None: 

28 """Start uvicorn, write port file, and clean up on shutdown.""" 

29 import atexit 

30 

31 from lilbee.parent_monitor import parse_parent_pid, watch_parent_async 

32 

33 port_path = _port_file() 

34 

35 def _cleanup_port_file() -> None: 

36 port_path.unlink(missing_ok=True) 

37 

38 if not config.loaded: 

39 config.load() 

40 server.lifespan = config.lifespan_class(config) 

41 await server.startup() 

42 

43 parent_pid = parse_parent_pid() 

44 parent_watcher: asyncio.Task[None] | None = None 

45 if parent_pid is not None: 

46 

47 def _on_parent_death() -> None: 

48 server.should_exit = True 

49 

50 parent_watcher = asyncio.create_task(watch_parent_async(parent_pid, _on_parent_death)) 

51 

52 try: 

53 if server.servers: 

54 sock = server.servers[0].sockets[0] 

55 actual_port = sock.getsockname()[1] 

56 port_path.parent.mkdir(parents=True, exist_ok=True) 

57 port_path.write_text(str(actual_port)) 

58 atexit.register(_cleanup_port_file) 

59 console.print(f"Listening on http://{host}:{actual_port}") 

60 await server.main_loop() 

61 finally: 

62 if parent_watcher is not None and not parent_watcher.done(): 

63 parent_watcher.cancel() 

64 port_path.unlink(missing_ok=True) 

65 await server.shutdown() 

66 

67 

68def serve( 

69 host: str = typer.Option(None, "--host", "-H", help="Bind address (default: 127.0.0.1)"), 

70 port: int = typer.Option(None, "--port", "-p", help="Port (default: 0/random)"), 

71 data_dir: Path | None = data_dir_option, 

72 use_global: bool = global_option, 

73) -> None: 

74 """Start the HTTP API server.""" 

75 apply_overrides(data_dir=data_dir, use_global=use_global) 

76 if host is not None: 

77 cfg.server_host = host 

78 if port is not None: 

79 cfg.server_port = port 

80 

81 import logging 

82 

83 import uvicorn 

84 

85 from lilbee.server import create_app 

86 

87 logging.getLogger("asyncio").setLevel(logging.ERROR) 

88 

89 config = uvicorn.Config(create_app(), host=cfg.server_host, port=cfg.server_port) 

90 server = uvicorn.Server(config) 

91 asyncio.run(_run_server(server, config, cfg.server_host)) 

92 

93 

94def mcp_cmd() -> None: 

95 """Start the MCP server (stdio transport) for agent integration.""" 

96 from lilbee.mcp_server import main 

97 

98 main()