扣子Coze实战—做一个自己的公众号助手！涉及数据安全，你自己看着办

痛点在开发工作流的时候，经常会遇到这样的问题

没有合适的插件解决当前的场景
有合适的插件，但人家收费
有合适的插件，但人家要你密码

这时候我们就要自己搞一个插件了。大部分同学遇到这个情况就怕了

我不会写代码啊，我怎么开发自己的插件啊

本文就解决这个问题，有 AI 工具，不用懂代码，也可以开发自己的插件，以微信公众号信息同步为例，我们来实现一个将文档同步到微信公众号的插件。
扣子Coze实战—做一个自己的公众号助手！涉及数据安全，你自己看着办需求梳理如果有现成的插件（但人家要你本金的情况）那就很容易梳理需求了。如果没有就基于你想要的效果，梳理出你的需求。去问AI，有没有实现你当前这个诉求的开源包，然后基于开源包来封装插件。
就我们今天这个场景，还是有很多类似的插件工具的

比如这个工具，他提供了 5 个能力

wx_access_token
配置 app_id/app_secret来获取token ，简单说，就是让微信知道你登录了，这也是把这个交给三方不安全的地方，他可以用这个去操作你的用户，当然我相信上边这个作者应该是不会这么做的（逃。
wx_upload_media
上传素材，输入上一步获取到的token，和你要上传的图片，这个图片应该是网图的链接（注意版权）
或者你的上传插件获取到的图片链接（你要没有，我再出一个自建上传插件的文章，挖坑啊~~哇呀~哇呀~哇)

wx_markdown_to_html
将生成的文本带格式转成微信公众号认识的样子，给到后边发布用

wx_upload_draft
上传草稿，这个参数比较多（加星的就是一定要传入的），逐个解析一下

access_token: 第一步获取到的 token
title，文章标题，对应 AI 生成的有吸引力的标题
digest 摘要、副标题，对应的是 AI 生成的 SEO 优化关键词
author，这个就是文章作者，但这个微信是否认还是要打个问号的
Content 就是 wx_markdown_to_html 生成的 HTML，引入一下即可
content_source_url，文章源链，原创不需要
need_open_comment，如图所述，我们搞个默认值，别必填了，1
only_fans_can_comment，如图所述，默认 0
thumb_media_id: 封面的 media_id，这个 media_id哪儿来的，应该就是wx_uplaod_media返回的，这里不设置，去人肉设置也没啥问题

wx_publish_draft
发布草稿，这个风险比较高，我们就不用了，还是去线上人肉发，你要是想搞，自己去搞一个就好了。

开始上手哦了，经过上边一番分析，把这个需求丢给 AI （如果你有 Cursor/Trae 返工率会低一点，豆包或者 deepseek 这种通用模型也凑合用），让他给你写插件代码，环境要按照 Coze 能够支持的模式来写。

获取 Token

进来以后是空的，先创建一个工具wx_access_token

创建好了以后，切换到 “元数据” 这部分

将我们上边分析的输入和输出创建进去，点击编辑、输入，保存即可。这里的参数名就是未来代码里面读取的字段，要符合创建规范。（这里涉及到一丢丢代码知识…

切回代码，把 AI 提供给你的脚本贴在代码中，我这个是豆包给我的def handler(args: Args[Input]): """ 获取微信公众号的access_token


  参数:
    app_id: 微信公众号的appid
    app_secret: 微信公众号的appsecret
  返回:
    dict: 包含access_token和过期时间的字典
       成功时格式: {"access_token": "xxx", "expires_in": 7200}
       失败时格式: {"error": "错误信息", "errcode": 错误码, "errmsg": "错误描述"}
  """
  # 微信获取access_token的API地址
  url = "https://api.weixin.qq.com/cgi-bin/token"
  # 请求参数
  params = {
    "grant_type": "client_credential", # 固定值
    "appid": args.input.app_id,
    "secret": args.input.app_secret
  }
  try:
    # 发送GET请求
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status() # 抛出HTTP请求异常
    # 解析响应结果
    result = response.json()
    # 检查是否包含错误信息
    if "errcode" in result and result["errcode"] != 0:
      return {
        "access_token": "",
        "message": result["errmsg"]
      }
    # 成功返回结果
    return {
      "access_token": result["access_token"],
      "message": result["expires_in"]
    }

