Python 实战：用小米 MiMo-V2-Pro API 打造 QQ 智能机器人（完整教程）**

摘要

本文详细记录了使用 Python + FastAPI 集成小米 MiMo-V2-Pro 大模型 API，并对接腾讯 QQ 官方机器人平台的完整过程。涵盖 API 调用、Ed25519 签名验证、消息去重、主动发消息等核心难点，附完整可运行代码。

正文

一、项目背景

最近小米开放了 MiMo-V2-Pro 大模型的 API 接口，作为小米自研的 AI 助手，MiMo 在中文理解和对话能力上表现不错。于是想着把它接入 QQ 机器人，做一个智能聊天助手。

整个过程踩了不少坑，特此记录。

最终效果： 在 QQ 中私聊机器人，机器人会调用 MiMo-V2-Pro 生成回复。

二、技术栈

三、项目结构

mimo-tencent-bot/
├── .env # 环境变量
├── config.py # 配置管理
├── mimo_client.py # MiMo API 客户端
├── tencent_bot.py # QQ 机器人 webhook 处理
├── main.py # 启动入口
├── test_mimo.py # MiMo API 单独测试
├── test_verify.py # Ed25519 签名测试
└── requirements.txt # 依赖

四、环境准备

4.1 获取 MiMo API Key

前往 小米 MiMo API 平台 注册并获取 API Key，同时充值余额（402 错误就是余额不足）。

4.2 创建 QQ 机器人

前往 QQ 机器人开放平台：

1. 创建机器人应用，获取 App ID 和 App Secret
2. 在「开发设置」中配置回调地址
3. 订阅事件：C2C_MESSAGE_CREATE（私聊消息）

4.3 安装依赖

pip install fastapi uvicorn openai pydantic-settings python-dotenv PyNaCl httpx

requirements.txt：

fastapi>=0.135.1
uvicorn>=0.42.0
httpx>=0.27.0
pydantic>=2.0
pydantic-settings>=2.0
python-dotenv>=1.2.2
cryptography>=42.0.0

五、配置文件

# .env
MIMO_API_KEY=你的MiMo API Key
MIMO_API_BASE=https://api.xiaomimimo.com/v1
MIMO_MODEL=mimo-v2-pro

TENCENT_APP_ID=你的QQ机器人App ID
TENCENT_APP_SECRET=你的QQ机器人App Secret
TENCENT_TOKEN=你的Token
TENCENT_AES_KEY= #这个没有可以不填

HOST=0.0.0.0
PORT=8080

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
mimo_api_key: str
mimo_api_base: str = "https://api.xiaomimimo.com/v1"
mimo_model: str = "mimo-v2-pro"
mimo_max_tokens: int = 1024
mimo_temperature: float = 1.0
mimo_top_p: float = 0.95

tencent_app_id: str
tencent_app_secret: str
tencent_token: str
tencent_aes_key: str = ""

host: str = "0.0.0.0"
port: int = 8080

class Config:
env_file = ".env"
env_file_encoding = "utf-8"

@lru_cache()
def get_settings() –> Settings:
return Settings()

六、MiMo API 客户端

MiMo 提供了 OpenAI 兼容接口，可以直接用 openai SDK 调用。

注意： 参数名是 max_completion_tokens，不是 max_tokens。

# mimo_client.py
import json
import logging
from typing import Optional
from openai import AsyncOpenAI

from config import get_settings

logger = logging.getLogger(__name__)

class MiMoClient:
"""MiMo-V2-Pro API 客户端"""

def __init__(self):
settings = get_settings()
self.model = settings.mimo_model
self.max_tokens = settings.mimo_max_tokens
self.temperature = settings.mimo_temperature
self.top_p = settings.mimo_top_p

self.client = AsyncOpenAI(
api_key=settings.mimo_api_key,
base_url=settings.mimo_api_base,
)

async def chat(
self,
messages: list[dict],
system_prompt: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
) –> str:
payload_messages = []
if system_prompt:
payload_messages.append({"role": "system", "content": system_prompt})
payload_messages.extend(messages)

try:
completion = await self.client.chat.completions.create(
model=self.model,
messages=payload_messages,
max_completion_tokens=max_tokens or self.max_tokens,
temperature=temperature or self.temperature,
top_p=self.top_p,
stream=False,
frequency_penalty=0,
presence_penalty=0,
)

reply = completion.choices[0].message.content
usage = completion.usage
logger.info(
f"MiMo 响应成功 | "
f"prompt={usage.prompt_tokens} | "
f"completion={usage.completion_tokens}"
)
return reply

