**Status:** Approved — ready for implementation

**Date:** 2026-04-19

**Scope:** Ship `app.mount_grpc(servicer, port=..., ...)` — a thin integration around `grpc.aio` (official grpcio). Users bring their own `*_pb2.py` / `*_pb2_grpc.py` (generated by `protoc` / `buf`). Framework ships: server lifecycle tied to ASGI lifespan, built-in observability interceptor, reflection toggle, TLS passthrough. Zero runtime deps on the default path — `grpcio` imported lazily.

from hawkapi import HawkAPI

from myproto import greeter_pb2, greeter_pb2_grpc

app = HawkAPI()

class Greeter(greeter_pb2_grpc.GreeterServicer):

async def SayHello(self, request, context):

# The HawkAPI app is attached to every ServicerContext.

app = context.hawkapi_app

name = await app.flags.string("greeter.prefix", default="Hello")

return greeter_pb2.HelloReply(message=f"{name}, {request.name}!")

app.mount_grpc(

Greeter(),

add_to_server=greeter_pb2_grpc.add_GreeterServicer_to_server,

port=50051,

)

The gRPC server starts when the ASGI app starts (`startup` lifespan event) and stops on `shutdown`. Works out-of-the-box under uvicorn/granian — one process serves both HTTP (ASGI) and gRPC (HTTP/2) on different ports.

def mount_grpc(

self,

servicer: object,

*,

add_to_server: Callable[[object, grpc.aio.Server], None],

port: int = 50051,

host: str = "[::]",

interceptors: Sequence[grpc.aio.ServerInterceptor] = (),

observability: bool = True,

reflection: bool = False,

ssl_credentials: grpc.ServerCredentials | None = None,

autostart: bool = True,

max_workers: int | None = None,

options: Sequence[tuple[str, Any]] = (),

) -> GrpcMount: ...

- `servicer` — an instance of a user-written `*Servicer` subclass.

