## 消息内容支持的格式

| 样式 | 语法 | 效果 |
|------|------|------|
| 加粗 | `**文字**` | **粗体** |
| 斜体 | `*文字*` | *斜体* |
| 删除线 | `~~文字~~` | ~~删除线~~ |
| 颜色 | `<font color='green'>文字</font>` | 绿色文字 |
| @指定人 | `<at id=open_id></at>` | @用户名 |
| @所有人 | `<at user_id="all"></at>` | @所有人 |
| 超链接 | `[文字](https://url)` | 可点击链接 |

# 测试 Webhook (macOS)
curl -X POST -H "Content-Type: application/json" \
 -d '{"msg_type":"text","content":{"id":"1","name":"Tom"}}' \
 [Webhook地址]

# 测试 Webhook (Windows)
curl -X POST -H "Content-Type: application/json" \
 -d "{\"msg_type\":\"text\",\"content\":{\"text\":\"test\"}}" \
 [Webhook地址]

import requests

def get_tenant_access_token(app_id: str, app_secret: str) -> str:
    """获取 tenant_access_token"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = {
        "app_id": app_id,
        "app_secret": app_secret
    }
    response = requests.post(url, json=payload)
    data = response.json()

    if data.get("code") == 0:
        return data["tenant_access_token"]
    else:
        raise Exception(f"获取 token 失败: {data}")

# 使用示例
token = get_tenant_access_token("cli_xxx", "xxx")

// Node.js 版本
const axios = require('axios');

async function getTenantAccessToken(appId, appSecret) {
    const response = await axios.post(
        'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
        { app_id: appId, app_secret: appSecret }
    );

    if (response.data.code === 0) {
        return response.data.tenant_access_token;
    }
    throw new Error(`获取 token 失败: ${JSON.stringify(response.data)}`);
}

# Python
pip install lark-oapi

# Node.js (v1.56.1)
npm install @larksuiteoapi/node-sdk

# Go (v3)
go get github.com/larksuite/oapi-sdk-go/v3

# Java
# Maven: com.larksuite.oapi:oapi-sdk

# 安装 MCP Server
npm install @larksuiteoapi/lark-mcp

{
  "mcpServers": {
    "feishu": {
      "command": "npx",
      "args": ["@larksuiteoapi/lark-mcp"],
      "env": {
        "FEISHU_APP_ID": "your_app_id",
        "FEISHU_APP_SECRET": "your_app_secret"
      }
    }
  }
}

# 使用模板发送卡片（推荐）
def send_card_by_template(receive_id: str, template_id: str, template_variable: dict):
    """使用卡片模板发送消息"""
    content = {
        "type": "template",
        "data": {
            "template_id": template_id,
            "template_variable": template_variable
        }
    }
    return send_message(receive_id, "interactive", content)

# 使用示例
send_card_by_template(
    receive_id="ou_xxx",
    template_id="AAqkxxxxxxxx",
    template_variable={
        "title": "周报提醒",
        "content": "请在今天下班前提交周报"
    }
)

import lark_oapi as lark
from lark_oapi.api.im.v1 import *

# 创建客户端
client = lark.Client.builder() \
    .app_id("cli_xxx") \
    .app_secret("xxx") \
    .build()

# 发送文本消息
def send_text_message(receive_id: str, text: str, receive_id_type: str = "open_id"):
    """
    发送文本消息

    Args:
        receive_id: 接收者ID (open_id/user_id/union_id/email/chat_id)
        text: 消息内容
        receive_id_type: ID类型
    """
    request = CreateMessageRequest.builder() \
        .receive_id_type(receive_id_type) \
        .request_body(CreateMessageRequestBody.builder()
            .receive_id(receive_id)
            .msg_type("text")
            .content('{"text": "' + text + '"}')
            .build()) \
        .build()

    response = client.im.v1.message.create(request)

    if not response.success():
        raise Exception(f"发送失败: {response.msg}")

    return response.data.message_id

# 发送富文本消息
def send_rich_text_message(receive_id: str, title: str, content_list: list):
    """
    发送富文本消息

    Args:
        receive_id: 接收者ID
        title: 消息标题
        content_list: 富文本内容列表
    """
    import json

    content = {
        "zh_cn": {
            "title": title,
            "content": content_list
        }
    }

    request = CreateMessageRequest.builder() \
        .receive_id_type("open_id") \
        .request_body(CreateMessageRequestBody.builder()
            .receive_id(receive_id)
            .msg_type("post")
            .content(json.dumps(content))
            .build()) \
        .build()

    response = client.im.v1.message.create(request)
    return response.data.message_id

# 使用示例
send_text_message("ou_xxx", "Hello, 飞书!")

# 富文本示例
content = [
    [
        {"tag": "text", "text": "项目进度更新："},
        {"tag": "a", "text": "查看详情", "href": "https://example.com"}
    ],
    [
        {"tag": "at", "user_id": "ou_xxx"},
        {"tag": "text", "text": " 请确认"}
    ]
]
send_rich_text_message("ou_xxx", "项目周报", content)

const lark = require('@larksuiteoapi/node-sdk');

// 创建客户端
const client = new lark.Client({
    appId: 'cli_xxx',
    appSecret: 'xxx',
});

// 发送文本消息
async function sendTextMessage(receiveId, text, receiveIdType = 'open_id') {
    const response = await client.im.message.create({
        params: { receive_id_type: receiveIdType },
        data: {
            receive_id: receiveId,
            msg_type: 'text',
            content: JSON.stringify({ text }),
        },
    });

    if (response.code !== 0) {
        throw new Error(`发送失败: ${response.msg}`);
    }

    return response.data.message_id;
}

// 发送卡片消息
async function sendCardMessage(receiveId, card) {
    const response = await client.im.message.create({
        params: { receive_id_type: 'open_id' },
        data: {
            receive_id: receiveId,
            msg_type: 'interactive',
            content: JSON.stringify(card),
        },
    });

    return response.data.message_id;
}

// 卡片消息示例
const card = {
    "config": { "wide_screen_mode": true },
    "header": {
        "title": { "tag": "plain_text", "content": "项目通知" },
        "template": "blue"
    },
    "elements": [
        {
            "tag": "div",
            "text": { "tag": "lark_md", "content": "**任务完成**\n项目A已完成部署" }
        },
        {
            "tag": "action",
            "actions": [
                {
                    "tag": "button",
                    "text": { "tag": "plain_text", "content": "查看详情" },
                    "type": "primary",
                    "url": "https://example.com"
                }
            ]
        }
    ]
};

sendCardMessage('ou_xxx', card);

from lark_oapi.api.im.v1 import *

def create_chat(name: str, description: str, owner_id: str, user_id_list: list):
    """创建群组"""
    request = CreateChatRequest.builder() \
        .user_id_type("open_id") \
        .request_body(CreateChatRequestBody.builder()
            .name(name)
            .description(description)
            .owner_id(owner_id)
            .user_id_list(user_id_list)
            .chat_mode("group")
            .chat_type("private")
            .build()) \
        .build()

    response = client.im.v1.chat.create(request)
    return response.data.chat_id

# 使用示例
chat_id = create_chat(
    name="项目讨论群",
    description="项目A进度讨论",
    owner_id="ou_xxx",
    user_id_list=["ou_aaa", "ou_bbb", "ou_ccc"]
)

def add_chat_members(chat_id: str, member_ids: list):
    """添加群成员"""
    request = CreateChatMembersRequest.builder() \
        .chat_id(chat_id) \
        .member_id_type("open_id") \
        .request_body(CreateChatMembersRequestBody.builder()
            .id_list(member_ids)
            .build()) \
        .build()

    response = client.im.v1.chat_members.create(request)
    return response.success()

from lark_oapi.api.docx.v1 import *

def create_document(title: str, folder_token: str = None):
    """创建飞书文档"""
    request = CreateDocumentRequest.builder() \
        .request_body(CreateDocumentRequestBody.builder()
            .title(title)
            .folder_token(folder_token)
            .build()) \
        .build()

    response = client.docx.v1.document.create(request)
    return {
        "document_id": response.data.document.document_id,
        "title": response.data.document.title
    }

# 使用示例
doc = create_document("周会纪要 2024-01-15")
print(f"文档ID: {doc['document_id']}")

def create_document_block(document_id: str, block_type: str, content: dict):
    """向文档添加内容块"""
    request = CreateDocumentBlockRequest.builder() \
        .document_id(document_id) \
        .request_body(CreateDocumentBlockRequestBody.builder()
            .block_type(block_type)
            .text_run(TextRun.builder()
                .content(content.get("text", ""))
                .build())
            .build()) \
        .build()

    response = client.docx.v1.document_block.create(request)
    return response.data.block

from lark_oapi.api.bitable.v1 import *

def list_bitable_records(app_token: str, table_id: str, page_size: int = 20):
    """获取多维表格记录"""
    request = ListAppTableRecordRequest.builder() \
        .app_token(app_token) \
        .table_id(table_id) \
        .page_size(page_size) \
        .build()

    response = client.bitable.v1.app_table_record.list(request)
    return response.data.items

def create_bitable_record(app_token: str, table_id: str, fields: dict):
    """新增多维表格记录"""
    request = CreateAppTableRecordRequest.builder() \
        .app_token(app_token) \
        .table_id(table_id) \
        .request_body(AppTableRecord.builder()
            .fields(fields)
            .build()) \
        .build()

    response = client.bitable.v1.app_table_record.create(request)
    return response.data.record

# 使用示例
record = create_bitable_record(
    app_token="bascnxxx",
    table_id="tblxxx",
    fields={
        "姓名": "张三",
        "部门": "技术部",
        "入职日期": "2024-01-15"
    }
)

from lark_oapi.api.calendar.v4 import *

def create_calendar_event(
    summary: str,
    start_time: int,
    end_time: int,
    attendee_ids: list = None
):
    """创建日程"""
    attendees = None
    if attendee_ids:
        attendees = [
            CalendarEventAttendee.builder()
                .type("user")
                .user_id(uid)
                .build()
            for uid in attendee_ids
        ]

    request = CreateCalendarEventRequest.builder() \
        .calendar_id("primary") \
        .request_body(CalendarEvent.builder()
            .summary(summary)
            .start_time(TimeInfo.builder()
                .timestamp(str(start_time))
                .build())
            .end_time(TimeInfo.builder()
                .timestamp(str(end_time))
                .build())
            .attendees(attendees)
            .build()) \
        .build()

    response = client.calendar.v4.calendar_event.create(request)
    return response.data.event

# 使用示例
import time
start = int(time.time()) + 3600  # 1小时后
end = start + 3600  # 持续1小时

event = create_calendar_event(
    summary="项目评审会议",
    start_time=start,
    end_time=end,
    attendee_ids=["ou_xxx", "ou_yyy"]
)

from flask import Flask, request
import lark_oapi as lark
from lark_oapi.adapter.flask import *

app = Flask(__name__)

# 创建事件处理器
def handle_message_receive(data):
    """处理接收消息事件"""
    message = data.event.message
    content = message.content
    sender_id = message.sender.sender_id.open_id

    print(f"收到消息: {content}, 发送者: {sender_id}")

    # 自动回复
    # send_text_message(sender_id, "收到您的消息")

handler = lark.EventDispatcherHandler.builder("", "") \
    .register_p2_im_message_receive_v1(handle_message_receive) \
    .build()

@app.route("/webhook/event", methods=["POST"])
def event_handler():
    return handler.do(parse_req())

if __name__ == "__main__":
    app.run(port=8080)

def safe_api_call(func, *args, **kwargs):
    """安全的 API 调用封装"""
    try:
        response = func(*args, **kwargs)

        if hasattr(response, 'success') and not response.success():
            print(f"API 错误: code={response.code}, msg={response.msg}")
            return None

        return response.data

    except Exception as e:
        print(f"调用异常: {e}")
        return None

import time

class TokenManager:
    """Token 管理器，支持自动刷新"""

    def __init__(self, app_id, app_secret):
        self.app_id = app_id
        self.app_secret = app_secret
        self._token = None
        self._expire_time = 0

    def get_token(self):
        # 提前5分钟刷新
        if time.time() > self._expire_time - 300:
            self._refresh_token()
        return self._token

    def _refresh_token(self):
        token = get_tenant_access_token(self.app_id, self.app_secret)
        self._token = token
        self._expire_time = time.time() + 7200  # 2小时有效期

import time
from functools import wraps

def rate_limit(max_calls, period):
    """限流装饰器"""
    calls = []

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [c for c in calls if c > now - period]

            if len(calls) >= max_calls:
                sleep_time = period - (now - calls[0])
                time.sleep(sleep_time)

            calls.append(time.time())
            return func(*args, **kwargs)

        return wrapper
    return decorator

# 使用示例：每秒最多5次调用
@rate_limit(max_calls=5, period=1)
def send_message(receive_id, text):
    # ...
    pass