gRPC¶

pip install "servicewright[grpc]"

A gRPC server entrypoint built on grpc-server-kit, with the standard health service, per-RPC DI scopes, correlation from metadata and ServiceError mapping already wired.

from servicewright import Service
from servicewright.adapters.grpc import GrpcConfig, GrpcEntrypoint


def register_servicers(server, ctx) -> None:
    orders_pb2_grpc.add_OrdersServicer_to_server(OrdersServicer(), server)


grpc_ep = GrpcEntrypoint(
    config=GrpcConfig(port=50051),
    servicers=register_servicers,
)

service = Service(spec, entrypoints=[grpc_ep])

servicers is called during bind with the raw grpc.aio.Server and the ServiceContext, so it can resolve singletons from ctx.app_scope. It may be async.

Per-RPC dependency scope¶

UnitScopeInterceptor is added first — the outermost interceptor — so every downstream interceptor and your servicer run inside a live unit scope:

from servicewright.adapters.grpc import current_unit_scope


class OrdersServicer(orders_pb2_grpc.OrdersServicer):
    async def GetOrder(self, request, context):
        scope = current_unit_scope()
        use_case = await scope.get(GetOrder)
        order = await use_case.execute(request.order_id)
        return orders_pb2.Order(id=order.id)

The scope stays open for the whole RPC, including the full response stream of a streaming call.

Correlation from metadata¶

The interceptor extracts and binds these into the context store:

Helpers for reading them straight off a servicer context:

from servicewright.adapters.grpc import (
    get_client_context,
    get_client_ip,
    get_idempotency_key,
    get_user_agent,
)

Error mapping¶

A ServiceError raised anywhere in your servicer is converted into the matching abort:

raise OrderNotFoundError("no such order")
# → the RPC aborts with NOT_FOUND
# → the machine code travels in the "x-error-code" trailing metadata

Non-public errors are masked exactly like over HTTP: the client sees a generic INTERNAL and the real code only reaches your logs. The full mapping table is in Errors.

Turn it off with map_service_errors=False.

Anything that is not a ServiceError¶

UnhandledErrorInterceptor is the last resort, and it is always installed:

raise RuntimeError("dividing by the number of retries, which is zero")
# → the RPC aborts with INTERNAL
# → details "internal_error", trailing metadata "x-error-code: internal_error"
# → the exception and its traceback go to the log, with the RPC's correlation ids

Byte for byte what a public=False ServiceError produces, which is the point: the two are indistinguishable to a caller. Without it grpc.aio answers UNKNOWN with repr() of the exception, so the one error class whose wording nobody reviewed is the one that reaches the wire verbatim.

Three things are left alone, because each already carries a decision: grpc.aio.AbortError and grpc.RpcError (a status a handler chose itself) are re-raised, and asyncio.CancelledError never reaches it — a caller that walked away is not a failure to report.

map_service_errors=False does not remove it. That flag turns off the mapping of the errors you declared; with it off a ServiceError reaches the client as a masked INTERNAL like any other unhandled exception, never as its own message.

Interceptor ordering¶

grpc.aio hands control in list order, so the first entry is the outermost wrapper. The entrypoint assembles the chain like this:

UnitScopeInterceptor            ← outermost: everything below sees a live scope
  UnhandledErrorInterceptor     ← the net: inside the scope, so the log is correlated
    metrics interceptor         ← records the status the client actually receives
      your static interceptors
      your interceptors_factory results
        ServiceErrorInterceptor ← innermost: closest to the servicer
          your servicer

GrpcEntrypoint(
    config=config,
    servicers=register_servicers,
    interceptors=[AuthInterceptor()],
    interceptors_factory=lambda ctx: [RateLimitInterceptor(ctx.app_scope)],
)

Health, reflection and channelz¶

The standard grpc.health.v1.Health service is always registered, driven by the same HealthRegistry as the HTTP probes:

readinessProbe:
  grpc:
    port: 50051

Because the gRPC health API is a push API, the entrypoint polls readiness every health_refresh_interval seconds while serving, so a dependency failing mid-life flips the status to NOT_SERVING within roughly one interval. On drain, every tracked name is pinned to NOT_SERVING so load balancers stop routing.

By default only the overall (empty-string) name is reported — the one a plain gRPC readiness probe checks. List your own services to answer Check for them too:

GrpcConfig(health_service_names=("my.pkg.Orders", "my.pkg.Payments"))

GrpcConfig(enable_reflection=True, enable_channelz=True)   # not in production

Metrics¶

GrpcEntrypoint(config=config, servicers=register_servicers, enable_metrics=True)

Records through whatever metrics backend the app configured — a null object when metrics are off, so the flag is always safe:

Prefix them with metrics_prefix="myapp".

GrpcConfig¶

The main knobs:

Plus the full server-tuning surface — keepalive (keepalive_time_ms, keepalive_timeout_ms, keepalive_permit_without_calls), HTTP/2 ping policy, message and metadata size limits, flow control windows, connection age limits, compression — and TLS (ssl_enabled, ssl_cert_file, ssl_key_file, ssl_ca_file, ssl_client_auth). See the API reference for every field.

GrpcConfig satisfies grpc-server-kit's settings protocol, so it can be handed straight to the kit's primitives.

From settings¶

grpc-server-kit ships the environment-facing side of the same field set, grpc_server_kit.settings.BaseGrpcServerSettings — a pydantic model, so grpc-server-kit[settings], which servicewright[settings] covers already. GrpcConfig.from_settings reads every field off it, so the mapping lives here rather than in each service:

from grpc_server_kit.settings import BaseGrpcServerSettings

from servicewright.adapters.grpc import GrpcConfig
from servicewright.adapters.settings import BaseServiceSettings


class Settings(BaseServiceSettings):
    grpc: BaseGrpcServerSettings = BaseGrpcServerSettings()


settings = Settings()  # GRPC__PORT=50051 GRPC__KEEPALIVE_TIME_MS=30000 GRPC__SSL_ENABLED=true ...
config = GrpcConfig.from_settings(settings.grpc, health_service_names=("my.pkg.Orders",))

settings is anything satisfying the kit's GrpcServerSettingsProtocol — the pydantic model, the stdlib grpc_server_kit.GrpcServerConfig, a model of your own — and every value comes from it, defaults included: the kit binds [::] with a 5 s grace_period, 8 KiB of metadata and keepalive_permit_without_calls=False where GrpcConfig() has 0.0.0.0, 30 s, 16 KiB and True. A test in servicewright asserts that every GrpcConfig field is either read from the settings object or in the table below, so a knob added to GrpcConfig cannot leave the mapping silently incomplete.

Three fields have no source on a settings object and are keyword arguments instead:

The kit's metrics_enabled has no GrpcConfig field either; that switch is GrpcEntrypoint(enable_metrics=settings.grpc.metrics_enabled).

Drain semantics¶

The effective drain budget is min(host_grace, config.grace_period):

• the Host's drain_grace_seconds bounds the shutdown;
• grace_period can only shorten it.

Raising grace_period above the Host's allowance has no effect — raise AppSpec.drain_grace_seconds instead.

Lifecycle notes¶

• bind() creates the server, registers your servicers and the health servicer, and binds the port. A clash aborts startup.
• serve() starts the server, starts the health poller, and waits for the stop event — still accepting.
• drain(grace) stops the poller, pins health to NOT_SERVING and lets in-flight RPCs finish.
• stop() is the hard stop.

grpc_ep.bound_port   # what the OS picked when port=0

grpc-server-kit's own run_async_grpc_server helper — which installs its own signal handlers and blocks — is deliberately not used. The Host owns signals.

Plugin form¶

from servicewright.adapters.grpc import GrpcPlugin

service = Service(spec, plugins=[GrpcPlugin(config=config, servicers=register_servicers)])