- `add_to_server` — the generated `add_FooServicer_to_server` function (passed explicitly; we don't introspect).

- `interceptors` — user-provided `grpc.aio.ServerInterceptor` list; appended **after** the built-in observability interceptor (if enabled).

- `observability=True` — installs `HawkAPIObservabilityInterceptor` (logs method/status/duration, exports to the same Prometheus registry as HTTP middleware).

- `reflection=False` by default to avoid schema leak in prod; set `True` for local dev / staging.

- `ssl_credentials` — passed through to `server.add_secure_port`; when `None`, `add_insecure_port` is used.

- `autostart=True` — server starts on ASGI `startup`; set `False` for tests/custom wiring.

- Returns `GrpcMount` handle with `.server`, `.port`, `.start()`, `.stop(grace: float = 5.0)` for manual control.

Multiple `mount_grpc` calls are supported — each call can target a different port, or the same port with a different servicer (handlers are merged on the same grpc.aio.Server instance when ports match).

HawkAPI already fires `startup` / `shutdown` callbacks. `mount_grpc` registers:

- `startup` → `await grpc_mount._start()` — builds `grpc.aio.server(...)`, calls `add_to_server(servicer, server)`, registers interceptors, binds port, calls `await server.start()`.

- `shutdown` → `await grpc_mount._stop(grace=5.0)` — `await server.stop(grace)`.

Errors during `_start` propagate to the ASGI runner (uvicorn shows them and exits). No background task leaks.

Built-in interceptor attaches two attributes to every `ServicerContext`:

context.hawkapi_app # reference to the HawkAPI app

context.hawkapi_request_id # a uuid4() string, surfaced in logs

Users who want typed context helpers can build them on top — we stay out of DI territory in v1. `Depends()` resolution is explicitly deferred (see Out of scope).

`HawkAPIObservabilityInterceptor(app)` instruments every unary + streaming call:

- Logs structured entry `{event: "grpc.request", method, peer, request_id}` and exit `{event: "grpc.response", method, code, duration_ms}` via the existing `StructuredLoggingMiddleware`-compatible logger (uses `hawkapi.observability.get_logger()`).

- Exports Prometheus counters/histograms on the **same** `CollectorRegistry` as `PrometheusMiddleware`:

- `hawkapi_grpc_requests_total{method, code}`

- `hawkapi_grpc_request_duration_seconds{method}` (histogram)

- Reuses `ObservabilityConfig` for sampling / exporter choice.

Emission is best-effort; exceptions inside the interceptor are caught and logged, never break the call.

When `reflection=True`, we call `grpc_reflection.v1alpha.reflection.enable_server_reflection(service_names, server)`. `grpc-reflection` is an optional extra — if the module isn't importable, raise `ConfigurationError` with a clear install hint (`pip install hawkapi[grpc]` — which pulls `grpcio` + `grpcio-reflection`).

`ssl_credentials` is a pure passthrough to `server.add_secure_port(address, ssl_credentials)`. HawkAPI doesn't load certs itself — the user constructs `grpc.ssl_server_credentials(...)` and hands it in. Documented with a short example in the guide.

No constructor changes. `mount_grpc` is a method; internal state lives on `app._grpc_mounts: list[GrpcMount]`. Startup/shutdown hooks are installed the first time `mount_grpc` is called.

Each file < 200 lines. All `grpc`, `grpc.aio`, `grpc_reflection`, protobuf imports are lazy (inside function bodies / inside `_mount.py` guarded by a single `_require_grpc()` helper).

- **Runtime:** none by default. `grpcio>=1.60` and `grpcio-reflection>=1.60` as optional extras under `pyproject.toml` `[project.optional-dependencies].grpc`.

- **Dev/test:** add `grpcio`, `grpcio-reflection`, `protobuf` to the test dependency group so CI runs the real-server tests.

Target ~18 tests. Use a tiny hand-written servicer + `grpc.aio.insecure_channel` client (no `protoc` needed — skip codegen by defining servicer against a hand-rolled `_pb2_grpc.py` fixture that registers a method by name through `grpc.method_handlers_generic_handler`). All tests guarded by `pytest.importorskip("grpc")`.

- Unary call → returns expected payload.

- `context.hawkapi_app` is the right app instance.

- `context.hawkapi_request_id` is a valid uuid4 string.

- Observability interceptor increments Prometheus counter.

- Observability interceptor logs entry + exit.

- User-provided interceptor runs (appended after built-in).

- User interceptor runs in declared order (for 2 user interceptors).

- Error in handler → gRPC status `INTERNAL`, counter labelled with the code.

- `autostart=False` → server not listening until `await mount.start()`.

- Manual `.start()` / `.stop()` happy path.

- Lifespan startup binds port; shutdown releases it (assert via second bind attempt).

- `ssl_credentials=...` → `add_secure_port` path exercised (use a self-signed pair from a fixture).

- `reflection=True` with module present → `ServerReflection/ServerReflectionInfo` responds.

- `reflection=True` without module → `ConfigurationError` with install hint.

- Multiple `mount_grpc` calls with different ports → both servers start.

- Multiple `mount_grpc` calls with the same port → servicers merged on one server.

- Streaming call (server-streaming) works end-to-end.

- Integration: HTTP request handled by HawkAPI and gRPC call served concurrently in the same process.

Tests that need `grpcio-reflection` use a second `pytest.importorskip`.

Covers:

- Quickstart with `protoc`-generated stubs.

- `add_to_server` wiring + why we pass it explicitly (no introspection magic).

- ASGI lifespan integration and running under uvicorn/granian.

- Accessing `app` from servicer methods.

- TLS example (`grpc.ssl_server_credentials`).

- Reflection toggle (+ grpcurl/Postman hint).

- Observability — what metrics/logs ship and where they land.

- Roadmap: Connect/gRPC-Web adapter, `Depends()` resolution, streaming interceptors.

- `mkdocs.yml`: `- gRPC: guide/grpc.md` after GraphQL, before Bulkhead.

- `CHANGELOG.md`: one `[Unreleased] ### Added` bullet.

- `pyproject.toml`: add `[project.optional-dependencies] grpc = ["grpcio>=1.60", "grpcio-reflection>=1.60"]`.

- **`Depends()` resolution in servicer methods** — v2. Users use `context.hawkapi_app` for now.

- **Connect / gRPC-Web over ASGI** — v2. Separate mount, shares interceptors.

- **gRPC-Web gateway** — v2. Translates gRPC-Web → native gRPC.

- **Auto-generating servicer from annotations** — v3+, needs heavy design.

- **Built-in auth interceptors (JWT, OAuth2)** — users compose with existing `hawkapi.security` helpers + custom interceptor in v1.

- **Client-side gRPC helpers** — out of scope; users use `grpc.aio.insecure_channel` directly.

- **Load balancing / service mesh primitives** — deployment concern, not framework.

1. `app.mount_grpc(servicer, add_to_server=..., port=50051)` starts server on ASGI startup and stops on shutdown.

2. `context.hawkapi_app` is the HawkAPI instance inside every handler.

3. Observability interceptor emits Prometheus metrics on the shared registry.

4. `reflection=True` enables `grpcurl list` / `grpcurl describe`.

5. `ssl_credentials=` passes through to `add_secure_port`.

6. `autostart=False` → manual `.start()` / `.stop()` work.

7. Zero runtime deps on the default path; clear error when extras missing.

8. Full suite + ruff + mkdocs strict clean.

- `src/hawkapi/grpc/__init__.py` — new

- `src/hawkapi/grpc/_mount.py` — new

- `src/hawkapi/grpc/_interceptor.py` — new

- `src/hawkapi/grpc/_reflection.py` — new

- `src/hawkapi/app.py` — `mount_grpc` method + lifespan hooks

- `src/hawkapi/__init__.py` — lazy re-exports (`GrpcMount`, `HawkAPIObservabilityInterceptor`)

- `tests/unit/test_grpc.py` — new

- `docs/guide/grpc.md` — new

- `mkdocs.yml` — nav entry

- `CHANGELOG.md` — bullet

- `pyproject.toml` — optional-dependencies entry

New module + new method + additive deps (optional extra). No existing paths change. Revert = delete `grpc/` package, revert one method on HawkAPI, revert four doc/meta diffs.