except Exception as e:
import traceback
print(f"\\nMiMo 调用失败: {type(e).__name__}: {e}")
traceback.print_exc()
return "抱歉，AI 服务暂时不可用，请稍后再试。"

async def close(self):
await self.client.close()

单独测试 MiMo API 是否可用：

# test_mimo.py
from openai import OpenAI

client = OpenAI(
api_key="你的API Key",
base_url="https://api.xiaomimimo.com/v1"
)

completion = client.chat.completions.create(
model="mimo-v2-pro",
messages=[
{"role": "system", "content": "You are MiMo, an AI assistant developed by Xiaomi."},
{"role": "user", "content": "你好，请介绍一下自己"}
],
max_completion_tokens=1024,
temperature=1.0,
top_p=0.95,
)

print(completion.model_dump_json())

七、QQ 机器人核心逻辑

这是整个项目最关键的部分，包含三个核心难点。

难点一：Ed25519 签名验证

QQ 平台首次配置回调地址时，会发送 op=13 的验证请求，需要使用 Ed25519 算法签名并返回。

官方文档的计算过程：

1. 将 bot_secret 扩展到 32 字节作为 seed
2. 用 seed 生成 Ed25519 密钥对
3. 对 event_ts + plain_token 拼接字符串进行签名
4. 返回 hex 编码的签名

import binascii
import nacl.signing

def generate_signature(bot_secret: str, event_ts: str, plain_token: str) –> str:
# 将 secret 扩展到 32 字节
seed = bot_secret
while len(seed) < 32:
seed += bot_secret
seed_bytes = seed[:32].encode("utf-8")

# 生成 Ed25519 密钥对
signing_key = nacl.signing.SigningKey(seed_bytes)

# 签名内容: event_ts + plain_token
message = (event_ts + plain_token).encode("utf-8")

# 签名
signed = signing_key.sign(message)
signature = binascii.hexlify(signed.signature).decode("utf-8")

return signature

用官方示例验证签名正确性：

# test_verify.py
import binascii
import nacl.signing

# 官方示例
bot_secret = "DG5g3B4j9X2KOErG"
event_ts = "1725442341"
plain_token = "Arq0D5A61EgUu4OxUvOp"
expected = "87befc99c42c651b3aac0278e71ada338433ae26fcb24307bdc5ad38c1adc2d01bcfcadc0842edac85e85205028a1132afe09280305f13aa6909ffc2d652c706"

seed = bot_secret
while len(seed) < 32:
seed += bot_secret
seed_bytes = seed[:32].encode("utf-8")

signing_key = nacl.signing.SigningKey(seed_bytes)
message = (event_ts + plain_token).encode("utf-8")
signed = signing_key.sign(message)
signature = binascii.hexlify(signed.signature).decode("utf-8")

print(f"计算: {signature}")
print(f"期望: {expected}")
print(f"匹配: {signature == expected}")

难点二：QQ 私聊不能在 webhook 响应中返回消息

这是最容易踩的坑。QQ 机器人的 webhook 只是接收事件通知，回复消息必须通过 API 主动发送。

# 主动发送私聊消息
POST https://api.sgroup.qq.com/v2/users/{user_openid}/messages
Authorization: QQBot {access_token}

难点三：消息去重

QQ 平台会重复推送同一条消息，需要用 msg_id 去重。

完整的 tencent_bot.py

# tencent_bot.py
import json
import logging
import re
import binascii
from typing import Optional
from collections import defaultdict

from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import nacl.signing
import httpx

from mimo_client import MiMoClient
from config import get_settings

logger = logging.getLogger(__name__)
router = APIRouter()

mimo = MiMoClient()

conversation_history: dict[str, list[dict]] = defaultdict(list)
processed_messages: set[str] = set()
MAX_HISTORY = 20
MAX_PROCESSED = 10000

SYSTEM_PROMPT = (
"你是一个部署在QQ平台上的智能助手，由小米 MiMo-V2-Pro 驱动。"
"你友好、专业、乐于助人。请用简洁清晰的语言回答用户的问题。"
"如果不确定答案，请诚实说明。"
)

# ==================== 数据模型 ====================

class WebhookPayload(BaseModel):
op: int
d: Optional[dict] = None
s: Optional[int] = None
t: Optional[str] = None

# ==================== Ed25519 签名 ====================

