Claude Opus 5.5 接入 飞书 完整指南：自定义机器人 + 自建应用双路径实测

1✅ 把 3 小时的会议录音转写稿压成结论 + 待办清单
2✅ 技术方案评审，逐条指出风险点并给出替代方案
3✅ 跨十几个文件的代码 diff 审查
4✅ 客服群里判断用户情绪并给出不同口径的回复
5✅ 长合同 / 长需求文档的条款级问答
6✅ 需要严格输出 JSON 才能驱动后续自动化的场景

1❌ 「在吗」「帮我写个请假条」这类 trivial 任务（用 Haiku 4.5 就行）
2❌ 每秒几十次的批量调用（Opus 5.5 慢且贵，批量任务放离线跑）
3❌ 需要实时联网查最新信息（先接检索，再让模型总结）

1from anthropic import Anthropic
2
3client = Anthropic(
4    api_key=os.environ["CLAUDE_API_KEY"],      # 从环境变量取，别硬编码
5    base_url="https://api.anthropic.com",      # 换中转就改这一行，路径 /v1/messages 不用动
6    default_headers={"anthropic-version": "2023-06-01"},
7)
8
9resp = client.messages.create(
10    model="claude-opus-5-5",                   # 旗舰型号，长任务用它
11    max_tokens=8192,                           # Anthropic 协议必填，少了会被截断
12    system="你是飞书群里的技术助理，回答控制在 200 字以内，先给结论再给理由。",
13    messages=[{"role": "user", "content": "帮我看下这段代码为什么挂了"}],
14    temperature=0.3,
15)
16print(resp.content[0].text)

1from openai import OpenAI
2
3client = OpenAI(
4    api_key=os.environ["GATEWAY_KEY"],         # 网关发的 Key，格式可能不是 sk-ant 开头
5    base_url="https://你的网关域名/v1",          # 注意这里带了 /v1，路径走 /chat/completions
6)
7
8resp = client.chat.completions.create(
9    model="claude-opus-5-5",                   # 型号名以网关文档为准，有的网关要加 -claude 后缀
10    messages=[
11        {"role": "system", "content": "你是飞书群里的技术助理，回答控制在 200 字以内。"},
12        {"role": "user", "content": "帮我看下这段代码为什么挂了"},
13    ],
14    max_tokens=8192,
15)
16print(resp.choices[0].message.content)

1待签字符串 = timestamp + "\n" + secret
2签名 = HMAC-SHA256(key=secret, message=待签字符串)  →  取小写十六进制 hex

1import hashlib
2import hmac
3import time
4import json
5import os
6import requests
7from base64 import b64encode  # 注意：飞书签名本身不用 base64，这里只是对比演示
8
9WEBHOOK = os.environ["FEISHU_WEBHOOK_URL"]   # https://open.feishu.cn/open-apis/bot/v2/hook/xxx
10SECRET  = os.environ["FEISHU_BOT_SECRET"]
11
12def gen_sign(timestamp: int, secret: str) -> str:
13    """飞书签名：timestamp 秒级 + 换行 + secret，HMAC-SHA256 取 hex"""
14    string_to_sign = f"{timestamp}\n{secret}"
15    return hmac.new(
16        secret.encode("utf-8"),              # 密钥是 secret
17        string_to_sign.encode("utf-8"),
18        hashlib.sha256,
19    ).hexdigest()                            # ❗ 这里必须是 hexdigest()，不是 base64
20
21def push(payload_body: dict) -> dict:
22    ts = int(time.time())                    # ❗ 单位是秒，不能乘 1000
23    body = dict(payload_body)
24    body["timestamp"] = str(ts)              # 官方要求字符串形式
25    body["sign"] = gen_sign(ts, SECRET)
26    r = requests.post(
27        WEBHOOK,
28        headers={"Content-Type": "application/json; charset=utf-8"},
29        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
30        timeout=10,
31    )
32    return r.json()

1import anthropic
2
3ai = anthropic.Anthropic(
4    api_key=os.environ["CLAUDE_API_KEY"],
5    base_url=os.environ["CLAUDE_BASE_URL"],   # 官方 https://api.anthropic.com，或你的中转网关
6    default_headers={"anthropic-version": "2023-06-01"},
7)
8
9def ask_claude(question: str) -> str:
10    resp = ai.messages.create(
11        model="claude-opus-5-5",
12        max_tokens=1024,
13        system="你是飞书群里的技术助理。回答用中文，200 字以内，先结论后理由，不要用 Markdown 标题。",
14        messages=[{"role": "user", "content": question}],
15        temperature=0.3,
16    )
17    return resp.content[0].text
18
19reply = ask_claude("数据库连接池报 Too many connections 一般怎么处理？")
20print(push({"msg_type": "text", "content": {"text": reply}}))

1{ "Extra": null, "StatusCode": 0, "StatusMessage": "success" }

1body = {
2    "msg_type": "post",
3    "content": {
4        "post": {
5            "zh_cn": {
6                "title": "每日技术简报 / 2026-05-26",
7                "content": [
8                    [
9                        {"tag": "text", "text": "今天的头条是 "},
10                        {"tag": "a", "text": "订单服务扩容公告", "href": "https://example.com/post/123"},
11                        {"tag": "text", "text": "，请相关同学留意。"}
12                    ],
13                    [{"tag": "at", "user_id": "ou_xxxxxxxxxxxx"}],   # @ 某人，填 open_id
14                    [{"tag": "text", "text": "完整数据见下方卡片。"}],
15                    [{"tag": "img", "image_key": "img_xxxxx"}]        # 图要先上传拿到 image_key
16                ]
17            }
18        }
19    }
20}
21push(body)

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": {
5      "wide_screen_mode": true,      // 宽屏模式，PC 端铺满，不设会窄一半很难看
6      "enable_forward": true         // 允许转发，设为 false 则可防止敏感内容外传
7    },
8    "header": {
9      "template": "blue",            // 主题色：blue/wathet/turquoise/green/yellow/orange/red/carmine/violet/purple/indigo/grey
10      "title": {
11        "tag": "plain_text",         // ❗ 必须是这个对象结构，直接写字符串会报 230002
12        "content": "线上告警 · P1"
13      }
14    },
15    "elements": [
16      {
17        "tag": "div",
18        "text": { "tag": "lark_md", "content": "**服务**：order-service\n**时间**：2026-05-26 03:12\n**错误率**：8.7%" }
19      },
20      { "tag": "hr" },
21      { "tag": "markdown", "content": "Claude Opus 5.5 初判：连接池上限被打满，与 03:05 的发布相关。" },
22      {
23        "tag": "action",
24        "actions": [
25          {
26            "tag": "button",
27            "text": { "tag": "plain_text", "content": "查看详情" },
28            "type": "primary",           // primary / default / danger
29            "url": "https://grafana.example.com/d/xxx"
30          },
31          {
32            "tag": "button",
33            "text": { "tag": "plain_text", "content": "我来处理" },
34            "type": "danger",
35            "value": { "action": "ack", "alert_id": "20260526-0312" }   // ❗ value 必须是对象
36          }
37        ]
38      },
39      {
40        "tag": "note",
41        "elements": [{ "tag": "plain_text", "content": "由 Claude Opus 5.5 自动生成，结论仅供参考" }]
42      }
43    ]
44  }
45}

1App ID:     cli_a1b2c3d4e5f6g7h8
2App Secret: Xd8xxxxxxxxxxxxxxxxxxxxxxxxxxxx

1import time, threading, requests
2
3TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
4
5class TokenCache:
6    """tenant_access_token 有效期 7200 秒，缓存并提前 300 秒续期"""
7    def __init__(self, app_id, app_secret):
8        self.app_id, self.app_secret = app_id, app_secret
9        self._token, self._expire_at = None, 0
10        self._lock = threading.Lock()
11
12    def get(self) -> str:
13        with self._lock:
14            if time.time() < self._expire_at:
15                return self._token
16            r = requests.post(TOKEN_URL, json={
17                "app_id": self.app_id,
18                "app_secret": self.app_secret,
19            }, timeout=10).json()
20            if r.get("code") != 0:
21                raise RuntimeError(f"取 token 失败: {r}")
22            self._token = r["tenant_access_token"]
23            self._expire_at = time.time() + r["expire"] - 300   # expire 一般为 7200
24            return self._token
25
26tokens = TokenCache(os.environ["FEISHU_APP_ID"], os.environ["FEISHU_APP_SECRET"])

