飞书开放接口 SDK

为帮助开发者更便捷地使用飞书开放能力开发应用，简化接入飞书开放平台时的操作步骤，开放平台提供了统一的服务端 SDK。开发者可使用 SDK 快速开发功能，提高开发效率。

安装

pip install lark-oapi -U

支持 Python 3.7 及以上版本

简单示例

import lark_oapi as lark

# 创建客户端
client = lark.Client.builder().app_id("APP_ID").app_secret("APP_SECRET").build()

# 构造请求对象
request = lark.contact.v3.GetUserRequest.builder().user_id("7be5fg9a").build()

# 发起请求
response = client.contact.v3.user.get(request)

更多示例可参考：请求示例

API 客户端

开发者在调用 API 前，需要先创建一个 API 客户端，然后才能基于该客户端发起 API 调用。

创建客户端

自建应用

import lark_oapi as lark

client = lark.Client.builder() \
    .app_id("APP_ID") \
    .app_secret("APP_SECRET") \
    .build()

商店应用

import lark_oapi as lark

client = lark.Client.builder() \
    .app_id("APP_ID") \
    .app_secret("APP_SECRET") \
    .app_type(lark.AppType.ISV) \
    .build()

客户端配置项

创建 API 客户端时，可以对客户端进行一些配置，比如设置日志级别、HTTP 请求超时时间等。

import lark_oapi as lark

client = lark.Client.builder() \
    .app_id("APP_ID") \
    .app_secret("APP_SECRET") \
    .domain(lark.FEISHU_DOMAIN) \       # 域名，默认为 https://open.feishu.cn
    .timeout(3) \                       # 客户端超时时间，单位秒，默认永不超时
    .app_type(lark.AppType.ISV) \       # 应用类型，默认为自建应用；若设为 ISV 需在 request_option 中配置 tenant_key
    .app_ticket("xxxx") \               # 获取 app_access_token 的凭证，app_type = ISV 时需配置
    .enable_set_token(False) \          # 是否允许手动设置 token，默认不开启；开启后需在 request_option 中配置 token
    .cache(Cache()) \                   # 自定义缓存，默认使用预置的本地缓存
    .log_level(lark.LogLevel.DEBUG) \   # 日志级别，默认为 WARNING
    .build()

API 调用

创建完 API 客户端后，我们可以使用 client.业务域.版本.资源.方法名称 来定位具体的 API 方法，然后对特定的 API 发起调用。

client_expr

飞书开放平台开放的所有 API 列表，可点击这里查看

基本用法

如下示例，我们通过 client 调用通过手机号或邮箱获取用户 ID 接口。

# Code generated by Lark OpenAPI.

import lark_oapi as lark
from lark_oapi.api.contact.v3 import *


client = lark.Client.builder() \
    .app_id("APP_ID") \
    .app_secret("APP_SECRET") \
    .log_level(lark.LogLevel.DEBUG) \
    .build()

# 构造请求对象
request: BatchGetIdUserRequest = BatchGetIdUserRequest.builder() \
    .user_id_type("open_id") \
    .request_body(BatchGetIdUserRequestBody.builder()
                  .emails(["xxxx@bytedance.com"])
                  .mobiles(["15000000000"])
                  .build()) \
    .build()

# 发起请求
response: BatchGetIdUserResponse = client.contact.v3.user.batch_get_id(request)

