基于Shamrock与Flask搭建可定制QQ机器人：从协议到消息推送的完整实践

1. 项目缘起：为什么现在还需要自己搭QQ机器人？

最近在折腾一些自动化通知和群管理的小工具，发现很多朋友还在到处找现成的QQ机器人框架，结果要么是年久失修，要么是配置复杂得让人头疼。正好看到Shamrock这个项目最近更新挺活跃，社区反馈也还不错，就决定自己动手搭一个试试水。我的核心需求很简单：能稳定接收QQ消息，能调用外部API（比如我自己的天气查询服务或者AI接口），然后能把处理结果发回QQ。说白了，就是想要一个轻量、可控、能自己定制的“消息中转站”。

你可能要问，现在不是有很多成熟的机器人平台吗？干嘛要自己搭？原因有几个：第一，自己搭建意味着数据完全掌握在自己手里，没有隐私泄露的担忧；第二，定制化程度高，想加什么功能就写什么代码，不受平台功能限制；第三，学习成本其实没想象中高，尤其是对于有一定Python或Web开发基础的朋友来说，这就是一个标准的“后端服务+HTTP API”的集成问题。Shamrock作为一款基于Mirai的衍生框架，它提供了稳定、高效的QQ协议实现，而我们只需要关心业务逻辑，这大大降低了开发门槛。

这次搭建，我会重点结合Qmsg酱这个免费的消息推送服务来演示。Qmsg的好处是它提供了现成的HTTP API，我们不用自己处理复杂的QQ消息发送逻辑，只需要向Qmsg的服务器发一个POST请求，它就能帮我们把消息推送到指定的QQ或QQ群，非常适合用来做通知类机器人。整个技术栈非常清晰：Shamrock负责登录QQ并监听消息，我们自建一个Flask Web服务作为“大脑”来处理消息和决策，最后通过调用Qmsg的API来发送回复。下面，我就把从零开始搭建的完整过程，包括几个关键的避坑点，详细拆解一遍。

2. 环境准备与核心组件选型解析

动手之前，我们得先把“厨房”收拾好，把需要的“食材”备齐。这个项目的核心是三个部分：QQ客户端环境、消息处理后端、以及消息发送通道。

2.1 Shamrock：为什么选它作为QQ协议端？

Shamrock是目前社区内比较活跃的一个Mirai系框架。选择它，主要是基于以下几点考虑：

  • 协议兼容性与稳定性：它持续跟进QQ的新协议，减少了因为协议更新导致机器人掉线的风险。对于需要7x24小时运行的机器人来说，稳定性是第一位的。
  • 开发友好性：它提供了完善的HTTP API和WebSocket API。这意味着我们可以用任何熟悉的编程语言（Python、Java、Go等）来编写业务逻辑，通过标准的HTTP请求与QQ客户端交互，解耦做得非常好。
  • 社区与文档：虽然不如一些明星项目火爆，但它的文档和社区问答足够解决大部分部署问题，遇到坑的时候能找到参考。

部署Shamrock，我推荐使用Docker，这是最干净、最避免环境冲突的方式。你需要先确保服务器上安装了Docker和Docker Compose。

2.2 业务后端：为什么是Flask？

消息处理的后端，我选择了Python的Flask框架。这是一个非常轻量级的Web框架，对于我们这个主要处理HTTP API请求的场景来说，它足够简单、灵活，且生态丰富。几行代码就能拉起一个服务，非常适合快速原型开发和中小型应用。相比Django等“全家桶”框架，Flask给了开发者更大的自由去组合需要的组件（比如数据库连接池、任务队列等），不会引入不必要的复杂性。

2.3 消息推送：Qmsg酱的妙用

Qmsg酱是一个免费的QQ消息推送平台。它的角色很关键：它充当了我们自建后端与QQ之间的一个可靠、免鉴权的发送通道。我们自己搭建的后端服务（Flask应用）在需要发送消息时，无需直接与复杂的QQ协议打交道，只需向Qmsg提供的固定API地址发送一个携带了密钥和消息内容的HTTP POST请求，Qmsg服务器就会帮我们把消息送达目标QQ或群。

这样做的好处显而易见：

  1. 简化发送逻辑：我们不需要在Flask服务里维护QQ的登录状态或处理发送重试。
  2. 提升可靠性：Qmsg作为专业服务，其发送成功率通常高于我们自己实现的简易客户端。
  3. 规避风险：将协议层面的操作交给专门的客户端（Shamrock）和推送服务（Qmsg），我们的业务代码可以更专注于逻辑，结构更清晰。

