yop-mcp

Instructions

Implementation Reference

• yop_mcp/main.py:216-249 (handler)

  Handler function for the MCP tool 'yeepay_yop_download_cert'. Decorated with @mcp.tool() for registration. Defines input parameters with type hints and detailed docstring serving as schema. Delegates core logic to download_cert helper.

  def yeepay_yop_download_cert(  # pylint: disable=too-many-arguments,too-many-positional-arguments
      algorithm: str = "RSA",
      serial_no: str = "",
      auth_code: str = "",
      private_key: str = "",
      public_key: str = "",
      pwd: str = "",
  ) -> Dict[str, Any]:
      """
      根据密钥算法、CFCA证书的序列号、授权码、非对称密钥对（公钥和私钥）、密码，下载该证书，并保存到本地路径
  
      Args:
          algorithm: 密钥算法，可选值为 "RSA" 或 "SM2"，默认为 "RSA"
          serial_no: cfca证书序列号
          auth_code: cfca证书授权码
          private_key: Base64 编码后的私钥字符串
          public_key: Base64 编码后的公钥字符串
          pwd: 密码，长度：12~16位
  
      Returns:
          Dict包含:
          - message: 响应信息
          - pfxCert: 私钥证书路径(.pfx)
          - pubCert: 公钥证书路径(.cer)
      """
      return download_cert(
          algorithm=algorithm,
          serial_no=serial_no,
          auth_code=auth_code,
          private_key=private_key,
          public_key=public_key,
          pwd=pwd,
      )

• tools/cert_utils.py:460-548 (helper)

  Core helper function implementing the certificate download logic: input validation, key pair verification, P10 certificate request generation, download from CFCA API, certificate verification, and file saving.

  def download_cert(
      algorithm: str = "RSA",
      serial_no: str = "",
      auth_code: str = "",
      private_key: str = "",
      public_key: str = "",
      pwd: str = "",
  ) -> Dict[str, Any]:
      # 确定密钥类型
      key_type = KeyType.SM2 if algorithm.upper() == "SM2" else KeyType.RSA2048
  
      # 检查输入参数
      check_result = CertUtils.check_input(
          serial_no, auth_code, key_type, private_key, public_key, pwd
      )
      if not check_result.result:
          return {"message": check_result.msg}
  
      # 检查公私钥匹配
      p10_generated = False  # 标记是否已生成P10请求
      try:
          if not p10_generated and not CertUtils.check_key(
              private_key, public_key, key_type
          ):
              return {"message": "商户公私钥不匹配，请重新输入"}
      except Exception as e:
          return {"message": f"密钥解析异常: {str(e)}"}
  
      # 生成证书请求
      if p10_generated:
          cert_req = private_key
      else:
          try:
              cert_req = CertUtils.gen_p10(private_key, public_key, key_type)
          except Exception as e:
              return {"message": f"生成证书请求失败: {str(e)}"}
  
      # 确定证书保存路径
      cert_path = (
          Config.SM2_CERT_SAVE_PATH
          if key_type == KeyType.SM2
          else Config.RSA_CERT_SAVE_PATH
      )
      pri_cert_path = os.path.join(cert_path, f"{serial_no}.pfx")
      pub_cert_path = os.path.join(cert_path, f"{serial_no}.cer")
  
      # 检查证书是否已存在
      if SupportUtil.is_file_exists(pri_cert_path) and SupportUtil.is_file_exists(
          pub_cert_path
      ):
          return {
              "message": "本地证书已存在",
              "pfxCert": pri_cert_path,
              "pubCert": pub_cert_path,
          }
  
      try:
          # 获取证书
          cert: Optional[str] = None
          if SupportUtil.is_file_exists(pub_cert_path):
              cert = SupportUtil.read_file_as_string(pub_cert_path)
          else:
              cert_download_result = CertUtils.download_cert_from_cfca(
                  serial_no, auth_code, cert_req
              )
              if cert_download_result.error_msg:
                  return {"message": cert_download_result.error_msg}
              cert = cert_download_result.cert
  
          # 检查证书与私钥匹配
          if cert and not CertUtils.check_cert(private_key, cert, key_type):
              return {"message": "证书已下载过，且证书与输入的私钥不匹配，请核对"}
  
          # 保存证书
          if cert:
              pub_cert_path = CertUtils.make_pub_cert(cert, serial_no, cert_path)
          if not p10_generated and cert:
              pri_cert_path = CertUtils.make_pfx_cert(
                  private_key, cert, key_type, pwd, serial_no, cert_path
              )
  
          return {
              "message": "CFCA证书激活并下载成功",
              "pfxCert": pri_cert_path,
              "pubCert": pub_cert_path,
          }
      except Exception as e:
          return {"message": f"系统异常，请稍后重试: {str(e)}"}