Preparing a Python gRPC server for production¶

Starting a gRPC server in Python is easy: a handler and a few lines of startup code. Making it behave predictably when things go wrong is the hard part.

Take a GetInvoice method. It works on the happy path, but it still has to survive a request for someone else's order, a database outage, overload and a rollout. We will handle all of them in one gRPC server and check what the client actually receives.

The code was verified with grpc-server-kit 0.2.0, servicewright 0.13.1 and grpcio 1.84.0 on Python 3.13.

Start with the invoice operation¶

A buyer requests the invoice of a paid order. The buyer's identity comes from verified authentication; the application function checks ownership and payment. Its errors are servicewright ServiceError subclasses: kind is the category, code is a stable string for clients, and public (true by default) allows that code to leave the service. None of them knows about HTTP or gRPC:

from servicewright import ErrorKind, ServiceError


class OrderNotFound(ServiceError):
    kind = ErrorKind.NOT_FOUND
    code = "order_not_found"


class InvoiceNotReady(ServiceError):
    kind = ErrorKind.PRECONDITION_FAILED
    code = "invoice_not_ready"


class InvoiceAccessDenied(ServiceError):
    kind = ErrorKind.FORBIDDEN
    code = "invoice_access_denied"


class InvoiceStoreUnavailable(ServiceError):
    kind = ErrorKind.UNAVAILABLE
    code = "invoice_store_unavailable"

async def get_invoice(store, order_id: str, buyer_id: str) -> dict:
    async with store.read(order_id) as order:
        if order is None:
            raise OrderNotFound("Order not found")
        if order["buyer_id"] != buyer_id:
            raise InvoiceAccessDenied("Access denied")
        if not order["paid"]:
            raise InvoiceNotReady("The order has not been paid")
        return {"invoice_id": order["invoice_id"], "amount": order["amount"]}

store.read() releases its resource on exceptions and cancellation too. The function checks existence before ownership, so NOT_FOUND and PERMISSION_DENIED reveal which order IDs exist; check permission first if that matters. In the lab, the method is registered over raw bytes and the buyer is fixed fixture data.

The order is unpaid: what does the client receive?¶

InvoiceNotReady becomes FAILED_PRECONDITION: retrying without changing the order cannot help. The other statuses follow the gRPC status guide. We take them from each error's kind, so the gRPC table cannot drift from the HTTP one:

import grpc
from grpc_server_kit.aio.interceptors import AsyncExceptionHandlerInterceptor
from servicewright.adapters.grpc import GRPC_STATUS_BY_KIND


SERVICE = "orders.Invoices"
PUBLIC_ERRORS = (
    OrderNotFound, InvoiceNotReady, InvoiceAccessDenied, InvoiceStoreUnavailable,
)
ERROR_STATUS = {error: GRPC_STATUS_BY_KIND[error.kind] for error in PUBLIC_ERRORS}


def public_details(error, status):
    if isinstance(error, PUBLIC_ERRORS) and error.public:
        return error.code
    return "internal_error"


def exception_handler():
    return AsyncExceptionHandlerInterceptor(
        error_status_map=ERROR_STATUS,
        detail_factory=public_details,
        merge_defaults=False,
    )

public_details sends only the stable code of a known public error, never the exception text; anything else becomes internal_error. merge_defaults=False keeps only our table, because the library's defaults map ValueError to INVALID_ARGUMENT and would blame the client for an internal bug. The lab raises an internal ValueError through a real RPC:

An explicit await context.abort(...), for example from an authentication check, keeps its status; only uncaught exceptions are mapped. One gap remains: in grpc-server-kit 0.2.0, a grpc.RpcError from a downstream call passes through unchanged and reaches the client as UNKNOWN, together with the downstream service's details. Catch such errors in the handler and convert them. Logs need their own secret filtering: a safe response does not sanitize them.

Bound message sizes and active calls¶

from grpc_server_kit import GrpcApp, GrpcServerConfig


def server_config():
    return GrpcServerConfig(
        host="127.0.0.1", port=0,
        max_receive_message_length=64 * 1024,
        max_send_message_length=64 * 1024,
        max_concurrent_rpcs=32,
        grace_period=1.0,
    )

These are lab values. In production, listen on the pod address rather than 127.0.0.1, give grace_period enough time for real RPCs (the default is 5 seconds) and size the limits from your messages and capacity; the 64 KiB limit applies to requests and responses alike. The lab lowers the RPC limit to one and holds the first request inside its read:

max_concurrent_rpcs sets both the server-wide limit and the connection's grpc.max_concurrent_streams, so overload does not always produce an immediate refusal: clients still need deadlines to bound their wait.

Metrics need the status; diagnostics need the original exception¶