注意：Qmsg免费版有频率限制，对于个人或小规模使用完全足够。如果你的机器人需要极高的消息推送频率，需要关注其使用条款或考虑升级。

3. 一步步搭建：从零到一的完整实操

理论说完了，我们开始动手。请严格按照步骤操作，我会指出其中容易出错的地方。

3.1 第一步：部署Shamrock服务

首先，在你的服务器上创建一个工作目录，例如qq_bot。然后在这个目录下创建docker-compose.yml文件。

version: '3.8' services: shamrock: image: whitechi73/opengot:shamrock container_name: shamrock restart: unless-stopped network_mode: host # 使用host网络模式，避免复杂的端口映射问题 environment: - TZ=Asia/Shanghai # 设置时区 - SHAMROCK_HTTP=0.0.0.0:8080 # 启用HTTP API，监听所有网卡的8080端口 - SHAMROCK_WS=0.0.0.0:8081 # 启用WebSocket API，端口8081 volumes: - ./shamrock_data:/data # 将容器内的/data目录挂载到本地，持久化配置和登录数据

解释一下关键配置：

  • network_mode: host: 让容器直接使用宿主机的网络栈。这样，容器内服务监听的端口（8080， 8081）就直接暴露在宿主机上，Flask应用可以直接通过localhost:8080来访问Shamrock的API，省去了配置Docker网络和端口映射的麻烦。
  • volumes: 挂载卷至关重要。它把Shamrock的配置、缓存和关键的登录凭证（设备文件）保存在宿主机上。即使容器被删除重建，只要挂载卷还在，重新登录的步骤都可以省略，机器人可以快速恢复上线。

创建好文件后，在终端进入该目录，执行命令启动服务：

docker-compose up -d

使用docker logs -f shamrock可以查看实时日志。当你看到日志里出现等待登录或相关的启动成功信息时，说明Shamrock服务已经正常运行。

3.2 第二步：登录QQ并获取Session Key

Shamrock启动后，它只是一个空壳，我们需要让它登录一个QQ号。这里我们使用其提供的HTTP API来完成认证。

  1. 获取登录验证码：Shamrock支持多种登录方式。对于首次登录，我们通常使用扫码。向它的API发送请求：

    curl -X POST http://localhost:8080/auth

    执行后，查看Shamrock的日志或返回的JSON响应，里面会包含一个二维码链接（url字段）或Base64格式的二维码图片数据。用你的手机QQ扫描这个二维码完成登录。

  2. 验证并获取Session Key：扫码授权成功后，需要验证登录会话并获取一个关键的session_key。这个key是后续所有API调用的凭证。

    curl -X POST http://localhost:8080/verify \ -H "Content-Type: application/json" \ -d '{ "qq": 你的QQ号, "verifyKey": "从/auth响应中获取的verifyKey" }'

    这个请求的响应中会包含session_key。请妥善保存这个key，下面的Flask服务配置会用到它。

    踩坑提示：session_key是有时效的，也可能在Shamrock重启后失效。在生产环境中，你需要编写逻辑在Flask服务启动时或检测到session失效时，自动重新执行这个验证流程来获取新的key。可以将这个逻辑封装成一个函数，在应用启动时调用。

3.3 第三步：编写Flask消息处理后端

现在我们来创建机器人的“大脑”。在qq_bot目录下（与docker-compose.yml同级），新建一个app.py文件。