# 处理失败返回
if not response.success():
    lark.logger.error(
        f"client.contact.v3.user.batch_get_id failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

# 处理业务结果
lark.logger.info(lark.JSON.marshal(response.data, indent=4))

更多示例可参考：请求示例

Request 配置项

在每次发起 API 调用时，可以设置请求级别的一些参数，比如传递 userAccessToken、自定义 headers 等。

import lark_oapi as lark
from lark_oapi.api.contact.v3 import *

# 创建客户端
import lark_oapi as lark
from lark_oapi.api.contact.v3 import *

# 创建客户端
client = lark.Client.builder() \
    .enable_set_token(True) \
    .log_level(lark.LogLevel.DEBUG) \
    .build()

# 构造请求对象
request: BatchGetIdUserRequest = BatchGetIdUserRequest.builder() \
    .user_id_type("open_id") \
    .request_body(BatchGetIdUserRequestBody.builder()
                  .emails(["xxxx@bytedance.com"])
                  .mobiles(["15000000000"])
                  .build()) \
    .build()

# 设置请求选项
headers = {"key1": "value1", "key2": "value2"}
req_opt = lark.RequestOption.builder()\
    .tenant_access_token("t-g1047hjTXIZKCBFYWXUCK3D2LJWZYCWYL7USXXXX")\
    .headers(headers)\
    .build()

# 发起请求
response: BatchGetIdUserResponse = client.contact.v3.user.batch_get_id(request, req_opt)

# 处理失败返回
if not response.success():
    lark.logger.error(
        f"client.contact.v3.user.batch_get_id failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

# 处理业务结果
lark.logger.info(lark.JSON.marshal(response.data, indent=4))

如上使用 RequestOptions 的 Builder 模式构建请求级别的参数。下表展示了所有可设置的选项：

<table> <thead align=left> <tr> <th> 配置选项 </th> <th> 描述 </th> </tr> </thead> <tbody align=left valign=top> <tr> <th> <code>tenant_key</code> </th> <td>租户 key，商店应用必须设置该选项。</td> </tr> <tr> <th> <code>user_access_token</code> </th> <td>用户 token，创建 Client 时 enable_set_token 需要设置为 True。</td> </tr> <tr> <th> <code>tenant_access_token</code> </th> <td>租户 token，创建 Client 时 enable_set_token 需要设置为 True。</td> </tr> <tr> <th> <code>app_access_token</code> </th> <td>应用 token，创建 Client 时 enable_set_token 需要设置为 True。</td> </tr> <tr> <th> <code>headers</code> </th> <td>自定义请求头，这些请求头会被透传到飞书开放平台服务端。</td </tr> </tbody> </table>

原生方式调用

部分老版本接口，由于没有元数据信息，所以无法生成对应的 SDK 模型，需要使用原生方式调用。

如下示例，使用 client 原生方式同样调用通过手机号或邮箱获取用户 ID 接口。

import lark_oapi as lark


# 创建客户端
client = lark.Client.builder() \
    .app_id("APP_ID") \
    .app_secret("APP_SECRET") \
    .log_level(lark.LogLevel.DEBUG) \
    .build()

# 构造请求对象
request: lark.BaseRequest = lark.BaseRequest.builder() \
    .http_method(lark.HttpMethod.POST) \
    .uri("/open-apis/contact/v3/users/batch_get_id") \
    .token_types({lark.AccessTokenType.TENANT}) \
    .queries([("user_id_type", "open_id")]) \
    .body({"emails": ["xxxx@bytedance.com"], "mobiles": ["15000000000"]}) \
    .build()

# 发起请求
response: lark.BaseResponse = client.request(request)

# 处理失败返回
if not response.success():
    lark.logger.error(
        f"client.request failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

# 处理业务结果
lark.logger.info(str(response.raw.content, lark.UTF_8))

更多示例可参考：原生调用

处理消息事件回调

了解消息订阅相关的知识，可以点击这里

获取飞书开放平台开放的所有事件列表，可以点击这里

基本用法

开发者订阅消息事件后，可以使用下面代码，对飞书开放平台推送的消息事件进行处理。

如下示例中使用 flask 启动 httpServer，如使用其他 web 框架，只需处理 http 出入参转换即可。

from flask import Flask

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

app = Flask(__name__)


def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    print(lark.JSON.marshal(data))


def do_customized_event(data: lark.CustomizedEvent) -> None:
    print(lark.JSON.marshal(data))


handler = lark.EventDispatcherHandler.builder(lark.ENCRYPT_KEY, lark.VERIFICATION_TOKEN, lark.LogLevel.DEBUG) \
    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
    .register_p1_customized_event("这里填入你要自定义订阅的 event 的 key，例如 out_approval", do_customized_event) \
    .build()


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


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

其中 EventDispatcherHandler.builder(encrypt_key: str, verification_token: str) 方法参数用于签名验证和消息解密使用，可在开发者后台 ->「事件订阅」中查看。

console 需要注意的是，在注册处理器时，例如使用 register_p2_im_message_receive_v1 注册接收消息事件回调时，其中的 P2 表示消息协议版本。目前飞书开放平台有两种消息协议，分别是 1.0 和 2.0。

如下图所示，开发者在注册消息处理器时，需要从事件列表中查看自己需要的是哪种协议的事件。

如果是 1.0 的消息协议，则注册处理器时需要找以 register_p1_xxxx 开头的。如果是 2.0 的消息协议，则注册处理器时需要找以 register_p2_xxxx 开头的。

如果在 SDK 中未找到处理器，可以使用 register_p1_customized_event 或 register_p2_customized_event 注册自定义事件。

event_prot

更多示例可参考：事件回调

处理卡片行为回调

关于卡片行为相关的知识，可以点击这里查看

基本用法

开发者可以使用以下代码处理卡片回调，示例中使用 flask 启动 httpServer，如使用其他 web 框架，只需处理 http 出入参转换即可。

from typing import Any

from flask import Flask

import lark_oapi as lark
from lark_oapi.adapter.flask import *

app = Flask(__name__)


def do_interactive_card(data: lark.Card) -> Any:
    print(lark.JSON.marshal(data))
    content = {
        "header": {
            "title": {
                "tag": "plain_text",
                "content": "更新卡片成功"
            },
            "template": "green"
        },
        "elements": [
            {
                "tag": "div",
                "text": {
                    "tag": "lark_md",
                    "content": "**Success!\n成功啦😄**"
                }
            },
        ]
    }
    return content


handler = lark.CardActionHandler.builder(lark.ENCRYPT_KEY, lark.VERIFICATION_TOKEN, lark.LogLevel.DEBUG) \
    .register(do_interactive_card) \
    .build()


@app.route("/card", methods=["POST"])
def card():
    resp = handler.do(parse_req())
    return parse_resp(resp)


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