except requests.exceptions.RequestException as e: # 处理网络请求异常 return { "access_token":"", "message": f"网络请求异常: {str(e)}" } except json.JSONDecodeError: # 处理JSON解析异常 return { "access_token":"", "message": "无法解析API返回的JSON数据" } except Exception as e: # 处理其他未知异常 return { "access_token":"", "message": "未知错误" }

其中需要的 app_id和 app_secret应该在微信公众号后台获取

跑一下测试（这里可能会遇到各种报错，直接丢给 AI，让他帮你解决

填入刚才获取到的 IP

测试通过了以后，就可以把这个功能发布掉了

这里有个坑，Coze 的在线 IDE 是没有固定 IP 的，我是用的本地部署的 Coze，如果是在线使用，就需要一个跳板机，然后把跳板机 IP 配置给微信公众号后端就好了。（如果你不知道如何搭建一个跳板机，我再出一个自建跳板机的文章，挖坑啊~~哇呀~哇呀~哇)

上传素材

我们跳过重复的动作，回到刚才你创建的插件，在右上角继续“创建工具”，如果你没有退出在线 IDE，直接左上角点下小加号，就可以继续新建工具了。

这个过程中难免涉及到一些代码的调试手段，这部分也是有点挑战性的，我先把我的代码给到你from runtime import Args from typings.wx_upload_media.wx_upload_media import Input, Output import requests import io


"""
Each file needs to export a function named `handler`. This function is the entrance to the Tool.
Parameters:
args: parameters of the entry function.
args.input - input parameters, you can get test input value by args.input.xxx.
args.logger - logger instance used to print logs, injected by runtime.
Remember to fill in input/output in Metadata, it helps LLM to recognize and use tool.
Return:
The return data of the function, which should match the declared output parameters.
"""
def handler(args: Args[Input])->Output:
  """
  从网络URL下载图片并上传至微信公众号素材库，返回media_id
  参数:
    access_token: 微信公众号的访问令牌
    image_url: 网络图片的URL地址（需可直接访问，支持http/https）
  返回:
    dict: 包含上传结果的字典
       成功时格式: {"media_id": "素材ID", "message": "上传成功", "url": "图片URL"}
       失败时格式: {"media_id": "", "message": "错误信息", "errcode": "错误码", "errmsg": "错误描述"}
  """
  # 微信上传图片素材的API地址
  upload_url = f"https://api.weixin.qq.com/cgi-bin/media/uploadimg?access_token={args.input.access_token}"
  try:
    # 步骤1: 从image_url下载图片内容
    response = requests.get(args.input.image_url, timeout=15)
    response.raise_for_status() # 抛出HTTP错误
    # 检查响应内容是否为图片（简单验证Content-Type）
    content_type = response.headers.get("Content-Type", "")
    if not content_type.startswith(("image/jpeg", "image/png", "image/gif")):
      return {
        "media_id": "",
        "message": f"URL指向的内容不是有效图片，Content-Type: {content_type}",        
      }
    # 步骤2: 构造multipart/form-data请求上传至微信
    # 将图片内容转为文件流
    image_file = io.BytesIO(response.content)
    files = {"media": ("image.jpg", image_file, content_type)}
    # 发送上传请求
    upload_response = requests.post(upload_url, files=files, timeout=20)
    upload_result = upload_response.json()
    # 步骤3: 处理上传结果
    if "errcode" in upload_result and upload_result["errcode"] != 0:
      return {
        "media_id": "",
        "message": f"上传失败：{upload_result['errmsg']}",        
      }
    # 成功返回结果（微信API返回的url为图片永久链接）
    return {
      "media_id": upload_result["media_id"],
      "message": "图片上传成功",      
    }

except requests.exceptions.RequestException as e: # 处理网络请求异常（下载图片或上传过程失败） return { "media_id": "", "message": f"网络请求失败：{str(e)}", } except Exception as e: # 处理其他未知异常 return { "media_id": "", "message": f"处理失败：{str(e)}", }

其实这段代码实现的功能就是微信公众号提供的 API 能力，也没多啥，你可以直接去微信公众号后台拿到这样的一些 API 调用示例，然后把这个调用示例给 AI，让他按照这个接口的样式来写

这里有个坑是默认只支持 jpeg 而不支持 png，我就不知道为啥了。

上传草稿

跟上边一样，直接上代码

上下滑动查看完整内容

# 用户编码规范与重构指南
## 基础交互规则1. 请用中文回答我2. 如果回答的是代码，请给每个关键节点，比较难懂的代码增加中文注释3. 当生成的代码行数超过 20 行时，请考虑聚合代码以及考虑其颗粒度是否适合## 代码质量与重构规范### 通用编码规范1. 避免不必要的对象复制或克隆2. 避免多层嵌套，提前返回3. 使用适当的并发控制机制## 代码坏味道识别与处理基于Martin Fowler《重构》一书的核心观点，以下是应当注意的代码坏味道及其处理方法：### 1. 神秘命名- **问题**：变量、函数、类或模块的名称不能清晰表达其用途和含义- **处理**：重命名为具有描述性的名称，使代码自解释- **示例**：将`fn p()`改为`fn calculate_price()`### 2. 重复代码- **问题**：相同或相似的代码出现在多个地方- **处理**：提取为函数、类或模块；应用模板方法模式- **示例**：将重复的验证逻辑提取为共享函数### 3. 过长函数- **问题**：函数过长，难以理解和维护- **处理**：提取函数，将大函数分解为多个小函数- **示例**：将200行的处理函数分解为多个职责单一的小函数### 4. 过大的类/结构体- **问题**：类或结构体承担了过多责任，字段和方法过多- **处理**：提取类，将相关字段和方法组合成新的类- **示例**：将User类中的地址相关字段提取为Address类### 5. 过长参数列表- **问题**：函数参数过多，难以理解和使用- **处理**：引入参数对象，将相关参数组合成对象- **示例**：将`fn create_user(name, email, phone, address, city, country)`改为`fn create_user(user_info: UserInfo)`### 6. 发散式变化- **问题**：一个类因为多种原因而被修改- **处理**：按照变化原因拆分类- **示例**：将既处理数据库又处理业务逻辑的类拆分为两个类### 7. 霰弹式修改- **问题**：一次修改需要改动多个类- **处理**：将相关功能移到同一个类中- **示例**：将分散在多个类中的订单处理逻辑合并到一个OrderProcessor类### 8. 依恋情结- **问题**：一个函数对其他类的兴趣超过自己所在的类- **处理**：移动函数或提取函数- **示例**：将过度使用另一个类数据的方法移动到那个类中### 9. 数据泥团- **问题**：相同的数据项总是一起出现- **处理**：提取为对象- **示例**：将经常一起出现的起始日期和结束日期提取为DateRange类### 10. 基本类型偏执- **问题**：使用基本类型表示有特定含义的数据- **处理**：使用小对象替代基本类型- **示例**：用PhoneNumber类替代表示电话的字符串## 重构过程原则### 1. 小步重构- 每次只做一个小改动，然后测试- 频繁提交，保持代码随时可工作### 2. 测试保障- 重构前确保有足够的测试覆盖- 每次修改后运行测试确保行为不变### 3. 代码审查- 重构后进行代码审查，确保质量- 分享重构经验，提高团队能力## 代码可读性优化### 1. 命名约定- 使用有意义的、描述性的名称- 遵循项目或语言的命名规范- 避免缩写和单字母变量（除非是约定俗成的，如循环中的i）### 2. 代码组织- 相关代码应该放在一起- 函数应该只做一件事- 保持适当的抽象层次### 3. 注释与文档- 注释应该解释为什么，而不是做什么- 为公共API提供清晰的文档- 更新注释以反映代码变化## 性能相关重构### 1. 内存优化- 避免不必要的对象创建- 及时释放不再需要的资源- 注意内存泄漏问题### 2. 计算优化- 避免重复计算- 使用适当的数据结构和算法- 延迟计算直到必要时### 3. 并行优化- 识别可并行化的任务- 避免不必要的同步
- 注意线程安全问题

这样我们核心的几个插件工具都搞定了。去工作流里面调用试一下。

总结一下