from flask import Flask, request, jsonify import requests import json import logging # 配置日志，方便排查问题 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = Flask(__name__) # ========== 核心配置区 ========== # 这里填写你从Shamrock获取的实际信息 SHAMROCK_BASE_URL = "http://localhost:8080" # Shamrock HTTP API 地址 SESSION_KEY = "你的SessionKey" # 替换为实际的session_key BOT_QQ_NUMBER = 你的机器人QQ号 # 替换为机器人的QQ号 # Qmsg酱的配置 QMSG_KEY = "你在Qmsg官网获取的KEY" # 在 https://qmsg.zendee.cn 注册后获取 QMSG_API_SEND = "https://qmsg.zendee.cn/send/" # Qmsg发送消息API # ========== 工具函数 ========== def send_via_qmsg(qq, message, msg_type='private'): """ 通过Qmsg酱发送消息 :param qq: 接收者的QQ号（私聊）或群号（群聊需在Qmsg后台绑定） :param message: 要发送的消息内容 :param msg_type: 消息类型，'private' 或 'group' (需Qmsg付费版支持) """ # Qmsg免费版主要支持私聊，这里以私聊为例 if msg_type == 'private': api_url = f"{QMSG_API_SEND}{QMSG_KEY}" else: # 群聊逻辑，可能需要不同的API或处理方式 api_url = f"{QMSG_API_SEND}{QMSG_KEY}?qq={qq}" # 请根据Qmsg最新文档调整 logger.warning("群聊发送可能需付费版支持，请查阅Qmsg文档。") data = { 'msg': message, 'qq': qq # 对于私聊，这里是接收者QQ } headers = {'Content-Type': 'application/x-www-form-urlencoded'} try: resp = requests.post(api_url, data=data, headers=headers, timeout=5) resp.raise_for_status() # 如果状态码不是200，抛出异常 result = resp.json() if result.get('success'): logger.info(f"通过Qmsg向{qq}发送消息成功") return True else: logger.error(f"Qmsg发送失败: {result.get('reason')}") return False except requests.exceptions.RequestException as e: logger.error(f"调用Qmsg API时发生网络错误: {e}") return False except json.JSONDecodeError as e: logger.error(f"解析Qmsg响应JSON失败: {e}") return False # ========== 消息路由与处理 ========== @app.route('/webhook', methods=['POST']) def webhook(): """ 接收Shamrock转发过来的QQ消息。 Shamrock需要配置将消息事件POST到这个地址。 """ try: data = request.json if not data: logger.warning("收到空的POST请求") return jsonify({'code': 400, 'msg': 'Invalid data'}), 400 # 解析通用字段 post_type = data.get('post_type') message_type = data.get('message_type') raw_message = data.get('raw_message', '').strip() sender_id = data.get('user_id') if message_type == 'private' else data.get('group_id') user_id = data.get('user_id') # 发送者QQ号 logger.info(f"收到事件: post_type={post_type}, message_type={message_type}, sender={sender_id}, msg={raw_message[:50]}...") # 只处理私聊和群聊的消息类型 if post_type == 'message': # 示例1：私聊消息，且包含“天气”关键词 if message_type == 'private' and '天气' in raw_message: # 这里可以调用真实的天气API，例如和风天气 # weather_info = get_weather_from_api('北京') # reply_msg = f"北京的天气是：{weather_info}" reply_msg = "【示例】今天北京晴转多云，15~25℃，微风。" # 使用Qmsg回复私聊 send_via_qmsg(qq=user_id, message=reply_msg, msg_type='private') # 示例2：群聊消息，且@了机器人 elif message_type == 'group' and f'[CQ:at,qq={BOT_QQ_NUMBER}]' in raw_message: # 移除@机器人的CQ码，获取纯文本指令 command = raw_message.replace(f'[CQ:at,qq={BOT_QQ_NUMBER}]', '').strip() if command == '帮助': reply_msg = "我是测试机器人，支持命令：\n1. 天气 [城市] - 查询天气\n2. 帮助 - 显示此帮助" # 注意：Qmsg免费版向群发送可能需要特殊配置，这里先回复到私聊 # 更常见的做法是直接使用Shamrock的API回复群消息（见下文备选方案） send_via_qmsg(qq=user_id, message=reply_msg, msg_type='private') # 或者，使用Shamrock API直接回复到群（推荐，但需处理session） # send_via_shamrock_group(group_id=sender_id, message=reply_msg) # 示例3：简单的复读机（测试用） elif message_type == 'private' and raw_message.startswith('echo '): reply_msg = raw_message[5:] # 去掉‘echo ’前缀 send_via_qmsg(qq=user_id, message=reply_msg, msg_type='private') return jsonify({'code': 0, 'msg': 'success'}), 200 except Exception as e: logger.exception(f"处理webhook请求时发生未预期错误: {e}") return jsonify({'code': 500, 'msg': 'Internal server error'}), 500 # ========== 备选发送方案：直接调用Shamrock API ========== def send_via_shamrock_private(target_qq, message): """直接通过Shamrock API发送私聊消息（需有效session）""" api_url = f"{SHAMROCK_BASE_URL}/send_private_msg" payload = { "sessionKey": SESSION_KEY, "target": target_qq, "messageChain": [{"type": "Plain", "text": message}] } try: resp = requests.post(api_url, json=payload, timeout=5) resp.raise_for_status() return resp.json().get('code') == 0 except Exception as e: logger.error(f"通过Shamrock发送私聊消息失败: {e}") return False if __name__ == '__main__': # 启动Flask应用，监听5000端口，允许外部访问（host='0.0.0.0'） app.run(host='0.0.0.0', port=5000, debug=False) # 生产环境务必设置debug=False