from grpc_server_kit.aio.interceptors import (
    AsyncMetricsInterceptor, AsyncSentryInterceptor,
)
from grpc_server_kit.aio.interceptors.exception_handler import find_mapped_status


def build_app(store, metrics, reporter, *, config=None):
    app = GrpcApp(config or server_config(), interceptors=[
        AsyncMetricsInterceptor(metrics, service_name=SERVICE),
        exception_handler(),
        AsyncSentryInterceptor(reporter, capture_filter=lambda error:
            find_mapped_status(type(error), ERROR_STATUS) in {
                grpc.StatusCode.INTERNAL, grpc.StatusCode.UNAVAILABLE,
            }),
    ])
    app.register(lambda server: register_invoice_service(server, store))
    return app

metrics and reporter are lab recorders; AsyncSentryInterceptor accepts any reporter. The list runs outermost first, and exceptions travel back outward: the reporter sees the original exception, the handler turns it into INTERNAL, and metrics record the final status. The filter reports internal errors and storage outages but not refusals such as InvoiceNotReady. With the reporter placed outside the exception handler, the lab shows it receiving a finished abort instead of the original exception.

Health checks report readiness¶

class InvoiceStoreHealth:
    def __init__(self, store):
        self.store = store

    async def check(self) -> bool:
        return await self.store.ping()


def enable_readiness(app, store):
    app.enable_health(
        checkers=[InvoiceStoreHealth(store)],
        service_names=[SERVICE],
        cache_ttl=0,
        check_timeout=0.2,
    )

Call enable_readiness(app, store) before starting the server; check_timeout bounds the whole round of checks. With cache_ttl=0, the lab sees SERVING, NOT_SERVING and SERVING again as storage fails and recovers, and a hanging check becomes NOT_SERVING after the timeout. In production, cache the result so load-balancer probes do not hit storage on every call. A direct GetInvoice call still reaches the handler: the health service reports state, it does not block RPCs.

The client leaves, then the server shuts down¶

Release resources with context managers and finally, and do not treat CancelledError as an application failure. In the lab, both cancellation and a client deadline end the read and release the store; the client sees DEADLINE_EXCEEDED, while the server's metric records CANCELLED, two views of one termination.

async def serve_invoices(app, store):
    try:
        await app.run()
    finally:
        await store.close()

app.run() installs signal handlers and serves until a signal or app.request_shutdown(). It then stops accepting RPCs and gives active ones grace_period to finish: a short request returns its response, a stuck one is cancelled and its client gets UNAVAILABLE. Storage closes only after that. If servicewright owns the process, it stops entrypoints and resources in order, as in the lifecycle article.

Enable TLS and verify the client¶

TLS needs ssl_enabled, ssl_cert_file and ssl_key_file in GrpcServerConfig; ssl_ca_file with ssl_client_auth=True also requires client certificates (mTLS). The lab generates temporary certificates and checks real connections: a TLS client succeeds, mTLS requires a client certificate, and plaintext fails in both setups. The library checks PEM files only for their BEGIN/END markers. On Unix it refuses a private key readable by group or others; on Windows, ACLs control access instead.

One application function for HTTP and gRPC¶

The transport lab serves the same get_invoice through servicewright's FastAPI and gRPC entrypoints. servicewright maps each error by its kind:

---
config:
  theme: default
  look: classic
  flowchart:
    useMaxWidth: false
    wrappingWidth: 150
    padding: 12
    nodeSpacing: 24
    rankSpacing: 32
---
flowchart TD
    accTitle: Translate errors at the transport boundary
    accDescr: Application code reports the failure. Each transport adapter selects a status and the public details; internal details stay in diagnostics.
    A["Application error"]
    B["HTTP adapter"]
    C["gRPC adapter"]
    D["Status and safe response"]
    E["Internal diagnostics"]
    A --> B --> D
    A --> C --> D
    A -.-> E

The stable code travels in the problem JSON body and in x-error-code trailing metadata. Unlike our GrpcApp, servicewright also sends the message of a public error to the client, as gRPC details and as the HTTP detail, so keep those messages fit for clients. The 412 is servicewright's choice: RFC 9110 reserves 412 for failed conditional headers, and Google's API guidelines map FAILED_PRECONDITION to 400. The entrypoint does not reuse the GrpcApp assembled above either: its limits, interceptors, health and TLS are configured separately.

What to use in your service¶

Give every expected failure a kind and a stable code, derive statuses from that one place, and check what actually reaches the client, internal and downstream errors included. grpc-server-kit assembles the gRPC server with these interceptors, limits, health checks and TLS; servicewright runs HTTP and gRPC under one lifecycle. Failures while reading streams need their own checks, covered in the streaming interceptors article.

Examples and labs¶

• Lab: the anatomy of a production gRPC server
• Lab: Python exceptions to gRPC status codes
• Lab: one domain error, two transports