如何从Python异步gRPC客户端获取初始和尾部元数据

客户端侧获取元数据 #

你提到异步版本的UnaryUnaryMultiCallable没有对应方法其实是误解，异步gRPC接口同样支持获取两类元数据，有两种常用实现方式：

方式1：先获取调用对象分步获取 #

__call__方法返回的对象同时实现了可等待协议和grpc.aio.Call接口，你不需要直接await拿到响应，先获取调用实例即可按需读取元数据：

import grpc
from your_proto import your_service_pb2, your_service_pb2_grpc

async def client_demo():
    async with grpc.aio.insecure_channel("127.0.0.1:50051") as channel:
        stub = your_service_pb2_grpc.YourServiceStub(channel)
        # 先实例化调用对象，不要直接await
        rpc_call = stub.YourRpcMethod(
            your_service_pb2.YourRequest(param="test"),
            # 此处可传入客户端要发送的元数据
            metadata=(("client-custom-key", "custom-value"),)
        )
        # 获取服务端返回的初始元数据
        initial_metadata = await rpc_call.initial_metadata()
        # 等待调用完成获取响应
        response = await rpc_call
        # 获取服务端返回的尾部元数据
        trailing_metadata = await rpc_call.trailing_metadata()
        # 额外也可以获取调用状态码、状态详情
        code = await rpc_call.code()
        details = await rpc_call.details()

方式2：使用with_call方法一键获取 #

异步UnaryUnaryMultiCallable同样提供with_call方法，await后直接返回「响应+调用对象」的元组，写法更简洁：

async def client_demo_simple():
    async with grpc.aio.insecure_channel("127.0.0.1:50051") as channel:
        stub = your_service_pb2_grpc.YourServiceStub(channel)
        # 直接调用with_call同时拿到响应和调用对象
        response, rpc_call = await stub.YourRpcMethod.with_call(
            your_service_pb2.YourRequest(param="test")
        )
        initial_metadata = await rpc_call.initial_metadata()
        trailing_metadata = await rpc_call.trailing_metadata()

两类元数据的返回格式都是tuple[Tuple[str, str]]，每个元素是键值对二元组，和同步版本格式完全一致。 #

服务端侧获取/设置元数据 #

如果是异步服务端需要读取客户端发送的初始元数据、或设置返回给客户端的元数据，直接操作grpc.aio.ServicerContext对象即可：

class YourServiceServicer(your_service_pb2_grpc.YourServiceServicer):
    async def YourRpcMethod(self, request, context: grpc.aio.ServicerContext):
        # 获取客户端发来的初始元数据
        client_metadata = context.invocation_metadata()
        # 设置服务端初始元数据，会优先返回给客户端
        await context.send_initial_metadata((("server-init-key", "init-value"),))
        # 业务逻辑
        res = your_service_pb2.YourResponse(result="success")
        # 设置服务端尾部元数据，会在调用结束时返回给客户端
        context.set_trailing_metadata((("server-trail-key", "trail-value"),))
        return res

内容的提问来源于stack exchange，提问作者A Poor