这个app.py是整个机器人的逻辑核心，我把它分成了几个部分来便于理解。首先是配置区，这里需要你填入三个关键信息：Shamrock的地址、刚才获取的session_key、以及机器人的QQ号。然后是send_via_qmsg函数，它封装了调用Qmsg API发送消息的所有细节，包括错误处理，这样我们在业务逻辑里调用它就会非常干净。

最核心的是/webhook这个路由。Shamrock会把收到的所有QQ消息事件，以JSON格式POST到这个地址。我们在函数里解析这个JSON，根据消息类型（私聊/群聊）、发送者、消息内容来决定如何回复。我写了三个简单的示例逻辑：私聊查询天气、群聊中@机器人后响应帮助、以及一个私聊的复读机。你可以在这里无限扩展你的机器人功能，比如接入ChatGPT、查询数据库、监控服务器状态等等。

重要提示：session_key直接写在代码里是不安全的，尤其是在开源或共享代码时。正式部署时，务必通过环境变量、配置文件或密钥管理服务来读取这些敏感信息。例如，可以使用os.environ.get('SHAMROCK_SESSION_KEY')来从环境变量获取。

3.4 第四步：配置Shamrock的消息上报

我们的Flask服务写好了，但Shamrock还不知道要把消息往哪里送。我们需要告诉Shamrock：“嘿，以后收到消息，都发到http://我的Flask服务器IP:5000/webhook这个地址。”

通过调用Shamrock的配置API来完成：

curl -X POST http://localhost:8080/config \ -H "Content-Type: application/json" \ -d '{ "sessionKey": "你的SessionKey", "config": { "enableWebhook": true, "webhookUrls": [ "http://你的Flask服务器IP:5000/webhook" ] } }'

请将你的Flask服务器IP替换为运行app.py的那台机器的真实IP地址。如果Shamrock和Flask运行在同一台机器，可以用http://host.docker.internal:5000/webhook（Docker容器内访问宿主机）或者直接使用宿主机在局域网内的IP。

3.5 第五步：注册Qmsg并获取KEY

  1. 访问 Qmsg酱 官网（例如qmsg.zendee.cn，请以最新搜索为准）。
  2. 使用QQ登录。
  3. 在控制台，你可以找到你的唯一KEY。这个KEY是用来标识你的身份的，所有通过你代码发送的消息，Qmsg都靠这个KEY知道要推送给哪个QQ。
  4. 通常你需要将接收消息的QQ号（可以是你的个人QQ，也可以是机器人QQ）与这个KEY进行“绑定”或“添加”，这样Qmsg才有权限向这个QQ发送消息。具体操作请遵循官网指引。

将获取到的QMSG_KEY填入上面app.py的配置区。

3.6 第六步：启动与测试

  1. 启动Flask服务：在qq_bot目录下，运行python app.py。你应该看到输出表明服务在0.0.0.0:5000上启动。
  2. 测试消息流：
    • 用你的手机QQ，给机器人QQ号发送一条包含“天气”二字的私聊消息。
    • 观察Flask服务的日志输出，应该能看到它收到了消息事件。
    • 同时，你的手机QQ应该会很快收到一条来自“Qmsg酱”的天气回复消息。
  3. 测试群聊（如果配置了）：
    • 将机器人拉入一个群。
    • 在群里 @机器人 并输入“帮助”。
    • 观察日志和私聊回复。

4. 关键问题排查与进阶优化

按照上面的步骤，大部分朋友应该能成功跑通。但如果遇到问题，别慌，我们来系统性地排错。

4.1 网络连接与防火墙检查