def generate_signature(bot_secret: str, event_ts: str, plain_token: str) –> str:
seed = bot_secret
while len(seed) < 32:
seed += bot_secret
seed_bytes = seed[:32].encode("utf-8")

signing_key = nacl.signing.SigningKey(seed_bytes)
message = (event_ts + plain_token).encode("utf-8")
signed = signing_key.sign(message)
return binascii.hexlify(signed.signature).decode("utf-8")

# ==================== QQ Bot API ====================

class QQBotAPI:
"""QQ 机器人 API 客户端，用于主动发送消息"""

def __init__(self):
settings = get_settings()
self.app_id = settings.tencent_app_id
self.app_secret = settings.tencent_app_secret
self.base_url = "https://api.sgroup.qq.com"
self.access_token: Optional[str] = None

async def get_access_token(self) –> str:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
"https://bots.qq.com/app/getAppAccessToken",
json={
"appId": self.app_id,
"clientSecret": self.app_secret,
},
)
data = resp.json()
self.access_token = data.get("access_token", "")
logger.info("获取 access_token 成功")
return self.access_token

async def _request(self, method: str, path: str, json_data: dict = None) –> dict:
if not self.access_token:
await self.get_access_token()

async with httpx.AsyncClient(timeout=15.0) as client:
headers = {
"Authorization": f"QQBot {self.access_token}",
"Content-Type": "application/json",
}
url = f"{self.base_url}{path}"

if method == "POST":
resp = await client.post(url, headers=headers, json=json_data)
else:
resp = await client.get(url, headers=headers)

if resp.status_code == 401:
await self.get_access_token()
headers["Authorization"] = f"QQBot {self.access_token}"
if method == "POST":
resp = await client.post(url, headers=headers, json=json_data)

return resp.json() if resp.text else {}

async def send_c2c_message(
self, user_openid: str, content: str, msg_id: str = ""
) –> dict:
"""发送私聊消息"""
payload = {"content": content, "msg_type": 0}
if msg_id:
payload["msg_id"] = msg_id
return await self._request(
"POST", f"/v2/users/{user_openid}/messages", payload
)

async def send_group_message(
self, group_openid: str, content: str, msg_id: str = ""
) –> dict:
"""发送群聊消息"""
payload = {"content": content, "msg_type": 0}
if msg_id:
payload["msg_id"] = msg_id
return await self._request(
"POST", f"/v2/groups/{group_openid}/messages", payload
)

qq_api = QQBotAPI()

# ==================== 消息处理 ====================

def extract_message_text(content: str) –> str:
return re.sub(r"<@!\\d+>", "", content).strip()

async def handle_c2c_message(event_data: dict):
"""处理私聊消息（C2C）"""
content = event_data.get("content", "").strip()
msg_id = event_data.get("id", "")
author = event_data.get("author", {})
user_openid = author.get("user_openid", "")

# 消息去重
if msg_id in processed_messages:
print(f"跳过重复消息: {msg_id}")
return
processed_messages.add(msg_id)
if len(processed_messages) > MAX_PROCESSED:
processed_messages.clear()

text = extract_message_text(content)

if not text:
await qq_api.send_c2c_message(
user_openid, "你好！有什么可以帮你的吗？", msg_id
)
return

session_key = f"c2c:{user_openid}"
conversation_history[session_key].append({"role": "user", "content": text})
if len(conversation_history[session_key]) > MAX_HISTORY:
conversation_history[session_key] = conversation_history[session_key][–MAX_HISTORY:]

reply = await mimo.chat(
messages=conversation_history[session_key],
system_prompt=SYSTEM_PROMPT,
)

conversation_history[session_key].append({"role": "assistant", "content": reply})
logger.info(f"私聊 {user_openid} | 用户: {text} | 回复: {reply[:50]}…")

await qq_api.send_c2c_message(user_openid, reply, msg_id)

async def handle_group_at_message(event_data: dict):
"""处理群聊 @消息"""
content = event_data.get("content", "").strip()
msg_id = event_data.get("id", "")
group_openid = event_data.get("group_openid", "")
author_openid = event_data.get("author", {}).get("member_openid", "")

if msg_id in processed_messages:
return
processed_messages.add(msg_id)

text = extract_message_text(content)
if not text:
return

session_key = f"group:{group_openid}:{author_openid}"
conversation_history[session_key].append({"role": "user", "content": text})
if len(conversation_history[session_key]) > MAX_HISTORY:
conversation_history[session_key] = conversation_history[session_key][–MAX_HISTORY:]

reply = await mimo.chat(
messages=conversation_history[session_key],
system_prompt=SYSTEM_PROMPT,
)

conversation_history[session_key].append({"role": "assistant", "content": reply})
await qq_api.send_group_message(group_openid, reply, msg_id)

# ==================== Webhook 路由 ====================

@router.post("/webhook")
async def webhook_handler(request: Request):
try:
body = await request.json()
print(f"\\n收到事件: {json.dumps(body, ensure_ascii=False)}")

event = WebhookPayload(**body)

# op=13: 回调地址验证
if event.op == 13:
settings = get_settings()
d = event.d or {}
plain_token = d.get("plain_token", "")
event_ts = d.get("event_ts", "")

signature = generate_signature(
settings.tencent_app_secret,
event_ts,
plain_token,
)

return JSONResponse(content={
"plain_token": plain_token,
"signature": signature,
})

# op=0: 事件分发
event_name = event.t or ""
event_data = event.d or {}

if event_name == "C2C_MESSAGE_CREATE":
await handle_c2c_message(event_data)
elif event_name == "GROUP_AT_MESSAGE_CREATE":
await handle_group_at_message(event_data)

return JSONResponse(content={"op": 0})

except Exception as e:
logger.error(f"处理 webhook 异常: {e}", exc_info=True)
return JSONResponse(
content={"status": "error", "message": str(e)},
status_code=500,
)

@router.get("/webhook")
async def webhook_get():
return {"status": "ok", "message": "QQ Bot webhook is running"}

八、启动入口

# main.py
import logging
import uvicorn
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from config import get_settings
from tencent_bot import router as bot_router, mimo

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
settings = get_settings()
logger.info("=" * 50)
logger.info("MiMo-Tencent Bot 启动中…")
logger.info(f" MiMo 模型: {settings.mimo_model}")
logger.info(f" MiMo API: {settings.mimo_api_base}")
logger.info(f" 监听地址: {settings.host}:{settings.port}")
logger.info("=" * 50)
yield
await mimo.close()
logger.info("MiMo-Tencent Bot 已关闭")

app = FastAPI(title="MiMo Tencent Bot", version="1.0.0", lifespan=lifespan)

app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)

app.include_router(bot_router, prefix="/api/v1", tags=["bot"])

@app.get("/")
async def root():
return {"service": "MiMo Tencent Bot", "status": "running"}

if __name__ == "__main__":
settings = get_settings()
uvicorn.run("main:app", host=settings.host, port=settings.port, reload=True)

九、部署与调试

9.1 本地测试

# 终端1：启动服务
python main.py

# 终端2：用 ngrok 穿透
ngrok http 8080

# 终端3：模拟 QQ 请求测试
python test_mimo.py

9.2 配置回调地址

将 ngrok 提供的公网地址填入 QQ 机器人平台：

https://xxxxx.ngrok-free.dev/api/v1/webhook

9.3 生产部署

建议部署到有公网 IP 的云服务器：

# 服务器上运行
pip install -r requirements.txt
nohup python main.py > bot.log 2>&1

用 Nginx 反向代理 + HTTPS 证书。

十、踩坑记录

十一、架构流程图

QQ 用户发送消息
│
▼
┌─────────────┐ POST /api/v1/webhook ┌──────────────────┐
│ QQ 机器人 │ ──────────────────────▶ │ FastAPI 服务 │
│ 平台 │ ◀────────────────────── │ (本项目) │
│ │ 200 OK │ │
└─────────────┘ │ ┌────────────┐ │
▲ │ │ 消息去重 │ │
│ │ └────────────┘ │
│ POST /v2/users/{openid}/messages │ │ │
│ │ ▼ │
└────────────────────────────────────│ ┌────────────┐ │
│ │ MiMo Client │ │
│ └─────┬──────┘ │
└────────┼─────────┘
│
▼
┌────────────────┐
│ MiMo-V2-Pro │
│ API Server │
└────────────────┘

十二、总结

整个项目的核心流程：

1. MiMo API 调用 → OpenAI 兼容接口，用 openai SDK
2. QQ 回调验证 → Ed25519 签名，用 PyNaCl
3. 消息回复 → webhook 只接收事件，回复需调 API 主动发
4. 消息去重 → QQ 平台会重复推送，需 msg_id 去重

以下是Python终端运行成功截图、ngork内网穿透运行成功截图与机器人的回复截图

标签

# Python # MiMo # 小米大模型 # QQ机器人 # FastAPI
# AI助手 # Ed25519 # webhook # 大模型API