Install
openclaw skills install qqQQ 开发助手,精通 QQ 机器人、频道开发、小程序、开放平台 API
openclaw skills install qq你是一个精通 QQ 开放平台全栈开发的 AI 助手,覆盖 QQ 机器人、QQ 频道、QQ 小程序、开放平台 OAuth2.0 等全平台开发能力。
QQ 机器人 API 使用 Bearer Token 鉴权,需先获取 access_token:
import requests
def get_qq_bot_token(app_id: str, app_secret: str) -> str:
"""获取 QQ 机器人 access_token"""
url = "https://bots.qq.com/app/getAppAccessToken"
payload = {
"appId": app_id,
"clientSecret": app_secret
}
resp = requests.post(url, json=payload)
data = resp.json()
# access_token 有效期通常为 7200 秒
return data["access_token"]
机器人通过 WebSocket 接收实时事件,流程如下:
1. GET /gateway → 获取 WebSocket 连接地址
2. 建立 WebSocket 连接
3. 收到 Hello 事件(opcode=10),获取 heartbeat_interval
4. 发送 Identify 鉴权(opcode=2)
5. 收到 Ready 事件,开始接收消息
6. 定时发送心跳(opcode=1)
import asyncio
import json
import aiohttp
class QQBotWebSocket:
def __init__(self, token: str):
self.token = token
self.api_base = "https://api.sgroup.qq.com"
self.session_id = None
self.seq = None
async def get_gateway(self) -> str:
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"QQBot {self.token}"}
async with session.get(f"{self.api_base}/gateway", headers=headers) as resp:
data = await resp.json()
return data["url"]
async def connect(self):
gateway_url = await self.get_gateway()
async with aiohttp.ClientSession() as session:
async with session.ws_connect(gateway_url) as ws:
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
payload = json.loads(msg.data)
await self.handle_event(ws, payload)
async def handle_event(self, ws, payload: dict):
op = payload.get("op")
if op == 10: # Hello
interval = payload["d"]["heartbeat_interval"]
# 发送 Identify
await ws.send_json({
"op": 2,
"d": {
"token": f"QQBot {self.token}",
"intents": 0 | (1 << 30) | (1 << 1) | (1 << 0),
"shard": [0, 1]
}
})
# 启动心跳
asyncio.create_task(self.heartbeat(ws, interval))
elif op == 0: # Dispatch 事件
self.seq = payload.get("s")
event_type = payload.get("t")
await self.dispatch(event_type, payload.get("d", {}))
async def heartbeat(self, ws, interval_ms: int):
while True:
await asyncio.sleep(interval_ms / 1000)
await ws.send_json({"op": 1, "d": self.seq})
async def dispatch(self, event_type: str, data: dict):
print(f"收到事件: {event_type}, 数据: {data}")
Intents 是位掩码,用于声明机器人需要接收的事件类型:
| Intent 值 | 事件类型 | 说明 |
|---|---|---|
1 << 0 | GUILDS | 频道变更(创建/更新/删除) |
1 << 1 | GUILD_MEMBERS | 成员变更(加入/退出) |
1 << 9 | GUILD_MESSAGE_REACTIONS | 表情表态事件 |
1 << 12 | DIRECT_MESSAGE | 频道私信消息 |
1 << 25 | GROUP_AND_C2C_EVENT | 群聊和 C2C 消息事件 |
1 << 30 | AT_MESSAGES | @机器人 消息(公域) |
# 常用 intents 组合
INTENTS_PUBLIC = (1 << 0) | (1 << 1) | (1 << 30) # 频道 + 成员 + @消息
INTENTS_PRIVATE = (1 << 0) | (1 << 1) | (1 << 12) # 含私信
INTENTS_GROUP = (1 << 25) # 群聊和 C2C
async def send_channel_message(channel_id: str, content: str, token: str,
msg_id: str = None, image: str = None):
"""向频道子频道发送消息"""
url = f"https://api.sgroup.qq.com/channels/{channel_id}/messages"
headers = {
"Authorization": f"QQBot {token}",
"Content-Type": "application/json"
}
payload = {"content": content}
if msg_id:
payload["msg_id"] = msg_id # 被动回复,引用原消息
if image:
payload["image"] = image # 图片 URL
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
return await resp.json()
async def send_group_message(group_openid: str, content: str, token: str, msg_id: str):
"""向 QQ 群发送消息(需要 msg_id 被动回复)"""
url = f"https://api.sgroup.qq.com/v2/groups/{group_openid}/messages"
headers = {
"Authorization": f"QQBot {token}",
"Content-Type": "application/json"
}
payload = {
"content": content,
"msg_type": 0, # 0=文本, 2=Markdown, 3=Ark, 4=Embed, 7=富媒体
"msg_id": msg_id
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
return await resp.json()
| msg_type | 类型 | 说明 |
|---|---|---|
| 0 | 文本 | 纯文本消息 |
| 2 | Markdown | Markdown 格式消息 |
| 3 | Ark | Ark 模板消息(卡片) |
| 4 | Embed | 嵌入式消息 |
| 7 | 富媒体 | 图片/视频/语音/文件 |
async def get_guild_info(guild_id: str, token: str) -> dict:
"""获取频道详情"""
url = f"https://api.sgroup.qq.com/guilds/{guild_id}"
headers = {"Authorization": f"QQBot {token}"}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as resp:
return await resp.json()
# 返回: {"id", "name", "icon", "owner_id", "member_count", ...}
async def get_channels(guild_id: str, token: str) -> list:
"""获取频道下的子频道列表"""
url = f"https://api.sgroup.qq.com/guilds/{guild_id}/channels"
headers = {"Authorization": f"QQBot {token}"}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as resp:
return await resp.json()
| type 值 | 类型 | 说明 |
|---|---|---|
| 0 | 文字子频道 | 发送文本/图片/Ark 消息 |
| 1 | 保留 | - |
| 2 | 语音子频道 | 语音通话 |
| 3 | 保留 | - |
| 4 | 子频道分组 | 分类目录 |
| 10005 | 直播子频道 | 直播推流 |
| 10006 | 应用子频道 | 小程序 |
| 10007 | 论坛子频道 | 帖子讨论 |
async def send_markdown_message(channel_id: str, token: str, msg_id: str):
"""发送 Markdown 消息"""
url = f"https://api.sgroup.qq.com/channels/{channel_id}/messages"
headers = {
"Authorization": f"QQBot {token}",
"Content-Type": "application/json"
}
payload = {
"msg_id": msg_id,
"markdown": {
"content": "# 标题\n**加粗文本**\n- 列表项1\n- 列表项2\n[链接](https://example.com)"
}
}
# 也可使用模板 Markdown
# payload = {
# "markdown": {
# "custom_template_id": "模板ID",
# "params": [
# {"key": "title", "values": ["标题内容"]},
# {"key": "content", "values": ["正文内容"]}
# ]
# }
# }
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
return await resp.json()
# Ark 23:链接+文本列表
ark_23 = {
"ark": {
"template_id": 23,
"kv": [
{"key": "#DESC#", "value": "描述信息"},
{"key": "#PROMPT#", "value": "通知提醒"},
{"key": "#LIST#", "obj": [
{"obj_kv": [
{"key": "desc", "value": "项目1: 详情内容"}
]},
{"obj_kv": [
{"key": "desc", "value": "项目2: 详情内容"}
]}
]}
]
}
}
# Ark 24:文本+缩略图
ark_24 = {
"ark": {
"template_id": 24,
"kv": [
{"key": "#DESC#", "value": "描述"},
{"key": "#PROMPT#", "value": "提示"},
{"key": "#TITLE#", "value": "标题"},
{"key": "#METADESC#", "value": "详细描述"},
{"key": "#IMG#", "value": "https://example.com/image.png"},
{"key": "#LINK#", "value": "https://example.com"}
]
}
}
QQ 小程序与微信小程序框架类似,使用 QXML + QSS + JS 开发:
├── app.js # 应用入口
├── app.json # 全局配置
├── app.qss # 全局样式
└── pages/
└── index/
├── index.js
├── index.json
├── index.qxml
└── index.qss
核心差异点:
.qxml(对应微信 .wxml).qss(对应微信 .wxss)qq.xxx(对应微信 wx.xxx)// QQ 小程序登录
qq.login({
success(res) {
qq.request({
url: 'https://your-server.com/api/qq-login',
method: 'POST',
data: { code: res.code },
success(resp) {
qq.setStorageSync('token', resp.data.token);
}
});
}
});
# 后端:code 换取 session
def qq_miniapp_login(code):
url = (
'https://api.q.qq.com/sns/jscode2session'
f'?appid={QQ_APPID}&secret={QQ_SECRET}&js_code={code}'
'&grant_type=authorization_code'
)
resp = requests.get(url).json()
return resp # {"openid": "...", "session_key": "..."}
用户 → 授权页面 → 同意授权 → 回调获取 code → 换取 access_token → 获取用户信息
# 第一步:构造授权 URL
auth_url = (
"https://graph.qq.com/oauth2.0/authorize"
f"?response_type=code&client_id={APP_ID}"
f"&redirect_uri={REDIRECT_URI}&state={STATE}&scope=get_user_info"
)
# 第二步:code 换取 access_token
def get_qq_access_token(code: str) -> dict:
url = (
"https://graph.qq.com/oauth2.0/token"
f"?grant_type=authorization_code&client_id={APP_ID}"
f"&client_secret={APP_SECRET}&code={code}&redirect_uri={REDIRECT_URI}"
"&fmt=json"
)
return requests.get(url).json()
# {"access_token": "...", "expires_in": 7776000, "refresh_token": "..."}
# 第三步:获取 openid
def get_openid(access_token: str) -> str:
url = f"https://graph.qq.com/oauth2.0/me?access_token={access_token}&fmt=json"
data = requests.get(url).json()
return data["openid"]
# 第四步:获取用户信息
def get_user_info(access_token: str, openid: str) -> dict:
url = (
"https://graph.qq.com/user/get_user_info"
f"?access_token={access_token}&oauth_consumer_key={APP_ID}&openid={openid}"
)
return requests.get(url).json()
# {"nickname", "figureurl_qq", "gender", "province", "city", ...}
| API 类型 | 限制 |
|---|---|
| 全局 | 单机器人 800 次/分钟 |
| 频道消息 | 单子频道 5 条/秒 |
| 群消息 | 被动回复,无主动推送 |
| C2C 消息 | 被动回复,无主动推送 |
| 环境 | Base URL |
|---|---|
| 正式环境 | https://api.sgroup.qq.com |
| 沙箱环境 | https://sandbox.api.sgroup.qq.com |
| API | 方法 | 路径 |
|---|---|---|
| 获取网关 | GET | /gateway |
| 频道详情 | GET | /guilds/{guild_id} |
| 子频道列表 | GET | /guilds/{guild_id}/channels |
| 发送频道消息 | POST | /channels/{channel_id}/messages |
| 发送私信 | POST | /dms/{guild_id}/messages |
| 发送群消息 | POST | /v2/groups/{group_openid}/messages |
| 发送 C2C 消息 | POST | /v2/users/{openid}/messages |
| 频道成员列表 | GET | /guilds/{guild_id}/members |
| 禁言成员 | PATCH | /guilds/{guild_id}/members/{user_id}/mute |
async def reconnect(self, ws_url: str):
"""使用 session_id 和 seq 恢复会话"""
async with aiohttp.ClientSession() as session:
async with session.ws_connect(ws_url) as ws:
# 发送 Resume(opcode=6)
await ws.send_json({
"op": 6,
"d": {
"token": f"QQBot {self.token}",
"session_id": self.session_id,
"seq": self.seq
}
})
# 恢复成功后继续接收事件
最后更新: 2026-03-16