Python实现Golang gRPC PerRPCCredentials等效功能的问题

我完全懂你遇到的困扰——在Go里用PerRPCCredentials实现自定义授权顺风顺水，到Python这边不仅踩了凭证类型不匹配的坑，还得确保授权信息只在每次RPC调用（包括流式）时发一次，不是每个HTTP帧都重复发送。下面一步步帮你搞定：

1. 修复凭证类型不匹配的错误 #

你碰到的TypeError核心原因是：grpc.metadata_call_credentials()返回的是CallCredentials（仅针对单次RPC调用的凭证），但创建gRPC Channel需要的是ChannelCredentials类型。解决办法很简单——把调用级凭证和TLS通道凭证组合起来就行。

正确的完整代码实现 #

你的AuthMetadataPlugin逻辑是对的，只需要调整凭证的组合方式：

import grpc

class AuthMetadataPlugin:
    def __init__(self, email, api_key):
        self.email = email
        self.api_key = api_key
    def __call__(self, context, callback):
        # 这里的元数据会在每次RPC调用触发时发送一次
        callback((('email', self.email), ('api_key', self.api_key)), None)

def get_authenticated_channel(target, email, api_key):
    # 创建TLS基础通道凭证（如果用自签名证书，需要传入root_certificates参数）
    tls_creds = grpc.ssl_channel_credentials()
    # 生成调用级的元数据凭证
    call_creds = grpc.metadata_call_credentials(AuthMetadataPlugin(email, api_key))
    # 组合成Channel可接受的凭证类型
    composite_creds = grpc.composite_channel_credentials(tls_creds, call_creds)
    # 返回带授权的安全通道
    return grpc.secure_channel(target, composite_creds)

这样就能解决类型不匹配的问题，composite_channel_credentials会把TLS通道凭证和自定义调用凭证合并成合法的ChannelCredentials。

2. 确保授权信息仅每次RPC调用发送一次 #

你担心的“每个HTTP帧都发送”其实是多余的——gRPC的metadata_call_credentials设计就是每个RPC调用（包括流式RPC）只发送一次元数据：

• 对于普通一元RPC：元数据会在请求的初始HTTP/2 HEADERS帧里发送一次；
• 对于流式RPC（客户端流、服务器流、双向流）：元数据会在流建立的初始HEADERS帧里发送一次，后续的数据流帧不会重复携带这些授权信息，完全符合你的需求。

3. 确认Go服务器端能正常提取元数据 #

你的Go服务器端的StreamInterceptor逻辑不需要任何改动——Python客户端发送的email和api_key元数据会被gRPC自动映射到Go的上下文元数据中，用你原本的metadata.FromIncomingContext(ctx)就能正常提取，和处理Go客户端的逻辑完全一致。

给你贴个Go端提取的示例（方便你核对）：

func authInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    md, ok := metadata.FromIncomingContext(ss.Context())
    if !ok {
        return status.Errorf(codes.Unauthenticated, "未获取到元数据")
    }
    email := md.Get("email")
    apiKey := md.Get("api_key")
    // 这里执行你的授权验证逻辑
    if len(email) == 0 || len(apiKey) == 0 {
        return status.Errorf(codes.Unauthenticated, "缺少email或api_key")
    }
    return handler(srv, ss)
}

额外注意事项 #

• 如果你的gRPC服务器用的是自签名TLS证书，在Python客户端创建ssl_channel_credentials时需要传入根证书内容：

  with open("server_root_cert.pem", "rb") as f:
      root_cert = f.read()
  tls_creds = grpc.ssl_channel_credentials(root_certificates=root_cert)
  

• gRPC元数据的key遵循HTTP/2规则，会被统一转为小写，所以建议Python端发送时直接用小写key，和Go端提取逻辑保持一致，避免不必要的问题。

内容的提问来源于stack exchange，提问作者Dan Jenson