Mac mini 接飞书机器人跑本地 AI Agent：手机发消息即可调度任务

打工仔折腾 AI

201人浏览 · 2026-11-03 17:49:42

打工仔折腾 AI · 2026-11-03 17:49:42 发布

把闲置 Mac mini 变成 AI 触发节点

家里有一台 M2 Mac mini，平时主要跑轻度编译和文件同步，CPU 利用率长期低于 5%。我想把它利用起来，做一个能随时响应外部指令的本地服务。思路很直接：用手机发一条飞书消息，Mac mini 收到后调用本地或云端模型生成回复，顺便执行预设脚本。不依赖第三方 SaaS 平台，数据留在本地，延迟可控，且能作为后续扩展 Agent 能力的底座。

这个方案的核心不是多复杂的框架，而是“事件接入 → 协议校验 → 路由分发 → 任务执行 → 结果回传”的完整链路。我选了飞书自建应用机器人作为消息入口，因为它的 Event Subscription 机制稳定、支持双向交互，且回调地址可以指向任意公网可达的服务。本地运行时用 Python + FastAPI 搭建网关，配合 httpx 调用模型，整体结构保持轻量，方便后续替换 LLM 或增加工具调用。

飞书自建应用的配置要点

飞书事件订阅的配置相对标准，但有几个细节在实际联调时容易被忽略。

进入飞书开放平台控制台，创建企业自建应用，添加“机器人”能力。
在“事件与回调”页面订阅 im.message.receive_v1（用户给机器人发消息）。回调地址必须填公网 HTTPS 地址，建议先写好本地测试端口，后面再用隧道映射。
开启“请求签名校验”，保存好 Request Verification Token 和 Verification Token。这两个值会用于初始握手和后续事件验签。
在“权限管理”中申请 im:message（获取与发送单聊、群聊消息）、im:message.receive_v1（接收消息事件）。权限开通需要管理员审批。

这里需要注意一点：飞书的事件推送是 JSON 格式，包含 encrypt 字段（加密消息）和 plain 字段（明文消息）。如果开启了消息内容加密，解密过程比较繁琐。我选择关闭加密，直接处理 plain 字段，这样调试阶段效率更高。这一点我没有验证带加密场景下的完整流程，生产环境建议按需开启。

初始握手时，飞书会 POST 一个包含 challenge 字段的请求，服务必须原样返回该值才能完成订阅绑定。很多教程会把这步跳过，导致一直报 “signature invalid”。我在代码里单独处理了握手逻辑，避免和常规事件混淆。

本地服务架构与依赖准备

Mac mini 的环境我用的是 Python 3.11，虚拟环境隔离依赖。主要包如下：

包名	版本	作用
fastapi	0.104.1	异步 HTTP 服务框架，内置自动文档
uvicorn	0.24.0	ASGI 服务器，支持热重载和多进程
pydantic	2.5.2	请求体校验与类型提示，减少手写解析
httpx	0.25.2	异步 HTTP 客户端，替代 requests 做非阻塞调用
python-dotenv	1.0.0	环境变量读取，避免硬编码密钥
cryptography	41.0.7	飞书签名验证所需的 HMAC 计算依赖

安装命令很简单：

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic httpx python-dotenv cryptography

Mac mini 的 macOS Ventura 及以上版本默认开启了防火墙，如果需要外部访问本机端口，需要在“系统设置 → 网络 → 防火墙”中允许 uvicorn 的入站连接。另外，macOS 会在屏幕关闭一段时间后进入休眠，这会切断网络。我通过终端执行 sudo pmset -a disablesleep 0 彻底关闭睡眠，并在电池/电源设置中勾选“唤醒以进行网络连接”。这一步实测有效，Mac mini 插着电源线挂机一周后网络依然正常。

核心代码实现

服务主体是一个 FastAPI 应用，负责三件事：握手校验、事件验签、业务路由。下面直接贴核心实现，我会按模块说明设计意图。

# main.py
import hashlib
import hmac
import logging
from typing import Any, Dict

import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import os

load_dotenv()

app = FastAPI(title="mac-mini-feishu-agent")

