userver: gRPC middleware
No Matches
gRPC middleware

Before you start

Make sure that you can compile and run core tests and read a basic example Writing your first HTTP server.

Step by step guide

In this example, you will write an authentication middleware for both 'GreeterService' and 'GreeterClient' of the basic grpc_service. See gRPC client and service


Generate and link with necessary libraries:

userver_add_grpc_library(${PROJECT_NAME}-proto PROTOS samples/greeter.proto)
target_link_libraries(${PROJECT_NAME} ${PROJECT_NAME}-proto)

The client middleware

Client middleware will add metadata to every GreeterClient call.

Derive Middleware and MiddlewareFactory from the respective base class:

class Middleware final : public ugrpc::client::MiddlewareBase {
explicit Middleware();
~Middleware() override;
void Handle(ugrpc::client::MiddlewareCallContext& context) const override;
class MiddlewareFactory final : public ugrpc::client::MiddlewareFactoryBase {
explicit MiddlewareFactory(const components::ComponentContext& context);
~MiddlewareFactory() override;
std::shared_ptr<const ugrpc::client::MiddlewareBase> GetMiddleware(
std::string_view) const override;

Handle method of Middleware does the actual work:

void ApplyCredentials(::grpc::ClientContext& context) {
context.AddMetadata(kKey, kCredentials);
Middleware::Middleware() = default;
Middleware::~Middleware() = default;
void Middleware::Handle(ugrpc::client::MiddlewareCallContext& context) const {
MiddlewareFactory::MiddlewareFactory(const components::ComponentContext&) {}
MiddlewareFactory::~MiddlewareFactory() = default;
std::shared_ptr<const Middleware::MiddlewareBase>
MiddlewareFactory::GetMiddleware(std::string_view) const {
return std::make_shared<Middleware>();

Then, wrap it into component, which just stores MiddlewareFactory:

class Component final : public ugrpc::client::MiddlewareComponentBase {
static constexpr const char* kName = "grpc-auth-client";
Component(const components::ComponentConfig& config,
std::shared_ptr<const ugrpc::client::MiddlewareFactoryBase>
GetMiddlewareFactory() override;
std::shared_ptr<ugrpc::client::MiddlewareFactoryBase> factory_;

Lastly, add this component to the static config:

# yaml

And connect it with ClientFactory:

# yaml
task-processor: grpc-blocking-task-processor
channel-args: {}
- grpc-auth-client

The server middleware

Server middleware, in its turn, will validate metadata that comes with an rpc.

Everything is the same as it is for client middleware, except there is no factory and the component stores the middleware itself:

class Middleware final : public ugrpc::server::MiddlewareBase {
explicit Middleware();
void Handle(ugrpc::server::MiddlewareCallContext& context) const override;

Handle method of Middleware does the actual work:

Middleware::Middleware() = default;
void Middleware::Handle(ugrpc::server::MiddlewareCallContext& context) const {
const auto& metadata = context.GetCall().GetContext().client_metadata();
auto it = metadata.find(kKey);
if (it == metadata.cend() || it->second != kCredentials) {
auto& rpc = context.GetCall();
"Invalid credentials"});
LOG_ERROR() << "Invalid credentials";

Respective component:

class Component final : public ugrpc::server::MiddlewareComponentBase {
static constexpr std::string_view kName = "grpc-auth-server";
Component(const components::ComponentConfig&,
std::shared_ptr<ugrpc::server::MiddlewareBase> GetMiddleware() override;
std::shared_ptr<ugrpc::server::MiddlewareBase> middleware_;

Lastly, add this component to the static config:

# yaml

And connect it with Service:

# yaml
task-processor: main-task-processor
greeting-prefix: Hello
- grpc-auth-server

int main()

Finally, register components and start the server.

int main(int argc, char* argv[]) {
const auto component_list =
return utils::DaemonMain(argc, argv, component_list);

Build and Run

To build the sample, execute the following build steps at the userver root directory:

mkdir build_release
cd build_release
cmake -DCMAKE_BUILD_TYPE=Release ..
make userver-samples-grpc_middleware_service

The sample could be started by running make start-userver-samples-grpc_middleware_service. The command would invoke testsuite start target that sets proper paths in the configuration files and starts the service.

To start the service manually run ./samples/grpc_middleware_service/userver-samples-grpc_middleware_service -c </path/to/static_config.yaml>.

The service is available locally at port 8091 (as per project static_config.yaml).

Functional testing

To implement Functional tests for the service some preparational steps should be done.


First of all, import the required modules and add the required pytest_userver.plugins.grpc pytest plugin:

import pytest
import samples.greeter_pb2_grpc as greeter_services # noqa: E402, E501
pytest_plugins = ['pytest_userver.plugins.grpc']

gRPC server mock

To mock the gRPC server provide a hook for the static config to change the endpoint:

USERVER_CONFIG_HOOKS = ['prepare_service_config']
def prepare_service_config(grpc_mockserver_endpoint):
def patch_config(config, config_vars):
components = config['components_manager']['components']
components['greeter-client']['endpoint'] = grpc_mockserver_endpoint
return patch_config

Write the mocking fixtures using grpc_mockserver:

@pytest.fixture(name='mock_grpc_greeter_session', scope='session')
def _mock_grpc_greeter_session(grpc_mockserver, create_grpc_mock):
mock = create_grpc_mock(greeter_services.GreeterServiceServicer)
mock.servicer, grpc_mockserver,
return mock
def mock_grpc_greeter(mock_grpc_greeter_session):
with mock_grpc_greeter_session.mock() as mock:
yield mock

gRPC client

To do the gRPC requests write a client fixture using grpc_channel:

def grpc_client(grpc_channel):
return greeter_services.GreeterServiceStub(grpc_channel)

Use it to do gRPC requests to the service:

async def test_correct_credentials(grpc_client):
request = greeter_protos.GreetingRequest(name='Python')
response = await grpc_client.SayHello(
request=request, metadata=[('x-key', 'secret-credentials')],
assert response.greeting == 'Hello, Python!'
async def test_incorrect_credentials(grpc_client):
request = greeter_protos.GreetingRequest(name='Python')
await grpc_client.SayHello(
request=request, metadata=[('x-key', 'secretcredentials')],
assert False
except AioRpcError as err:
assert err.code() == StatusCode.PERMISSION_DENIED
async def test_no_credentials(grpc_client):
request = greeter_protos.GreetingRequest(name='Python')
await grpc_client.SayHello(request=request)
assert False
except AioRpcError as err:
assert err.code() == StatusCode.PERMISSION_DENIED

Full sources

See the full example at: