Implementation:Bentoml BentoML GRPC Utils

Knowledge Sources	Bentoml_BentoML
Domains	gRPC, Utilities, Protocol Buffers
Last Updated	2026-02-13 15:00 GMT

Overview

Provides utility functions and types for BentoML's gRPC infrastructure, including status code conversion, method name parsing, proto field validation, and RPC handler wrapping.

Description

This module serves as the shared utility layer for all gRPC-related code in BentoML. Key components include:

• GRPC_CONTENT_TYPE -- The constant "application/grpc" used as the content-type identifier.
• grpc_status_code -- Converts BentoMLException.error_code HTTP status to the equivalent grpc.StatusCode.
• to_http_status -- Converts grpc.StatusCode to an HTTP status integer.
• http_status_to_grpc_status_map and grpc_status_to_http_status_map -- Cached bidirectional mappings between HTTP and gRPC status codes.
• parse_method_name -- Parses gRPC method strings (e.g., "/package.ServiceName/MethodName") into a MethodName dataclass with package, service, and method fields.
• validate_proto_fields -- Validates that a proto field name is accepted by a given IODescriptor.
• wrap_rpc_handler -- Applies a wrapper function to the appropriate handler method (unary_unary, unary_stream, stream_unary, or stream_stream) on an RpcMethodHandler.
• load_from_file -- Reads binary content from a file path.
• Re-exports import_generated_stubs, import_grpc, and LATEST_PROTOCOL_VERSION from the import hook submodule.

Usage

Use these utilities throughout BentoML's gRPC interceptors, servicers, and client code. They provide essential conversions between gRPC and HTTP concepts and simplify handler wrapping for interceptor implementations.

Code Reference

Source Location

• Repository: Bentoml_BentoML
• File: src/bentoml/grpc/utils/__init__.py
• Lines: 1-174

Signature

GRPC_CONTENT_TYPE = "application/grpc"

def load_from_file(p: str) -> bytes: ...

def validate_proto_fields(field: str | None, io_: IODescriptor[t.Any]) -> str | ProtoField: ...

@lru_cache(maxsize=1)
def http_status_to_grpc_status_map() -> dict[Enum, grpc.StatusCode]: ...

@lru_cache(maxsize=1)
def grpc_status_to_http_status_map() -> dict[grpc.StatusCode, Enum]: ...

def grpc_status_code(err: BentoMLException) -> grpc.StatusCode: ...

def to_http_status(status_code: grpc.StatusCode) -> int: ...

@dataclass
class MethodName:
    package: str = ""
    service: str = ""
    method: str = ""
    @property
    def fully_qualified_service(self): ...

def parse_method_name(method_name: str) -> tuple[MethodName, bool]: ...

def wrap_rpc_handler(
    wrapper: t.Callable[...],
    handler: RpcMethodHandler | None,
) -> RpcMethodHandler | None: ...

Import

from bentoml.grpc.utils import (
    GRPC_CONTENT_TYPE,
    grpc_status_code,
    to_http_status,
    parse_method_name,
    wrap_rpc_handler,
    validate_proto_fields,
    import_generated_stubs,
    import_grpc,
    LATEST_PROTOCOL_VERSION,
    load_from_file,
)

I/O Contract

Inputs

Name	Type	Required	Description
err	BentoMLException	Yes (grpc_status_code)	Exception to convert to gRPC status
status_code	grpc.StatusCode	Yes (to_http_status)	gRPC status code to convert to HTTP status
method_name	str	Yes (parse_method_name)	Full gRPC method path (e.g., "/pkg.Svc/Method")
field	str or None	Yes (validate_proto_fields)	Proto field name to validate
io_	IODescriptor	Yes (validate_proto_fields)	IO descriptor defining accepted proto fields
wrapper	Callable	Yes (wrap_rpc_handler)	Function to wrap the RPC handler behavior
handler	RpcMethodHandler or None	Yes (wrap_rpc_handler)	The handler to wrap

Outputs

Name	Type	Description
grpc.StatusCode	grpc.StatusCode	Equivalent gRPC status code (grpc_status_code)
int	int	HTTP status code integer (to_http_status)
(MethodName, bool)	tuple	Parsed method name and validity flag (parse_method_name)
RpcMethodHandler or None	RpcMethodHandler	Wrapped handler or None (wrap_rpc_handler)

Usage Examples

from bentoml.grpc.utils import parse_method_name, to_http_status, grpc_status_code
from bentoml.exceptions import NotFound
import grpc

# Parse a gRPC method name
method, valid = parse_method_name("/bentoml.grpc.v1.BentoService/Call")
print(method.package)   # "bentoml.grpc.v1"
print(method.service)   # "BentoService"
print(method.method)    # "Call"
print(valid)            # True

# Convert gRPC status to HTTP
http_code = to_http_status(grpc.StatusCode.NOT_FOUND)
print(http_code)  # 404

# Convert BentoML exception to gRPC status
status = grpc_status_code(NotFound("not found"))
print(status)  # grpc.StatusCode.UNIMPLEMENTED

Related Pages

• Implementation:Bentoml_BentoML_GRPC_Access_Interceptor
• Implementation:Bentoml_BentoML_GRPC_OpenTelemetry_Interceptor
• Implementation:Bentoml_BentoML_GRPC_Prometheus_Interceptor
• Implementation:Bentoml_BentoML_Exceptions