FEISHU_APP_SECRET = os.getenv("FEISHU_APP_SECRET")
FEISHU_VERIFY_TOKEN = os.getenv("FEISHU_VERIFY_TOKEN")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434/v1/chat/completions")

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class FeishuEvent(BaseModel):
    type: str
    event: Dict[str, Any]

@app.post("/chatbot/event/callback")
async def feishu_callback(request: Request):
    # 1. 握手验证
    raw_body = await request.body()
    try:
        body_dict = await request.json()
    except Exception:
        return JSONResponse(content={"error": "invalid json"}, status_code=400)

    if "challenge" in body_dict:
        return JSONResponse(content={"challenge": body_dict["challenge"]})

    # 2. 签名校验
    timestamp = request.headers.get("X-Timestamp", "")
    nonce = request.headers.get("X-Nonce", "")
    signature = request.headers.get("X-Signature", "")
    
    if not all([timestamp, nonce, signature]):
        return JSONResponse(content={"error": "missing headers"}, status_code=403)

    sign_str = f"{timestamp}\n{nonce}\n{raw_body.decode('utf-8')}"
    computed = hmac.new(
        FEISHU_APP_SECRET.encode(),
        sign_str.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(computed, signature):
        logging.warning("Signature mismatch")
        return JSONResponse(content={"error": "invalid signature"}, status_code=403)

    # 3. 事件解析
    event_data = body_dict.get("event", {})
    msg_type = event_data.get("message", {}).get("message_type")
    if msg_type != "text":
        return JSONResponse(content={"ok": True})

    sender_id = event_data.get("sender", {}).get("sender_id", {}).get("open_id")
    text_content = event_data.get("message", {}).get("content", "{}")
    # 飞书文本消息内容是 JSON 字符串，需二次解析
    import json
    try:
        text_data = json.loads(text_content)
        query = text_data.get("text", "").strip()
    except json.JSONDecodeError:
        query = ""

    if not query:
        return JSONResponse(content={"ok": True})

    # 4. 调用 LLM 并构造回复
    reply_text = await dispatch_llm(query)
    msg_id = event_data.get("message", {}).get("message_id")
    chat_id = event_data.get("message", {}).get("chat_id")
    
    await send_feishu_reply(chat_id, msg_id, reply_text)

    return JSONResponse(content={"ok": True})

async def dispatch_llm(prompt: str) -> str:
    """异步调用本地 Ollama 实例"""
    payload = {
        "model": "qwen2.5:7b",
        "messages": [{"role": "user", "content": prompt}],
        "stream": False
    }
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(OLLAMA_URL, json=payload)
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["message"]["content"]

async def send_feishu_reply(chat_id: str, msg_id: str, text: str):
    """通过飞书 v2 API 回复消息"""
    url = f"https://open.feishu.cn/open-apis/im/v1/messages/{msg_id}/reply"
    headers = {
        "Authorization": f"Bearer {os.getenv('FEISHU_TENANT_ACCESS_TOKEN')}",
        "Content-Type": "application/json; charset=utf-8"
    }
    payload = {
        "msg_type": "text",
        "content": {"text": text}
    }
    async with httpx.AsyncClient(timeout=10.0) as client:
        await client.post(url, headers=headers, json=payload)

代码不长，但有几个关键点值得展开：

为什么用 await request.body() 而不是直接解析 JSON？
飞书的签名校验依赖原始请求体。如果在验签前调用 request.json()，FastAPI 会消费流，导致后续计算哈希时内容为空。所以必须先读原始字节，再反序列化为字典。这是异步框架常见的陷阱，顺序不能颠倒。

为什么 LLM 调用用 httpx.AsyncClient 且设 timeout=30.0？
Mac mini 跑本地模型（如 Ollama 的 qwen2.5:7b）首句生成通常需要 5~15 秒。如果用同步请求，FastAPI 的 worker 线程会被阻塞，其他并发请求会排队甚至超时断开。异步客户端配合事件循环，能保证服务在高并发下依然响应握手和验签逻辑。超时时间设为 30 秒是平衡体验与稳定性的经验值，具体取决于模型大小和显存。

为什么回复接口独立封装？
飞书的消息回复 API 要求携带 msg_id 和 tenant_access_token。Token 可以通过服务端 API 换取，这里我假设已经预置到环境变量中。实际项目中，Token 应该做成定时刷新机制，避免过期导致 401。这一点我没有验证长周期运行的 Token 续期逻辑，短期测试够用。

部署运行与网络穿透

FastAPI 跑在本机 8080 端口没问题，但飞书的事件订阅只能回调公网地址。我有两条路径可选：

方案	优点	缺点	适用场景
Cloudflare Tunnel	免费、稳定、自带 HTTPS、无需公网 IP	首次配置稍复杂，依赖第三方节点	个人/小型项目首选
Ngrok	上手极快、支持自定义域名	免费版限速明显，稳定性一般	临时调试

我用了 Cloudflare Tunnel。安装 cloudflared 后执行 cloudflared tunnel --url http://localhost:8080，控制台会打印出类似 https://abc123.trycloudflare.com 的地址。把这个地址拼上 /chatbot/event/callback 填入飞书后台，点击“启用”即可完成订阅。

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8080 --reload

--reload 用于开发阶段代码热更新，生产环境建议去掉并使用多 worker 模式。

打开飞书手机 App，找到刚才创建的机器人，发送“帮我查一下当前目录下的日志文件大小”。Mac mini 终端会打印请求日志，几秒后手机上会收到模型生成的回复。整个过程延迟通常在 8~12 秒之间，取决于本地模型加载状态。

实际跑通后的几个细节调整

刚跑通时以为万事大吉，上线两天后发现两个问题：一是偶尔出现 504 Gateway Timeout；二是飞书并发推送时验签偶发失败。排查过程比较直接。

504 问题定位：
查看 Nginx/Tunnel 日志，发现上游 FastAPI 无响应。检查代码，dispatch_llm 里的 httpx.AsyncClient 没有复用连接池，每次请求都新建 TCP 握手，加上模型推理期间主事件循环空闲，Tunnel 认为连接已死。解决方案是在服务初始化时创建全局 Client 实例，或者使用 httpx.AsyncClient 的 limits 参数控制并发。改为全局单例后，连续 50 次调用未再超时。

验签偶发失败：
飞书推送到 Mac mini 的间隔有时不足 1 秒。我的验签逻辑里 timestamp 和 nonce 来自 Header，但部分测试环境推送的 nonce 重复率较高。严格来说，HMAC 校验只看内容一致性，不限制 nonce 唯一性。我最初误加了 nonce 缓存去重逻辑，反而导致合法请求被拦截。去掉去重后恢复正常。这里其实比较容易踩坑，验签只认 timestamp\nnonce\nbody 的拼接结果，不要自行加业务规则。

Mac mini 内存管理：
M2 芯片跑 7B 量化模型大约占用 4.5GB 内存。系统长期运行后，Swap 使用量会逐渐上升。我添加了简单的监控脚本，每 10 分钟检查一次内存水位，超过 90% 时触发 sync 并清空 Python 缓存。macOS 的内存回收机制本身比较激进，通常不需要干预，但在高负载 AI 场景下手动压一压 Swap 能避免偶发的模型加载卡顿。

总结与边界说明

把 Mac mini 接飞书跑 AI 触发，本质上是把本地算力暴露为可编程的回调节点。这套方案的优势在于低延迟、数据不出本地、扩展成本低。后续可以叠加 LangChain 的工具链、向量数据库检索，或者直接写函数调用逻辑，让它具备更明确的 Agent 行为。

但也存在明确边界：本地模型的上下文长度和推理速度受硬件限制，不适合处理超长文档或高并发实时对话；飞书 API 的频率限制（默认 50 次/秒）需要在上游做队列缓冲；macOS 的网络策略和电源管理在生产环境中不如 Linux 服务器稳定。如果你只是偶尔让 AI 帮忙跑脚本、整理文件、查询本地数据，这套轻架构完全够用。如果需要大规模调度或多租户隔离，还是应该把网关层迁移到容器化集群。

技术选型从来不是越重越好。能用 FastAPI + httpx 跑通的链路，没必要一开始就引入完整的 Agent 框架。先把消息收发、签名校验、异步调用这三块地基打牢，后面的功能迭代才会顺畅。