这是最常见的问题。请确保以下网络通路是畅通的：

  1. 宿主机内部：Flask应用（localhost:5000）能否访问到Shamrock（localhost:8080）？可以用curl http://localhost:8080/about测试。
  2. 容器与宿主机：如果Flask也在Docker中运行，需确保两个容器在同一个Docker网络下，或者使用host网络模式。
  3. 服务器防火墙：确保服务器的防火墙（如ufw, firewalld）或云服务商的安全组规则，放行了5000（Flask）和8080（Shamrock API）端口。Shamrock的host模式意味着它直接使用宿主机的端口，防火墙必须允许8080端口入站。
  4. 公网访问：如果你的Flask服务部署在公网服务器，Shamrock配置的webhookUrls必须是公网可访问的URL。可以用curl -X POST 你的公网URL/webhook简单测试。

4.2 Shamrock Session失效与自动维护

session_key不是永久的。Shamrock重启、QQ长时间未操作都可能使其失效。一个健壮的机器人需要能自动处理这种情况。

我们可以在Flask应用启动时，以及每次调用Shamrock API失败（返回特定的认证错误码，如3）时，触发一个重认证流程。思路如下：

  1. 将获取和验证session_key的逻辑封装成函数get_valid_session_key()。
  2. 在应用启动时调用一次，将获取到的key存入一个全局变量或缓存（如Redis）。
  3. 在send_via_shamrock_private等函数中，如果请求返回code==3（或对应的认证错误），则捕获异常，调用get_valid_session_key()刷新key，然后重试发送操作。
  4. 可以考虑增加一个定时任务（如使用APScheduler），每隔一段时间（如23小时）主动刷新一次session，防患于未然。

4.3 Qmsg发送失败的可能原因

  1. KEY错误或未绑定：检查QMSG_KEY是否填写正确，以及这个KEY是否在Qmsg后台绑定了接收消息的QQ号。
  2. 频率超限：免费用户有发送频率限制。如果短时间内触发大量消息，会被限流。需要在代码中做好限流和队列处理，或者考虑升级服务。
  3. 网络超时：Qmsg服务器偶尔可能不稳定。在send_via_qmsg函数中，我们已经增加了超时设置和异常捕获，并记录了日志。如果发送失败，可以根据业务需求决定是否重试。
  4. 消息格式问题：Qmsg对消息内容可能有长度或字符限制。如果发送超长或包含特殊字符的消息失败，可以尝试截断或转义。

4.4 性能与扩展性考量

当前的架构（Flask单进程）适合低并发场景。如果机器人需要处理大量群消息或复杂计算，需要考虑：

  • 使用生产级WSGI服务器：用Gunicorn或uWSGI替代Flask自带的开发服务器，以支持多Worker并发处理请求。
  • 引入消息队列：当收到一个消息事件，Flask的Webhook处理器只负责快速解析和验证，然后将任务（如“查询北京天气”）放入Redis或RabbitMQ这样的消息队列。再由后台的Worker进程从队列中取出任务，执行耗时的操作（如调用外部API），最后调用发送函数。这样能避免HTTP请求阻塞，大幅提升吞吐量。
  • 无状态化与水平扩展：将session_key等状态信息存入Redis，这样多个Flask实例或Worker可以共享状态。结合负载均衡，可以实现机器人的水平扩展。

4.5 安全加固建议

  1. 校验消息来源：理论上，任何知道你的Webhook地址的人都可以伪造消息POST过来。虽然Shamrock的请求来自本地或内网，风险较低，但在公网环境下，建议在Flask的/webhook端点中，校验请求头中的某个由Shamrock设置的特定Token（如果Shamrock支持配置），或者至少校验来源IP。
  2. 敏感信息脱敏：绝对不要将SESSION_KEY、QMSG_KEY等硬编码在代码或提交到公开仓库。使用.env文件配合python-dotenv库，或直接使用环境变量。
  3. 限制命令权限：在消息处理逻辑中，对于“重启服务”、“执行系统命令”等高危指令，可以校验发送者的QQ号是否在白名单内。

整个搭建过程就像搭积木，Shamrock是连接QQ世界的桥梁，Flask是我们自定义逻辑的车间，Qmsg则是一个高效的物流配送员。这个组合的优势在于每一层都职责清晰，可以独立替换或升级。比如，你觉得Qmsg不够用，完全可以替换成直接调用Shamrock的发送API，或者接入其他推送服务；你觉得Flask太重，也可以用FastAPI、Go、Node.js重写逻辑部分，只要保持HTTP接口一致就行。这种灵活性，正是自建机器人的魅力所在。