From 5cf327412efc2c0ce4c6df98905b11f8663dfa37 Mon Sep 17 00:00:00 2001 From: MinJun Choi Date: Tue, 27 Oct 2026 17:23:56 +0900 Subject: [PATCH 1/3] [None][feat] Add an opt-in pygrpc-native server for OpenEngine Signed-off-by: MinJun Choi --- setup.py | 1 + tensorrt_llm/commands/serve.py | 17 +++++- tensorrt_llm/grpc/openengine/server.py | 38 ++++++++++++- .../integration/test_lists/test-db/l0_cpu.yml | 1 + .../references/trtllm_serve_cli.yaml | 9 +++ .../grpc/openengine/test_openengine.py | 56 +++++++++++++------ tests/unittest/grpc/test_grpc_optional.py | 27 +++++++++ 7 files changed, 129 insertions(+), 20 deletions(-) diff --git a/setup.py b/setup.py index 6ff03926441a..954bb271c3c7 100644 --- a/setup.py +++ b/setup.py @@ -727,6 +727,7 @@ def get_build_state_options(): extras_require={ "devel": devel_deps + grpc_smg_deps, "openengine": [], + "openengine-native": ["pygrpc-native>=0.5.0,<1"], "mx": mx_deps, "grpc-smg": grpc_smg_deps, }, diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py index 184bd8539e20..f14be98eca7a 100644 --- a/tensorrt_llm/commands/serve.py +++ b/tensorrt_llm/commands/serve.py @@ -1209,6 +1209,14 @@ def launch_visual_gen_server( help= "Expose per-rank KV occupancy for OpenEngine routing, independently of KV events.", status="prototype") +@stability_option( + "--openengine-grpc-server", + type=click.Choice(["grpcio", "pygrpc-native"]), + default="grpcio", + help="gRPC server for OpenEngine. pygrpc-native runs the HTTP/2 and " + "gRPC work in Rust outside the GIL and needs " + "`pip install \"tensorrt_llm[openengine-native]\"`.", + status="prototype") @stability_option( "--served_model_name", type=str, @@ -1310,6 +1318,7 @@ def serve( grpc: bool, grpc_protocol: str, openengine_enable_load_metrics: bool, + openengine_grpc_server: str, enable_visual_gen: bool, served_model_name: Optional[str], visual_gen_args: Optional[str], @@ -1342,6 +1351,11 @@ def serve( raise click.UsageError( "--openengine-enable-load-metrics requires --grpc --grpc-protocol openengine" ) + if openengine_grpc_server != "grpcio" and (not grpc or grpc_protocol + != "openengine"): + raise click.UsageError( + "--openengine-grpc-server requires --grpc --grpc-protocol openengine" + ) if moe_cluster_parallel_size is not None: logger.warning( "--moe_cluster_parallel_size / --cluster_size is deprecated and " @@ -1601,7 +1615,8 @@ def _serve_llm(): served_model_name=served_model_name, report_failure=_report_observed_child_failure, enable_load_metrics=openengine_enable_load_metrics, - subagent_affinity_auth_key=internal_disagg_auth_key) + subagent_affinity_auth_key=internal_disagg_auth_key, + grpc_server=openengine_grpc_server) else: # Default: launch OpenAI HTTP server launch_server( diff --git a/tensorrt_llm/grpc/openengine/server.py b/tensorrt_llm/grpc/openengine/server.py index 1e764cb28a62..0a9996482288 100644 --- a/tensorrt_llm/grpc/openengine/server.py +++ b/tensorrt_llm/grpc/openengine/server.py @@ -5,6 +5,7 @@ import asyncio import gc +import importlib.metadata import ipaddress import os import signal @@ -20,6 +21,7 @@ import grpc import uvloop from grpc_health.v1 import health_pb2, health_pb2_grpc +from packaging.version import Version from tensorrt_llm import LLM as PyTorchLLM from tensorrt_llm.grpc._protobuf import _warn_if_python_protobuf @@ -58,11 +60,36 @@ ("grpc.http2.max_pings_without_data", 0), ("grpc.http2.min_ping_interval_without_data_ms", 10000), ] +# The first release that honors grpc.so_reuseport, which multiple frontends need. +_PYGRPC_NATIVE_MIN_VERSION = "0.5.0" _format_bind_address = _format_host_port +def _check_pygrpc_native() -> None: + """Fail before the engine loads if pygrpc-native is missing or too old.""" + try: + installed = importlib.metadata.version("pygrpc-native") + except importlib.metadata.PackageNotFoundError: + installed = None + if installed is None or Version(installed) < Version(_PYGRPC_NATIVE_MIN_VERSION): + raise click.ClickException( + f"--openengine-grpc-server pygrpc-native requires " + f"pygrpc-native>={_PYGRPC_NATIVE_MIN_VERSION}, found {installed or 'none'}. " + 'Install it with: pip install "tensorrt_llm[openengine-native]"' + ) + + +def _create_grpc_server(grpc_server: str, options: list[tuple[str, Any]]) -> Any: + """Build the grpc.aio server, or pygrpc-native's drop-in replacement for it.""" + if grpc_server == "pygrpc-native": + import pygrpc_native.aio + + return pygrpc_native.aio.server(options=options) + return grpc.aio.server(options=options) + + def _is_loopback(host: str) -> bool: """Whether `host` resolves to a loopback address.""" cleaned = host.strip("[]") @@ -99,6 +126,7 @@ class OpenEngineServer: llm: Initialized TensorRT-LLM LLM instance. model: Model name accepted by Generate requests. subagent_affinity_auth_key: Shared key for authenticated parent affinity hints. + grpc_server: gRPC server implementation, "grpcio" or "pygrpc-native". """ def __init__( @@ -111,6 +139,7 @@ def __init__( frontend: FrontendClient | None = None, instance_id: str | None = None, subagent_affinity_auth_key: str | None = None, + grpc_server: str = "grpcio", ) -> None: self.host = host self.port = port @@ -125,7 +154,7 @@ def __init__( self._fatal_engine_failure = False options = list(_SERVER_OPTIONS) options.append(("grpc.so_reuseport", 1 if frontend is not None else 0)) - self._server = grpc.aio.server(options=options) + self._server = _create_grpc_server(grpc_server, options) self._input_executor = ThreadPoolExecutor( max_workers=4, thread_name_prefix="openengine_input" ) @@ -167,7 +196,7 @@ def __init__( "The listener is unauthenticated and unencrypted: restrict it to a trusted " "network or front it with an authenticating TLS proxy." ) - # grpc.aio raises RuntimeError itself when the bind fails, naming the + # Both servers raise RuntimeError themselves when the bind fails, naming the # address, so there is no zero return to check for. try: bound_port = self._server.add_insecure_port(bind_address) @@ -307,6 +336,7 @@ def launch_server( report_failure: Callable[[int, str, str], None] | None = None, enable_load_metrics: bool = False, subagent_affinity_auth_key: str | None = None, + grpc_server: str = "grpcio", ) -> None: """Launch the dedicated OpenEngine gRPC server. @@ -318,8 +348,11 @@ def launch_server( report_failure: Records a child exit before startup readiness. enable_load_metrics: Enable scheduler-owned routing-load snapshots, independently of KV events. subagent_affinity_auth_key: Shared key for authenticated parent affinity hints. + grpc_server: gRPC server implementation, "grpcio" or "pygrpc-native". """ _warn_if_python_protobuf() + if grpc_server == "pygrpc-native": + _check_pygrpc_native() async def serve() -> None: logger.info("Initializing TensorRT-LLM OpenEngine server...") @@ -491,6 +524,7 @@ def monitor_done(task: asyncio.Task) -> None: frontend=frontend, instance_id=instance_id, subagent_affinity_auth_key=subagent_affinity_auth_key, + grpc_server=grpc_server, ) _disable_gc_if_requested() await server.start() diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml index 89ac19c97063..35702e21ce7d 100644 --- a/tests/integration/test_lists/test-db/l0_cpu.yml +++ b/tests/integration/test_lists/test-db/l0_cpu.yml @@ -211,6 +211,7 @@ l0_cpu: - unittest/grpc/test_grpc_optional.py::test_protobuf_warning_skips_unavailable_api - unittest/grpc/test_grpc_optional.py::test_protobuf_warning_does_not_block_model_startup - unittest/grpc/test_grpc_optional.py::test_smg_server_startup_failure_cleans_up + - unittest/grpc/test_grpc_optional.py::test_openengine_pygrpc_native_is_checked_before_engine_start - unittest/inputs - unittest/kv_cache_manager_v2_tests - unittest/llmapi/test_deepseek_v32_encoding.py diff --git a/tests/unittest/api_stability/references/trtllm_serve_cli.yaml b/tests/unittest/api_stability/references/trtllm_serve_cli.yaml index 8c06a6d0ea57..36efb8b22554 100644 --- a/tests/unittest/api_stability/references/trtllm_serve_cli.yaml +++ b/tests/unittest/api_stability/references/trtllm_serve_cli.yaml @@ -378,6 +378,15 @@ commands: is_flag: true flags: - "--openengine-enable-load-metrics" + openengine_grpc_server: + type: Choice(['grpcio', 'pygrpc-native']) + default: grpcio + status: prototype + required: false + multiple: false + is_flag: false + flags: + - "--openengine-grpc-server" otlp_traces_endpoint: type: str default: null diff --git a/tests/unittest/grpc/openengine/test_openengine.py b/tests/unittest/grpc/openengine/test_openengine.py index 7856d31c95bc..8e0637d17fa5 100644 --- a/tests/unittest/grpc/openengine/test_openengine.py +++ b/tests/unittest/grpc/openengine/test_openengine.py @@ -45,6 +45,14 @@ pytestmark = [pytest.mark.cpu_only, pytest.mark.threadleak(enabled=False)] +@pytest.fixture(params=["grpcio", "pygrpc-native"]) +def grpc_server(request: pytest.FixtureRequest) -> str: + """Runs the server tests on both gRPC server implementations.""" + if request.param == "pygrpc-native": + pytest.importorskip("pygrpc_native", minversion="0.5.0") + return request.param + + def test_format_bind_address_brackets_ipv6() -> None: """A bare IPv6 host is bracketed; IPv4 and already-bracketed hosts are not.""" from tensorrt_llm.grpc.openengine.server import _format_bind_address @@ -83,7 +91,7 @@ async def exercise_stop(): asyncio.run(exercise_stop()) -def test_openengine_server_serves_the_control_contract() -> None: +def test_openengine_server_serves_the_control_contract(grpc_server: str) -> None: """The server binds, attaches Control, answers, and shuts down cleanly. Exercises TRT-LLM's lifecycle around the servicer -- port-zero resolution, @@ -108,14 +116,22 @@ def test_openengine_server_serves_the_control_contract() -> None: ) async def exercise_server() -> None: - server = OpenEngineServer(host="127.0.0.1", port=0, llm=llm, model="test-model") + server = OpenEngineServer( + host="127.0.0.1", port=0, llm=llm, model="test-model", grpc_server=grpc_server + ) # port=0 must be replaced by the kernel-assigned port, or nothing # downstream (including this test's channel) can reach the server. assert server.port != 0 await server.start() with pytest.raises(RuntimeError, match="Failed to bind"): - OpenEngineServer(host="127.0.0.1", port=server.port, llm=llm, model="test-model") + OpenEngineServer( + host="127.0.0.1", + port=server.port, + llm=llm, + model="test-model", + grpc_server=grpc_server, + ) channel = grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") try: control = openengine_pb2_grpc.ControlStub(channel) @@ -139,7 +155,7 @@ async def exercise_server() -> None: asyncio.run(exercise_server()) -def test_health_and_control_are_reachable_on_pod_address() -> None: +def test_health_and_control_are_reachable_on_pod_address(grpc_server: str) -> None: """The shared listener accepts Pod-IP probes and control calls on one port.""" try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as route: @@ -158,7 +174,7 @@ def test_health_and_control_are_reachable_on_pod_address() -> None: ) async def exercise_server() -> None: - server = OpenEngineServer("0.0.0.0", 0, llm, "test-model") + server = OpenEngineServer("0.0.0.0", 0, llm, "test-model", grpc_server=grpc_server) await server.start() try: async with grpc.aio.insecure_channel(f"{pod_ip}:{server.port}") as channel: @@ -178,7 +194,9 @@ async def exercise_server() -> None: asyncio.run(exercise_server()) -def test_standard_grpc_health_check_and_watch(monkeypatch: pytest.MonkeyPatch) -> None: +def test_standard_grpc_health_check_and_watch( + monkeypatch: pytest.MonkeyPatch, grpc_server: str +) -> None: """Readiness changes reach both RPCs without changing server liveness.""" import tensorrt_llm.grpc.openengine.server as oe_server @@ -192,7 +210,7 @@ def test_standard_grpc_health_check_and_watch(monkeypatch: pytest.MonkeyPatch) - ) async def exercise_server() -> None: - server = OpenEngineServer("127.0.0.1", 0, llm, "test-model") + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) await server.start() try: async with grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") as channel: @@ -252,7 +270,7 @@ async def check(service: str) -> int: asyncio.run(exercise_server()) -def test_standard_grpc_health_watch_reports_shutdown() -> None: +def test_standard_grpc_health_watch_reports_shutdown(grpc_server: str) -> None: """Shutdown withdraws inference and finishes watchers without failing liveness.""" llm = SimpleNamespace( args=SimpleNamespace(guided_decoding_backend=None), @@ -262,7 +280,7 @@ def test_standard_grpc_health_watch_reports_shutdown() -> None: ) async def exercise_server() -> None: - server = OpenEngineServer("127.0.0.1", 0, llm, "test-model") + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) await server.start() stop_task = None try: @@ -293,7 +311,7 @@ async def exercise_server() -> None: def test_inference_health_timeout_does_not_start_overlapping_checks( - monkeypatch: pytest.MonkeyPatch, + monkeypatch: pytest.MonkeyPatch, grpc_server: str ) -> None: """A stalled predicate withdraws readiness without accumulating worker threads.""" import tensorrt_llm.grpc.openengine.server as oe_server @@ -317,7 +335,7 @@ def check_health() -> bool: ) async def exercise_server() -> None: - server = OpenEngineServer("127.0.0.1", 0, llm, "test-model") + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) await server.start() try: async with grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") as channel: @@ -356,7 +374,7 @@ async def exercise_server() -> None: @pytest.mark.parametrize("fatal_field", ["_fatal_error", "_engine_dead"]) def test_fatal_engine_failure_fails_liveness_after_temporary_unavailability( - monkeypatch: pytest.MonkeyPatch, fatal_field: str + monkeypatch: pytest.MonkeyPatch, fatal_field: str, grpc_server: str ) -> None: """Only a sticky fatal error fails liveness; temporary readiness can recover.""" import tensorrt_llm.grpc.openengine.server as oe_server @@ -373,7 +391,7 @@ def test_fatal_engine_failure_fails_liveness_after_temporary_unavailability( ) async def exercise_server() -> None: - server = OpenEngineServer("127.0.0.1", 0, llm, "test-model") + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) await server.start() try: async with grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") as channel: @@ -428,7 +446,9 @@ async def exercise_server() -> None: asyncio.run(exercise_server()) -def test_inference_health_recovers_after_monitor_error(monkeypatch: pytest.MonkeyPatch) -> None: +def test_inference_health_recovers_after_monitor_error( + monkeypatch: pytest.MonkeyPatch, grpc_server: str +) -> None: """A failed poll clears readiness and a later successful poll restores it.""" import tensorrt_llm.grpc.openengine.server as oe_server @@ -448,7 +468,7 @@ def check_health() -> bool: ) async def exercise_server() -> None: - server = OpenEngineServer("127.0.0.1", 0, llm, "test-model") + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) await server.start() try: async with grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") as channel: @@ -490,7 +510,7 @@ async def exercise_server() -> None: @pytest.mark.parametrize("publisher_enabled", [False, True]) -def test_event_startup_requires_loaded_publisher(publisher_enabled: bool) -> None: +def test_event_startup_requires_loaded_publisher(publisher_enabled: bool, grpc_server: str) -> None: """Configured events must not advertise a source absent from the loaded engine.""" from tensorrt_llm.grpc.openengine.kv_events import KvEventsUnavailableError @@ -508,7 +528,9 @@ def test_event_startup_requires_loaded_publisher(publisher_enabled: bool) -> Non ) async def exercise_server() -> None: - server = OpenEngineServer(host="127.0.0.1", port=0, llm=llm, model="test-model") + server = OpenEngineServer( + host="127.0.0.1", port=0, llm=llm, model="test-model", grpc_server=grpc_server + ) try: if publisher_enabled: await server.start() diff --git a/tests/unittest/grpc/test_grpc_optional.py b/tests/unittest/grpc/test_grpc_optional.py index 2c484e2f28b0..8901d934d666 100644 --- a/tests/unittest/grpc/test_grpc_optional.py +++ b/tests/unittest/grpc/test_grpc_optional.py @@ -356,6 +356,33 @@ def import_without_grpc( ) +@pytest.mark.cpu_only +@pytest.mark.parametrize("installed", [None, "0.4.0"]) +def test_openengine_pygrpc_native_is_checked_before_engine_start( + monkeypatch: pytest.MonkeyPatch, installed: Optional[str] +) -> None: + """A missing or too old pygrpc-native fails with an install hint, not after the model loads.""" + pytest.importorskip("grpc") + import click + + from tensorrt_llm.grpc.openengine import server as server_module + + def version(name: str) -> str: + if installed is None: + raise importlib.metadata.PackageNotFoundError(name) + return installed + + monkeypatch.setattr(server_module.importlib.metadata, "version", version) + monkeypatch.setattr(server_module, "PyTorchLLM", None) + with pytest.raises(click.ClickException, match=r"tensorrt_llm\[openengine-native\]"): + server_module.launch_server( + "127.0.0.1", + 0, + {"backend": "pytorch", "model": "test-model"}, + grpc_server="pygrpc-native", + ) + + @pytest.mark.cpu_only @pytest.mark.parametrize("via_config", [False, True]) def test_smg_rejects_multiple_frontends_before_engine_start(tmp_path, via_config): From 327c2f72d0a0d8523382c643d80b5d288b7d385e Mon Sep 17 00:00:00 2001 From: MinJun Choi Date: Tue, 27 Oct 2026 17:38:13 +0900 Subject: [PATCH 2/3] [None][doc] Document the pygrpc-native option for OpenEngine Signed-off-by: MinJun Choi --- tensorrt_llm/grpc/openengine/README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tensorrt_llm/grpc/openengine/README.md b/tensorrt_llm/grpc/openengine/README.md index 5ce68cf169c1..98827301f1bc 100644 --- a/tensorrt_llm/grpc/openengine/README.md +++ b/tensorrt_llm/grpc/openengine/README.md @@ -168,6 +168,10 @@ Frontends send heartbeats once per second. A missing heartbeat for 10 seconds wi Multiple frontends require the default classic IPC executor and a nonzero port. SMG and AutoDeploy do not support this mode. Frontends share host CPU resources, so provision sufficient CPU and benchmark the intended model, router, and connection pool before choosing a frontend count. More processes do not necessarily increase throughput once GPU execution becomes the bottleneck. +## gRPC server implementation + +The server runs on `grpc.aio` by default. `--openengine-grpc-server pygrpc-native` runs it on [pygrpc-native](https://github.com/pygrpc/pygrpc-native) instead, a drop-in replacement for `grpc.aio.server()` that does the HTTP/2 and gRPC work in Rust outside the GIL. The servicers, health service and server options stay the same. Install it with `pip install "tensorrt_llm[openengine-native]"`; the server checks for it before loading the model. It works with multiple frontend processes. The internal coordinator between frontends always uses `grpc.aio`. + ## Schema and binding provenance The schema source is the Apache-2.0-licensed [`ai-dynamo/openengine`](https://github.com/ai-dynamo/openengine) repository at signed Git tag [`v0.1.0`](https://github.com/ai-dynamo/openengine/releases/tag/v0.1.0), Git commit `b5f2bd93721f7b888d3e2440679e0ae7012939d1`. That release maps to the public [`buf.build/openengine/openengine`](https://buf.build/openengine/openengine) module at immutable BSR release `768a93c7b44e40f28c692ad0b471a8f2`. From 9fd8878006a5693c0e1886c67fa815a873a15477 Mon Sep 17 00:00:00 2001 From: MinJun Choi Date: Tue, 27 Oct 2026 21:35:37 +0900 Subject: [PATCH 3/3] [None][test] Stream a Generate call through both OpenEngine servers Signed-off-by: MinJun Choi --- .../grpc/openengine/test_openengine.py | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/tests/unittest/grpc/openengine/test_openengine.py b/tests/unittest/grpc/openengine/test_openengine.py index 8e0637d17fa5..e67c6495e77c 100644 --- a/tests/unittest/grpc/openengine/test_openengine.py +++ b/tests/unittest/grpc/openengine/test_openengine.py @@ -31,6 +31,7 @@ from grpc_health.v1 import health_pb2, health_pb2_grpc # noqa: E402 from tensorrt_llm.grpc.openengine.bindings import ( # noqa: E402 + generation_pb2, kv_pb2, openengine_pb2_grpc, server_pb2, @@ -194,6 +195,81 @@ async def exercise_server() -> None: asyncio.run(exercise_server()) +class _StreamingResult: + """Streams one token per result, as GenerationResult does.""" + + def __init__(self, results: list[SimpleNamespace]) -> None: + self._results = results + self.finished = False + self.prompt_token_ids: list[int] = [] + self.outputs: list[SimpleNamespace] = [] + self.cached_tokens = 0 + self.error = None + + async def __aiter__(self): + for result in self._results: + self.finished = all(output.finish_reason for output in result.outputs) + result.finished = self.finished + self.prompt_token_ids = result.prompt_token_ids + self.outputs = result.outputs + yield result + + def abort(self) -> None: + pass + + +def _token_result(token_ids: list[int], text: str, finish_reason: str | None) -> SimpleNamespace: + output = SimpleNamespace( + index=0, + token_ids=token_ids, + text=text, + logprobs=None, + prompt_logprobs=None, + finish_reason=finish_reason, + stop_reason=None, + ) + return SimpleNamespace(prompt_token_ids=[1, 2], outputs=[output], cached_tokens=0, error=None) + + +def test_generate_streams_through_the_server(grpc_server: str) -> None: + """A Generate call streams its tokens and finish event over a real listener.""" + llm = SimpleNamespace( + args=SimpleNamespace(guided_decoding_backend=None), + llm_id="test-instance", + tokenizer=None, + _check_health=lambda: True, + generate_async=lambda **kwargs: _StreamingResult( + [_token_result([10], "A", None), _token_result([10, 11], "AB", "length")] + ), + ) + + async def exercise_server() -> None: + server = OpenEngineServer("127.0.0.1", 0, llm, "test-model", grpc_server=grpc_server) + await server.start() + try: + async with grpc.aio.insecure_channel(f"127.0.0.1:{server.port}") as channel: + inference = openengine_pb2_grpc.InferenceStub(channel) + request = generation_pb2.GenerateRequest( + request_id="request-1", + model="test-model", + prompt="hello", + stopping=generation_pb2.StoppingOptions(max_tokens=2), + ) + responses = [r async for r in inference.Generate(request, timeout=5)] + finally: + await server.stop(grace=0) + + assert [r.WhichOneof("event") for r in responses] == ["token", "token", "finished"] + assert [t.token_id for t in responses[0].token.tokens] == [10] + assert responses[0].token.text == "A" + assert [t.token_id for t in responses[1].token.tokens] == [11] + assert responses[1].token.text == "B" + assert responses[2].finished.reason == generation_pb2.FINISH_REASON_LENGTH + assert responses[2].usage.completion_tokens == 2 + + asyncio.run(exercise_server()) + + def test_standard_grpc_health_check_and_watch( monkeypatch: pytest.MonkeyPatch, grpc_server: str ) -> None: