Index

Composition-first runtime for async gRPC services in the AIOps platform.

Features

  • Async gRPC server bootstrap with graceful shutdown
  • Declarative app spec instead of inheritance-heavy base classes
  • Reusable lifecycle hooks
  • Reusable observability configuration for logging, tracing, Sentry, and metrics
  • Warmup support before the server starts accepting traffic

Installation

1
uv add grpc-service-runtime --editable ../../../libs/python/runtime/api/grpc-service-runtime

Primary API

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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
import asyncio

import grpc
from grpc_service_runtime import (
    AppContext,
    AsyncGrpcApp,
    AsyncGrpcAppSpec,
    BootstrapContext,
)


def create_container(settings):
    return make_async_container(...)


def create_interceptors(ctx: BootstrapContext):
    return [
        MyTracingInterceptor(service_name=ctx.service_name),
        MyExceptionInterceptor(),
    ]


async def register_servicers(ctx: AppContext) -> None:
    my_pb2_grpc.add_MyServiceServicer_to_server(MyServicer(), ctx.server)


async def create_warmers(ctx: AppContext):
    return [MyDatabaseWarmer(await ctx.app_scope.get(MyDatabase))]


app = AsyncGrpcApp(
    AsyncGrpcAppSpec(
        service_name="my.package.MyService",
        health_service_name="grpc.health.v1.Health",
        create_container=create_container,
        interceptors_factory=create_interceptors,
        servicers_registerer=register_servicers,
        warmers_factory=create_warmers,
    )
)


if __name__ == "__main__":
    asyncio.run(app.serve(MySettings()))

Reusable Components

The runtime uses callbacks instead of complex module protocols:

  • servicers_registerer: Function (AppContext) -> None | Awaitable[None]
  • interceptors_factory: Function (BootstrapContext) -> Sequence[grpc.aio.ServerInterceptor] | Awaitable[Sequence[grpc.aio.ServerInterceptor]]
  • warmers_factory: Function (AppContext) -> Sequence[AsyncWarmer] | Awaitable[Sequence[AsyncWarmer]]
  • base_warmers: Direct list of AsyncWarmer instances
  • base_interceptors: Direct list of grpc.aio.ServerInterceptor instances

Development

1
2
uv run --extra dev ruff check .
uv run --extra test pytest tests/unit/async tests/integration/test_server_integration.py -q