[Serve] Add gRPC util
diff --git a/python/ray/serve/BUILD b/python/ray/serve/BUILD
index 68c057e623b2b..5cd110495c7e5 100644
--- a/python/ray/serve/BUILD
+++ b/python/ray/serve/BUILD
@@ -129,6 +129,14 @@ py_test(
     deps = [":serve_lib"],
 )
 
+py_test(
+    name = "test_grpc_util",
+    size = "small",
+    srcs = serve_tests_srcs,
+    tags = ["exclusive", "team:serve"],
+    deps = [":serve_lib"],
+)
+
 py_test(
     name = "test_advanced",
     size = "small",
diff --git a/python/ray/serve/_private/grpc_util.py b/python/ray/serve/_private/grpc_util.py
new file mode 100644
index 0000000000000..a47031d608f5c
--- /dev/null
+++ b/python/ray/serve/_private/grpc_util.py
@@ -0,0 +1,80 @@
+import grpc
+from typing import Sequence
+from grpc.aio._server import Server
+
+
+class gRPCServer(Server):
+    """Custom gRPC server to override gRPC method methods.
+
+    Original implementation see: https://github.com/grpc/grpc/blob/
+        60c1701f87cacf359aa1ad785728549eeef1a4b0/src/python/grpcio/grpc/aio/_server.py
+    """
+
+    def __init__(self, service_handler_factory, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.service_handler_factory = service_handler_factory
+        self.generic_rpc_handlers = []
+
+    def add_generic_rpc_handlers(
+        self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]
+    ):
+        """Override generic_rpc_handlers before adding to the gRPC server.
+
+        This function will override all user defined handlers to have
+            1. None `response_serializer` so the server can pass back the
+            raw protobuf bytes to the user.
+            2. `unary_unary` is always calling the unary function generated via
+            `self.service_handler_factory`
+            3. `unary_stream` is always calling the streaming function generated via
+            `self.service_handler_factory`
+        """
+        serve_rpc_handlers = {}
+        rpc_handler = generic_rpc_handlers[0]
+        for service_method, method_handler in rpc_handler._method_handlers.items():
+            serve_method_handler = method_handler._replace(
+                response_serializer=None,
+                unary_unary=self.service_handler_factory(
+                    service_method=service_method,
+                    stream=False,
+                ),
+                unary_stream=self.service_handler_factory(
+                    service_method=service_method,
+                    stream=True,
+                ),
+            )
+            serve_rpc_handlers[service_method] = serve_method_handler
+        generic_rpc_handlers[0]._method_handlers = serve_rpc_handlers
+        self.generic_rpc_handlers.append(generic_rpc_handlers)
+        super().add_generic_rpc_handlers(generic_rpc_handlers)
+
+
+def create_serve_grpc_server(service_handler_factory):
+    """Custom function to create Serve's gRPC server.
+
+    This function works similar to `grpc.server()`, but it creates a Serve defined
+    gRPC server in order to override the `unary_unary` and `unary_stream` methods
+
+    See: https://grpc.github.io/grpc/python/grpc.html#grpc.server
+    """
+    return gRPCServer(
+        thread_pool=None,
+        generic_handlers=(),
+        interceptors=(),
+        options=(),
+        maximum_concurrent_rpcs=None,
+        compression=None,
+        service_handler_factory=service_handler_factory,
+    )
+
+
+class DummyServicer:
+    """Dummy servicer for gRPC server to call on.
+
+    This is a dummy class that just pass through when calling on any method.
+    User defined servicer function will attempt to add the method on this class to the
+    gRPC server, but our gRPC server will override the caller to call gRPCProxy.
+    """
+
+    def __getattr__(self, attr):
+        # No-op pass through. Just need this to act as the callable.
+        pass
diff --git a/python/ray/serve/tests/test_grpc_util.py b/python/ray/serve/tests/test_grpc_util.py
new file mode 100644
index 0000000000000..4ef0b2f60bc30
--- /dev/null
+++ b/python/ray/serve/tests/test_grpc_util.py
@@ -0,0 +1,109 @@
+import pytest
+from typing import Callable
+import grpc
+from google.protobuf.any_pb2 import Any as AnyProto
+
+from ray.serve._private.grpc_util import (
+    create_serve_grpc_server,
+    DummyServicer,
+    gRPCServer,
+)
+
+
+def fake_service_handler_factory(service_method: str, stream: bool) -> Callable:
+    def foo() -> bytes:
+        return f"{'stream' if stream else 'unary'} call from {service_method}".encode()
+
+    return foo
+
+
+def test_dummy_servicer_can_take_any_methods():
+    """Test an instance of DummyServicer can be called with any method name without
+    error.
+
+    When dummy_servicer is called with any custom defined methods, it won't raise error.
+    """
+    dummy_servicer = DummyServicer()
+    dummy_servicer.foo
+    dummy_servicer.bar
+    dummy_servicer.baz
+    dummy_servicer.my_method
+    dummy_servicer.Predict
+
+
+def test_create_serve_grpc_server():
+    """Test `create_serve_grpc_server()` creates the correct server.
+
+    The server created by `create_serve_grpc_server()` should be an instance of
+    Serve defined `gRPCServer`. Also, the handler factory passed with the function
+    should be used to initialize the `gRPCServer`.
+    """
+    grpc_server = create_serve_grpc_server(
+        service_handler_factory=fake_service_handler_factory
+    )
+    assert isinstance(grpc_server, gRPCServer)
+    assert grpc_server.service_handler_factory == fake_service_handler_factory
+
+
+def test_grpc_server():
+    """Test `gRPCServer` did the correct overrides.
+
+    When a add_servicer_to_server function is called on an instance of `gRPCServer`,
+    it correctly overrides `response_serializer` to None, and `unary_unary` and
+    `unary_stream` to be generated from the factory function.
+    """
+    service_name = "ray.serve.ServeAPIService"
+    method_name = "ServeRoutes"
+
+    def add_test_servicer_to_server(servicer, server):
+        rpc_method_handlers = {
+            method_name: grpc.unary_unary_rpc_method_handler(
+                servicer.ServeRoutes,
+                request_deserializer=AnyProto.FromString,
+                response_serializer=AnyProto.SerializeToString,
+            ),
+        }
+        generic_handler = grpc.method_handlers_generic_handler(
+            service_name, rpc_method_handlers
+        )
+        server.add_generic_rpc_handlers((generic_handler,))
+
+    grpc_server = gRPCServer(
+        thread_pool=None,
+        generic_handlers=(),
+        interceptors=(),
+        options=(),
+        maximum_concurrent_rpcs=None,
+        compression=None,
+        service_handler_factory=fake_service_handler_factory,
+    )
+    dummy_servicer = DummyServicer()
+
+    # Ensure `generic_rpc_handlers` is not populated before calling
+    # the add_servicer_to_server function.
+    assert grpc_server.generic_rpc_handlers == []
+
+    add_test_servicer_to_server(dummy_servicer, grpc_server)
+
+    # `generic_rpc_handlers` should be populated after add_servicer_to_server is called.
+    assert len(grpc_server.generic_rpc_handlers) == 1
+
+    # The populated rpc handler should have the correct service name.
+    rpc_handler = grpc_server.generic_rpc_handlers[0][0]
+    assert rpc_handler.service_name() == service_name
+
+    # The populated method handlers should have the correct response_serializer,
+    # unary_unary, and unary_stream.
+    service_method = f"/{service_name}/{method_name}"
+    method_handlers = rpc_handler._method_handlers.get(service_method)
+    assert method_handlers.response_serializer is None
+    assert method_handlers.unary_unary() == f"unary call from {service_method}".encode()
+    assert (
+        method_handlers.unary_stream() == f"stream call from {service_method}".encode()
+    )
+
+
+if __name__ == "__main__":
+    import sys
+
+    sys.exit(pytest.main(["-v", "-s", __file__]))

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

Add Serve's custom logics to help setup gRPC server.

## Related issue number

Forth PR to close https://github.com/ray-project/ray/issues/38193

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [ ] I've added any new APIs to the API Reference. For example, if I added a 
           method in Tune, I've added it in `doc/source/tune/api/` under the 
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