1def send_text(receive_id: str, text: str, id_type: str = "chat_id"):
2    """
3    receive_id_type 可选：chat_id（群）/ open_id（用户）/ email / user_id
4    填错类型会报 "invalid receive_id" —— 这是第二高频的错误
5    """
6    url = f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type={id_type}"
7    resp = requests.post(
8        url,
9        headers={"Authorization": "Bearer " + tokens.get()},
10        json={
11            "receive_id": receive_id,
12            "msg_type": "text",
13            "content": json.dumps({"text": text}, ensure_ascii=False),  # ❗ content 是 JSON 字符串不是对象
14        },
15        timeout=15,
16    ).json()
17    if resp.get("code") != 0:
18        raise RuntimeError(f"发送失败 {resp.get('code')}: {resp.get('msg')}")
19    return resp["data"]["message_id"]

1def reply_card(message_id: str, card: dict):
2    """用 reply 接口在消息下挂 reply，或直接 patch 更新已有消息"""
3    return requests.patch(
4        f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}",
5        headers={"Authorization": "Bearer " + tokens.get()},
6        json={"content": json.dumps(card, ensure_ascii=False)},
7        timeout=15,
8    ).json()

1import lark_oapi as lark
2
3def handle_message(data: lark.im.v1.P2ImMessageReceiveV1) -> None:
4    msg = data.event.message
5    text = json.loads(msg.content).get("text", "").strip()
6    if not text:
7        return
8    answer = ask_claude(text)                       # 复用上面写的 Claude 调用
9    send_text(msg.chat_id, answer, "chat_id")
10
11ws = lark.ws.Client(
12    os.environ["FEISHU_APP_ID"],
13    os.environ["FEISHU_APP_SECRET"],
14    event_handler=lark.EventDispatcherHandler.builder("", "") \
15        .register_p2_im_message_receive_v1(handle_message) \
16        .build(),
17    log_level=lark.LogLevel.INFO,
18)
19ws.start()          # 阻塞运行，进程要保持常驻

1from Crypto.Cipher import AES
2import base64, json
3
4ENCRYPT_KEY = os.environ["FEISHU_ENCRYPT_KEY"]
5
6def decrypt(encrypted_b64: str) -> dict:
7    key = hashlib.sha256(ENCRYPT_KEY.encode()).digest()      # 32 字节密钥
8    cipher_bytes = base64.b64decode(encrypted_b64)
9    iv = cipher_bytes[:16]                                    # 前 16 字节是 IV
10    cipher = AES.new(key, AES.MODE_CBC, iv)
11    plain = cipher.decrypt(cipher_bytes[16:])
12    pad_len = plain[-1]
13    return json.loads(plain[:-pad_len].decode("utf-8"))

1我们 order-service 昨天上线后每隔几小时就报下面的错，帮我判断原因并给出排查顺序：
2
3java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available,
4request timed out after 30001ms
5  at com.zaxxer.hikari.pool.HikariPool.createTimeoutException(HikariPool.java:696)
6  ...
7
8补充背景：连接池 maximumPoolSize=10，日活不算高，同一台机器上还跑了 3 个别的 Java 服务。
9请按「最可能 → 次可能」给 3 个假设，每条给出具体验证命令。

1下面是本次周会的语音转写稿（约 8000 字，含口语重复和口误）。请输出：
2
31. 五条以内的核心结论，每条不超过 30 字
42. 待办清单，格式必须是「负责人 | 任务 | 截止时间」，没有提到责任人的标「待认领」
53. 本次会议未达成共识的分歧点，用一句话概括
64. 风险项（如果有人提到过时间紧、资源不够、依赖外部）
7
8转写稿：
9<粘贴文本>

1你是本团队的新人助理，只依据下面这段团队手册回答问题。
2
3规则：
4- 如果手册里没有答案，明确回答「手册里没有这部分，建议直接问 @张三」，不允许编造流程、金额、时效
5- 涉及金额和天数的必须原样引用，不允许换算或推断
6- 回答控制在 150 字以内，超过的部分不用
7- 所有链接必须完整给出，不要说「见相关文档」
8
9团队手册：
10<粘贴手册内容>