QQ群机器人开发与部署完整教程

# QQ 机器人开放平台凭证（https://q.qq.com 「我的机器人」页面获取）
APP_ID = "你的AppID"
APP_SECRET = "你的AppSecret"

xiyi-qq-bot/
├── main.py            # 入口：run 常驻运行 / notify 单次推送通知
├── config.py          # 全部配置（凭证、大脑、提示词、端口）
├── qq_gateway.py      # QQ WebSocket 网关：收群消息、心跳、断线重连
├── qq_api.py          # access_token 管理 + 群消息发送 + Markdown 清洗
├── brain.py           # LLM 大脑：主备切换、并发闸门、工具循环
├── mcp_tools.py       # 联网搜索 MCP 客户端（open-websearch）：长连接 + 只读白名单
├── memory.py          # 群友长期记忆：SQLite + 滚动摘要
├── store.py           # 群 openid 记录：自动发现群并持久化
├── requirements.txt   # 依赖清单
├── start.bat          # Windows 一键启动脚本
├── asset/             # 机器人头像
└── data/              # 运行时自动生成（群记录等）

conda create -n xiyi python=3.10
conda activate xiyi
cd xiyi-qq-bot
uv pip install -r requirements.txt

async def _session(self):
    token = await qq_api.get_access_token()
    url = ...  # 请求 https://api.sgroup.qq.com/gateway 获得网关地址
    async with websockets.connect(url) as ws:
        hello = json.loads(await ws.recv())          # op 10：心跳间隔
        interval = hello["d"]["heartbeat_interval"] / 1000
        if self.session_id:
            await ws.send(json.dumps({"op": 6, "d": {   # 断线恢复
                "token": f"QQBot {token}",
                "session_id": self.session_id,
                "seq": self.last_seq}}))
        else:
            await ws.send(json.dumps({"op": 2, "d": {   # 首次鉴权
                "token": f"QQBot {token}",
                "intents": 1 << 25,                      # 群聊和单聊事件
                "shard": [0, 1],
                "properties": {"os": "server", "browser": "eogee-bot", "device": "eogee-bot"}}}))
        heartbeat = asyncio.create_task(self._heartbeat(ws, interval))
        try:
            async for raw in ws:
                await self._dispatch(ws, json.loads(raw))
        finally:
            heartbeat.cancel()

async def _on_group_message(self, data: dict):
    group_openid = data.get("group_openid", "")
    raw = (data.get("content") or "").strip()
    content = re.sub(r"@[\S]+", "", raw).strip()   # 去掉 @ 标记，只留正文
    _record_group(group_openid)                     # 自动发现群并持久化
    reply = await brain.answer(self.rt, user_key, content)
    await qq_api.send_group_message(group_openid, reply, msg_id=data.get("id"))

async def send_group_message(group_openid, content, msg_id=None):
    content = to_plain(content).strip()
    chunks = [content[i:i + 450] for i in range(0, len(content), 450)]
    for seq, chunk in enumerate(chunks, 1):
        body = {"msg_type": 0, "content": chunk, "msg_seq": seq}
        if msg_id:
            body["msg_id"] = msg_id
        await _post_message(group_openid, body)

for provider in providers:               # 主备依次尝试
    for attempt in (1, 2):               # 每家失败后 3 秒重试一次
        if not _budget_left(deadline, call_timeout):
            break                        # 总预算耗尽，放弃剩余尝试
        try:
            text = await _answer_with_tools(provider, messages, tools)
            break
        except Exception as exc:
            log.warning("服务商 %s 失败：%s", provider["name"], exc)
            if attempt == 1:
                await asyncio.sleep(3)

for _ in range(max_rounds):
    message = await _chat(provider, messages, tools)
    tool_calls = message.get("tool_calls")
    if not tool_calls:
        return message["content"].strip()
    messages.append(message)
    for call in tool_calls:
        result = await rt.mcp.call_tool(call["function"]["name"],
                                        json.loads(call["function"]["arguments"]))
        messages.append({"role": "tool",
                         "tool_call_id": call["id"],
                         "content": result})

# config.py：open-websearch 以 stdio 本地命令接入
MCP_TRANSPORT = "stdio"
MCP_COMMAND = "node"
MCP_ARGS = '["-y", "open-websearch@latest"]'
MCP_ENV = '{"MODE": "stdio", "SEARCH_MODE": "auto"}'

params = StdioServerParameters(command=self.command, args=self.args,
                               env={**get_default_environment(), **self.env})
async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await self._serve(session)   # 初始化、拉取工具列表、常驻服务调用队列

while True:
    name, args, fut = await self._queue.get()   # 其他协程提交的调用请求
    result = await session.call_tool(name, args)
    fut.set_result(self._extract(result))       # 截断后回传给等待方

ctx = await memory.load_context(user_key, recent_rounds)
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
if ctx["summary"]:
    messages.append({"role": "system",
                     "content": f"你和这位群友过去的记忆要点：\n{ctx['summary']}"})
messages += ctx["messages"] + [{"role": "user", "content": question}]

try:
    await asyncio.wait_for(semaphore.acquire(), timeout=queue_wait_timeout)
except asyncio.TimeoutError:
    return BUSY_REPLY          # 「现在找希怡的人太多啦，过一会儿再问我嘛～」
try:
    ...                        # 正常处理
finally:
    semaphore.release()

# 方式一：命令行
python main.py notify --content "新教程已发布：xxx"

# 方式二：本地 HTTP 接口（服务运行中时）
curl -X POST http://127.0.0.1:8901/notify -H "Content-Type: application/json" -d "{\"content\":\"新教程已发布\"}"

python main.py run

schtasks /Create /TN xiyi /TR "C:\xiyi-qq-bot\start.bat" /SC ONLOGON