代码拉取完成,页面将自动刷新
README of Larksuite(Overseas) | 飞书
飞书开放平台,便于企业应用与飞书集成,让协同与管理更加高效,概述
飞书开发接口SDK,便捷调用服务端API与订阅服务端事件,例如:消息&群组、通讯录、日历、视频会议、云文档、 OKR等具体可以访问 飞书开放平台文档 看看【服务端 API】。
如有任何SDK使用相关问题,请提交 Github Issues, 我们会在收到 Issues 的第一时间处理,并尽快给您答复。
pip install typing # python version < 3.5
pip install larksuite-oapi==1.0.33
应用商店应用
调用 服务端API 示例企业自建应用
访问 发送消息API 示例# -*- coding: UTF-8 -*-
import logging
from larksuiteoapi.service.im.v1 import Service as ImService, model
from larksuiteoapi import DOMAIN_FEISHU, Config, LEVEL_DEBUG, LEVEL_INFO, \
LEVEL_WARN, LEVEL_ERROR
# 企业自建应用的配置
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(App ID、App Secret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(Verification Token、Encrypt Key)。
# 更多介绍请看:Github->README.zh.md->如何构建应用配置(AppSettings)
app_settings = Config.new_internal_app_settings_from_env()
# 当前访问的是飞书,使用默认存储、默认日志(Error级别)
# 更多介绍请看:Github->README.zh.md->如何构建整体配置(Config)
conf = Config(DOMAIN_FEISHU, app_settings, log_level=LEVEL_DEBUG)
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
service = ImService(conf)
def test_message_create():
# body
body = model.MessageCreateReqBody()
body.content = '{"text":"<at user_id=\\"ou_a11d2bcc7d852afbcaf37e5b3ad01f7e\\">Tom</at> test content"}'
body.msg_type = 'text'
body.receive_id = 'ou_a11d2bcc7d852afbcaf37e5b3ad01f7e'
req_call = service.messages.create(body)
req_call.set_receive_id_type('open_id')
resp = req_call.do()
print('request id = %s' % resp.get_request_id())
print('http status code = %s' % resp.get_http_status_code())
print('header = %s' % resp.get_header().items())
if resp.code == 0:
print(resp.data.message_id)
else:
print(resp.msg)
print(resp.error)
if __name__ == '__main__':
test_message_create()
企业自建应用
访问 发送文本消息API 示例原生
模式。import logging
from larksuiteoapi.api import Request, set_timeout
from larksuiteoapi import Config, ACCESS_TOKEN_TYPE_TENANT, DOMAIN_FEISHU, LEVEL_DEBUG
# 企业自建应用的配置
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(AppID、AppSecret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(VerificationToken、EncryptKey)
# 更多可选配置,请看:README.zh.md->如何构建应用配置(AppSettings)。
app_settings = Config.new_internal_app_settings_from_env()
# 当前访问的是飞书,使用默认存储、默认日志(Error级别),更多可选配置,请看:README.zh.md->如何构建整体配置(Config)。
conf = Config(DOMAIN_FEISHU, app_settings, log_level=LEVEL_DEBUG)
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
def test_send_message():
body = {
"user_id": "77bbc392",
"msg_type": "text",
"content": {
"text": "test send message",
}
}
req = Request('/open-apis/message/v4/send', 'POST', ACCESS_TOKEN_TYPE_TENANT, body, request_opts=[set_timeout(3)])
resp = req.do(conf)
print('request id = %s' % resp.get_request_id())
print(resp.code)
if resp.code == 0:
print(resp.data)
else:
print(resp.msg)
print(resp.error)
if __name__ == '__main__':
test_send_message()
企业自建应用
订阅 员工变更事件 示例# -*- coding: UTF-8 -*-
import logging
from larksuiteoapi.event import handle_event
from larksuiteoapi.service.contact.v3 import UserUpdatedEventHandler, UserUpdatedEvent
from larksuiteoapi.model import OapiHeader, OapiRequest
from flask import Flask, request
from flask.helpers import make_response
from larksuiteoapi import Config, Context, DOMAIN_FEISHU, DefaultLogger, LEVEL_DEBUG
# 企业自建应用的配置
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(AppID、AppSecret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(VerificationToken、EncryptKey)
# 更多可选配置,请看:README.zh.md->如何构建应用配置(AppSettings)。
app_settings = Config.new_internal_app_settings_from_env()
# 当前访问的是飞书,使用默认存储、默认日志(Error级别),更多可选配置,请看:README.zh.md->如何构建整体配置(Config)。
conf = Config(DOMAIN_FEISHU, app_settings, log_level=LEVEL_DEBUG)
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
def user_update_handle(ctx, conf, event):
# type: (Context, Config, UserUpdatedEvent) ->None
print(ctx.get_request_id())
print(event.header)
print(event.event)
pass
# set event type 'contact.user.updated_v3' handle
UserUpdatedEventHandler.set_callback(conf, user_update_handle)
app = Flask(__name__)
@app.route('/webhook/event', methods=['GET', 'POST'])
def webhook_event():
oapi_request = OapiRequest(uri=request.path, body=request.data, header=OapiHeader(request.headers))
resp = make_response()
oapi_resp = handle_event(conf, oapi_request)
resp.headers['Content-Type'] = oapi_resp.content_type
resp.data = oapi_resp.body
resp.status_code = oapi_resp.status_code
return resp
# 设置 "开发者后台" -> "事件订阅" 请求网址 URL:https://domain/webhook/event
# startup event http server, port: 8089
if __name__ == '__main__':
app.run(port=8089, host="0.0.0.0")
企业自建应用
订阅 首次启用应用事件 示例原生
模式import logging
from larksuiteoapi.event import handle_event, set_event_callback
from larksuiteoapi.model import OapiHeader, OapiRequest
from flask import Flask, request
from flask.helpers import make_response
from larksuiteoapi import Config, Context, DOMAIN_FEISHU, DefaultLogger, LEVEL_DEBUG
# 企业自建应用的配置
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(AppID、AppSecret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(VerificationToken、EncryptKey)
# 更多可选配置,请看:README.zh.md->如何构建应用配置(AppSettings)。
app_settings = Config.new_internal_app_settings_from_env()
# 当前访问的是飞书,使用默认存储、默认日志(Error级别),更多可选配置,请看:README.zh.md->如何构建整体配置(Config)。
conf = Config(DOMAIN_FEISHU, app_settings, log_level=LEVEL_DEBUG)
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
def app_open_event_handle(ctx, conf, event):
# type: (Context, Config, dict) -> None
print(ctx.get_request_id())
print(conf.app_settings)
print(event)
# set event type 'app_status_change' handle
set_event_callback(conf, 'app_open', app_open_event_handle)
app = Flask(__name__)
@app.route('/webhook/event', methods=['GET', 'POST'])
def webhook_event():
oapi_request = OapiRequest(uri=request.path, body=request.data, header=OapiHeader(request.headers))
resp = make_response()
oapi_resp = handle_event(conf, oapi_request)
resp.headers['Content-Type'] = oapi_resp.content_type
resp.data = oapi_resp.body
resp.status_code = oapi_resp.status_code
return resp
# 设置 "开发者后台" -> "事件订阅" 请求网址 URL:https://domain/webhook/event
# startup event http server, port: 8089
if __name__ == '__main__':
app.run(port=8089, host="0.0.0.0")
企业自建应用
处理消息卡片回调示例import logging
from typing import Any, Union, Dict
from larksuiteoapi import Config, Context, DOMAIN_FEISHU, DefaultLogger, LEVEL_DEBUG
from larksuiteoapi.card import Card, set_card_callback, handle_card
from larksuiteoapi.model import OapiHeader, OapiRequest
from flask import Flask, request
from flask.helpers import make_response
# 企业自建应用的配置
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(AppID、AppSecret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(VerificationToken、EncryptKey)
# 更多可选配置,请看:README.zh.md->如何构建应用配置(AppSettings)。
app_settings = Config.new_internal_app_settings_from_env()
# 当前访问的是飞书,使用默认存储、默认日志(Error级别),更多可选配置,请看:README.zh.md->如何构建整体配置(Config)。
conf = Config(DOMAIN_FEISHU, app_settings, log_level=LEVEL_DEBUG)
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
# 设置消息卡片的处理
# 返回值:可以为None、新的消息卡片的Json(dict)
def handle(ctx, conf, card):
# type: (Context, Config, Card) -> Union[None, Dict]
print('card = %s' % card)
return {
"config": {
"wide_screen_mode": True
},
"card_link": {
"url": "https://www.baidu.com",
"android_url": "https://developer.android.com/",
"ios_url": "https://developer.apple.com/",
"pc_url": "https://www.windows.com"
},
"header": {
"title": {
"tag": "plain_text",
"content": "this is header"
}
},
"elements": [
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": "This is a very very very very very very very long text;"
}
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": "Read"
},
"type": "default"
}
]
}
]
}
set_card_callback(conf, handle)
app = Flask(__name__)
@app.route('/webhook/card', methods=['POST'])
def webhook_card():
oapi_request = OapiRequest(uri=request.path, body=request.data, header=OapiHeader(request.headers))
resp = make_response()
oapi_resp = handle_card(conf, oapi_request)
resp.headers['Content-Type'] = oapi_resp.content_type
resp.data = oapi_resp.body
resp.status_code = oapi_resp.status_code
return resp
# 设置 "开发者后台" -> "应用功能" -> "机器人" 消息卡片请求网址:https://domain/webhook/card
# startup event http server, port: 8089
if __name__ == '__main__':
app.run(port=8089, host="0.0.0.0")
from larksuiteoapi import Config
# 防止应用信息泄漏,配置环境变量中,变量(4个)说明:
# APP_ID:"开发者后台" -> "凭证与基础信息" -> 应用凭证 App ID
# APP_SECRET:"开发者后台" -> "凭证与基础信息" -> 应用凭证 App Secret
# VERIFICATION_TOKEN:"开发者后台" -> "事件订阅" -> 事件订阅 Verification Token
# ENCRYPT_KEY:"开发者后台" -> "事件订阅" -> 事件订阅 Encrypt Key
# HELP_DESK_ID: 服务台设置中心 -> ID
# HELP_DESK_TOKEN: 服务台设置中心 -> 令牌
# 企业自建应用的配置,通过环境变量获取应用配置
app_settings = Config.new_internal_app_settings_from_env()
# 应用商店应用的配置,通过环境变量获取应用配置
app_settings = Config.new_isv_app_settings_from_env()
# 参数说明:
# AppID、AppSecret: "开发者后台" -> "凭证与基础信息" -> 应用凭证(App ID、App Secret)
# VerificationToken、EncryptKey:"开发者后台" -> "事件订阅" -> 事件订阅(Verification Token、Encrypt Key)
# HelpDeskID、HelpDeskToken:服务台设置中心 -> ID、令牌
# 企业自建应用的配置
app_settings = Config.new_internal_app_settings(app_id="AppID", app_secret="AppSecret",
verification_token="VerificationToken", encrypt_key="EncryptKey",
help_desk_id="HelpDeskID", help_desk_token="HelpDeskToken")
# 应用商店应用的配置
app_settings = Config.new_isv_app_settings(app_id="AppID", app_secret="AppSecret",
verification_token="VerificationToken", encrypt_key="EncryptKey",
help_desk_id="HelpDeskID", help_desk_token="HelpDeskToken")
app_ticket
,会保存到存储中,所以存储(Store)的实现需要支持分布式存储。from larksuiteoapi import Config, AppSettings,Logger, DefaultLogger, MemoryStore, LEVEL_DEBUG, LEVEL_INFO, LEVEL_WARN,\
LEVEL_ERROR,DOMAIN_FEISHU, DOMAIN_LARK_SUITE
# for Cutome APP(企业自建应用)
app_settings = Config.new_internal_app_settings_from_env()
# 参数说明:
# domain:DOMAIN_FEISHU / DOMAIN_LARK_SUITE / 其他域名地址
# app_settings:应用配置
# logger:[Logger](src/larksuiteoapi/logger.py),默认日志实现:python logging
# 当使用默认日志时,需要设置 logging.basicConfig(...) // import logging
# log_level:输出的日志级别 LEVEL_DEBUG/LEVEL_INFO/LEVEL_WARN/LEVEL_ERROR,默认:LEVEL_ERROR
# store: [Store](src/larksuiteoapi/store.py),用来存储 app_ticket/access_token,默认:内存存储,适合轻量的使用(不合适:应用商店应用)
conf = Config(DOMAIN_FEISHU, app_settings, logger=DefaultLogger(), log_level=LEVEL_ERROR, store=MemoryStore())
from larksuiteoapi import ACCESS_TOKEN_TYPE_APP, ACCESS_TOKEN_TYPE_TENANT, ACCESS_TOKEN_TYPE_USER
from larksuiteoapi.api import Request, FormData, FormDataFile,set_path_params, set_query_params, set_timeout, set_no_data_field,\
set_user_access_token, set_tenant_key, set_is_response_stream, set_response_stream, set_need_help_desk_auth
# 参数说明:
# http_path:API路径
# 例如:https://domain/open-apis/contact/v3/users/:user_id
# 支持域名之后的路径,则 http_path:"/open-apis/contact/v3/users/:user_id"(推荐)
# 也支持全路径,则 http_path:"https://domain/open-apis/contact/v3/users/:user_id"
# 也支持 /open-apis/ 之后的路径,则 http_path:"contact/v3/users/:user_id"
# http_Method: GET/POST/PUT/BATCH/DELETE
# access_token_type:API使用哪种访问凭证,取值范围:ACCESS_TOKEN_TYPE_APP, ACCESS_TOKEN_TYPE_TENANT, ACCESS_TOKEN_TYPE_USER
# request_body:请求体(可能是 FormData()(例如:文件上传)),如果不需要请求体(例如一些GET请求),则传:nil
# request_opts:请求选项,一些不常用的参数封装,如下:
# set_path_params({"user_id": 4}):设置URL Path参数(有:前缀)值,当httpPath="contact/v3/users/:user_id"时,请求的URL="https://{domain}/open-apis/contact/v3/users/4"
# set_query_params({"age":4,"types":[1,2]}):设置 URL query,会在url追加?age=4&types=1&types=2
# set_is_response_stream(),设置响应是否是流,response.data = bytes(文件内容)
# set_response_stream(IO[any]),设置响应是否是流且响应流写入目标IO,response.data = 目标IO
# set_no_data_field(), 有一些 API 的响应体没有`data`字段,需要设置
# set_tenant_key(str),以`应用商店应用`身份,表示使用`tenant_access_token`访问API,需要设置
# set_user_access_token(str),表示使用`user_access_token`访问API,需要设置
# set_timeout(int),设置请求超时时间(单位:秒)
# set_need_help_desk_auth() ,表示是服务台API,需要设置 config.AppSettings 的 help desk 信息
# (str, str, str, Any, T, List[Callable[[Option], Any]]) -> None
req = Request(http_path, http_Method, access_token_type, request_body, request_opts=None)
由于SDK已经封装了app_access_token、tenant_access_token的获取,所以在调业务API的时候,不需要去获取app_access_token、tenant_access_token。如果业务接口需要使用user_access_token,需要进行设置(request.SetUserAccessToken(" UserAccessToken")),具体请看 README.zh.md -> 如何构建请求(Request)
from larksuiteoapi.api import Request
req = Request(...)
# 发送请求
# 参数说明:
# conf:整体的配置(Config)
# 返回值说明:
# Response:请求的结果(= http response body)
resp = req.do(conf)
print('request id = %s' % resp.get_request_id())
print(resp.code)
if resp.code == 0:
print(resp.data.message_id)
else:
print(resp.msg)
print(resp.error)
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。