def create_bento_servicer(service: Service) -> services.BentoServiceServicer:
    """
    This is the actual implementation of BentoServicer.
    Main inference entrypoint will be invoked via /bentoml.grpc.<version>.BentoService/Call
    """

    class BentoServiceImpl(services.BentoServiceServicer):
        """An asyncio implementation of BentoService servicer."""

        async def Call(  # type: ignore (no async types) # pylint: disable=invalid-overridden-method
            self,
            request: pb.Request,
            context: BentoServicerContext,
        ) -> pb.Response | None:
            if request.api_name not in service.apis:
                raise InvalidArgument(
                    f"given 'api_name' is not defined in {service.name}",
                ) from None

            api = service.apis[request.api_name]
            response = pb.Response()
            output = None

            # NOTE: since IODescriptor.proto_fields is a tuple, the order is preserved.
            # This is important so that we know the order of fields to process.
            # We will use fields descriptor to determine how to process that request.
            try:
                # we will check if the given fields list contains a pb.Multipart.
                input_proto = getattr(
                    request,
                    validate_proto_fields(request.WhichOneof("content"), api.input),
                )
                input_data = await api.input.from_proto(input_proto)
                if is_async_callable(api.func):
                    if api.multi_input:
                        output = await api.func(**input_data)
                    else:
                        output = await api.func(input_data)
                else:
                    if api.multi_input:
                        output = await anyio.to_thread.run_sync(api.func, **input_data)
                    else:
                        output = await anyio.to_thread.run_sync(api.func, input_data)

                res = await api.output.to_proto(output)
                # TODO(aarnphm): support multiple proto fields
                response = pb.Response(**{api.output.proto_fields[0]: res})
            except BentoMLException as e:
                log_exception(request, sys.exc_info())
                if output is not None:
                    import inspect

                    signature = inspect.signature(api.output.to_proto)
                    param = next(iter(signature.parameters.values()))
                    ann = ""
                    if param is not inspect.Parameter.empty:
                        ann = param.annotation

                    # more descriptive errors if output is available
                    logger.error(
                        "Function '%s' has 'input=%s,output=%s' as IO descriptor, and returns 'result=%s', while expected return type is '%s'",
                        api.name,
                        api.input,
                        api.output,
                        type(output),
                        ann,
                    )
                await context.abort(code=grpc_status_code(e), details=e.message)
            except (RuntimeError, TypeError, NotImplementedError):
                log_exception(request, sys.exc_info())
                await context.abort(
                    code=grpc.StatusCode.INTERNAL,
                    details="A runtime error has occurred, see stacktrace from logs.",
                )
            except Exception:  # pylint: disable=broad-except
                log_exception(request, sys.exc_info())
                await context.abort(
                    code=grpc.StatusCode.INTERNAL,
                    details="An error has occurred in BentoML user code when handling this request, find the error details in server logs.",
                )
            return response

        async def ServiceMetadata(  # type: ignore (no async types) # pylint: disable=invalid-overridden-method
            self,
            request: pb.ServiceMetadataRequest,  # pylint: disable=unused-argument
            context: BentoServicerContext,  # pylint: disable=unused-argument
        ) -> pb.ServiceMetadataResponse:
            return pb.ServiceMetadataResponse(
                name=service.name,
                docs=service.doc,
                apis=[
                    pb.ServiceMetadataResponse.InferenceAPI(
                        name=api.name,
                        docs=api.doc,
                        input=make_descriptor_spec(
                            api.input.to_spec(), pb.ServiceMetadataResponse
                        ),
                        output=make_descriptor_spec(
                            api.output.to_spec(), pb.ServiceMetadataResponse
                        ),
                    )
                    for api in service.apis.values()
                ],
            )

    return BentoServiceImpl()
This is the actual implementation of BentoServicer.
Main inference entrypoint will be invoked via /bentoml.grpc.<version>.BentoService/Call
src/bentoml/_internal/server/grpc/servicer/v1/__init__.py