BKFlow SDK

English | 简体中文

📖 项目简介

BKFlow SDK 是一个用于集成蓝鲸流程引擎（BlueKing Flow Engine）的 Django 应用程序 SDK。它提供了一套完整的 API 接口和客户端，帮助开发者快速在自己的 Django 项目中对接 BKFlow 流程编排能力。

✨ 核心特性

• 🚀 快速集成：只需简单配置即可在 Django 项目中使用
• 🔌 完整的 API 封装：提供流程模板、任务、插件等完整的 API 接口
• 🎯 灵活的扩展：支持通过 Signal 机制自定义业务逻辑
• 🛡️ 安全认证：内置完善的认证和权限管理
• 🏢 多租户适配：支持单租户和全租户 SaaS，并自动向 BKFlow 网关传递可信租户
• 📊 丰富的功能：支持流程模板管理、任务执行、插件系统、变量管理等
• 🔧 自定义请求头：支持在 API 调用时传入自定义 headers，灵活控制请求行为

🎯 主要功能

• 流程模板管理：创建、查询、更新、删除流程模板，流程模板列表查询和筛选，流程操作记录追踪
• 任务管理：创建和执行任务，查询任务状态和详情，批量获取任务状态，任务 Mock 数据获取
• 插件系统：内置插件管理，第三方插件集成，插件元信息查询，决策表插件支持
• 变量管理：系统变量管理，自定义变量管理，变量引用分析，变量预览功能

🚀 快速开始

📦 安装

pip install bkflow-sdk

⚙️ 配置

1. 添加到 Django 项目

在 Django 项目的 settings.py 中添加配置：

INSTALLED_APPS = [
    # ... 其他应用
    "bkflow.interface",
]

# 蓝鲸应用配置（必填）
APP_CODE = "your-app-code"  # 蓝鲸应用ID
SECRET_KEY = "your-secret-key"  # 蓝鲸应用密钥

BKFlow SDK 配置支持两种方式（任选其一或混合使用）：

方式1：直接在 Django settings 中声明（推荐）

# BKFlow SDK 配置（必填）
BKFLOW_SDK_APIGW_HOST = "https://your-bkflow-api-gateway.com"  # BKFlow API 网关地址
BKFLOW_SDK_DEFAULT_SPACE_ID = 1  # 默认空间ID（可选，如果不配置 BKFLOW_SDK_SPACE_TRANSFORMER 则必须配置）
BKFLOW_SDK_SPACE_TRANSFORMER = "your_module.path.function_name"  # 可执行函数路径（可选），用于根据 scope_type 和 scope_value 动态获取 space_id，格式：module.path.function_name 或 module.path.ClassName.method_name
BKFLOW_SDK_APIGW_HEADERS_GENERATOR = "your_module.path.function_name"  # 可执行函数路径（可选），用于根据 request 动态获取 headers，格式：module.path.function_name 或 module.path.ClassName.method_name

方式2：在 BKFLOW_SDK 字典中配置（优先级更高）

# BKFlow SDK 配置（必填）
BKFLOW_SDK = {
    "BKFLOW_SDK_APIGW_HOST": "https://your-bkflow-api-gateway.com",  # BKFlow API 网关地址
    "BKFLOW_SDK_DEFAULT_SPACE_ID": 1,  # 默认空间ID（可选，如果不配置 BKFLOW_SDK_SPACE_TRANSFORMER 则必须配置）
    "BKFLOW_SDK_SPACE_TRANSFORMER": "your_module.path.function_name",  # 可执行函数路径（可选），用于根据 scope_type 和 scope_value 动态获取 space_id
    "BKFLOW_SDK_APIGW_HEADERS_GENERATOR": "your_module.path.function_name",  # 可执行函数路径（可选），用于根据 request 动态获取 headers
}

配置优先级说明：

• 如果同时使用两种方式配置相同的配置项，BKFLOW_SDK 字典中的配置优先级更高
• 两种方式可以混合使用，未在字典中配置的项会从直接声明的配置中读取
• 推荐使用方式1（直接声明），配置更简洁直观

2. 配置 URL 路由

在项目的 urls.py 中添加路由：

# urls.py
from django.urls import include, path

urlpatterns = [
    # ... 其他路由
    path("api/bkflow/", include("bkflow.interface.urls")),
]

3. 环境变量配置（可选）

# .env 文件
export BKFLOW_SDK_APIGW_HOST="https://your-bkflow-api-gateway.com"
export BKFLOW_SDK_DEFAULT_SPACE_ID=1
export BKFLOW_SDK_SPACE_TRANSFORMER="your_module.path.function_name"  # 可选，用于动态获取 space_id，支持 module.path.function_name 或 module.path.ClassName.method_name
export BKFLOW_SDK_APIGW_HEADERS_GENERATOR="your_module.path.function_name"  # 可选，用于动态获取 headers，支持 module.path.function_name 或 module.path.ClassName.method_name
export BK_APIGW_STAGE_NAME="prod"  # 环境：prod/stag/test

蓝鲸 PaaS 部署应用时还会自动注入以下变量，通常不需要业务自行配置：

export BKPAAS_MULTI_TENANT_MODE="true"  # 是否启用多租户模式
export BKPAAS_APP_TENANT_ID="tenant-a"  # 单租户应用固定租户；全租户应用为空字符串

BKPAAS_MULTI_TENANT_MODE 是模式开关，不是租户 ID。全租户应用的 BKPAAS_APP_TENANT_ID 为空是正常现象，用户请求的实际租户来自 request.user.tenant_id。

4. 配置自定义函数（可选）

BKFLOW_SDK_SPACE_TRANSFORMER 示例

# utils/space.py
def get_space_id(scope_type=None, scope_value=None):
    """根据 scope_type 和 scope_value 返回 space_id"""
    if scope_type == "project":
        return int(scope_value) + 1000  # 示例：项目ID映射
    return None  # 返回 None 时使用 BKFLOW_SDK_DEFAULT_SPACE_ID

# settings.py
BKFLOW_SDK_SPACE_TRANSFORMER = "utils.space.get_space_id"

BKFLOW_SDK_APIGW_HEADERS_GENERATOR 示例

# utils/headers.py
import json
from django.conf import settings

def generate_headers(request):
    """根据 request 生成 headers"""
    return {
        "X-Bkapi-Authorization": json.dumps({
            "bk_app_code": settings.APP_CODE,
            "bk_app_secret": settings.SECRET_KEY,
            "bk_ticket": request.COOKIES.get("bk_ticket", ""),
        })
    }

# settings.py
BKFLOW_SDK_APIGW_HEADERS_GENERATOR = "utils.headers.generate_headers"

📚 基本使用

使用 API 客户端

from bkflow.client.core import get_client_by_user

# 获取客户端实例
client = get_client_by_user("admin")

# 获取流程模板列表
templates = client.bkflow.list_templates(
    path_params={"space_id": 1},
    limit=20,
    offset=0
)

# 创建流程模板
template = client.bkflow.create_template(
    path_params={"space_id": 1},
    name="测试流程",
    pipeline_tree={
        "activities": {},
        "constants": {},
        "gateways": {},
    },
    operator="admin"
)

# 创建任务
task = client.bkflow.create_task(
    {
        "template_id": 100,
        "name": "测试任务",
        "creator": "admin",
        "description": "这是一个测试任务",
        "constants": {"key1": "value1"}
    },
    path_params={"space_id": 1}
)

多租户调用

SDK 会在 Client 创建时解析租户，并为所有 BKFlow 网关 API 统一添加 X-Bk-Tenant-Id。业务不需要在模板、任务、插件或调试等 API 上逐个传入租户头。

用户请求

使用 BlueApps 多租户认证时，用户对象需要提供可信的 tenant_id。SDK 会读取 request.user.tenant_id：

from bkflow.client.core import get_client_by_request


def list_templates(request):
    client = get_client_by_request(request)
    return client.bkflow.list_templates(path_params={"space_id": 1})

使用 SDK 自带的 Django interface 路由时，内部的 get_redirect_client(request) 和 get_redirect_client_with_auth(request) 也会执行相同的租户解析，因此不需要修改每一个本地 View。

也可以直接传入带 tenant_id 的用户对象：

from bkflow.client.core import get_client_by_user

client = get_client_by_user(request.user)

后台任务、定时任务和管理命令

单租户应用在没有用户上下文时，会回退到 PaaS 注入的 BKPAAS_APP_TENANT_ID。全租户应用没有固定应用租户，因此必须传入来自可信任务数据或数据库记录的 tenant_id：

from bkflow.client.core import get_client_by_user

client = get_client_by_user("admin", tenant_id="tenant-a")
task = client.bkflow.create_task(
    {
        "template_id": 100,
        "name": "租户 A 的后台任务",
        "creator": "admin",
    },
    path_params={"space_id": 1},
)

全租户应用在没有用户租户、也没有显式 tenant_id 时会在发起网络请求前报错，避免请求被错误地发送到默认租户。不要直接信任浏览器请求参数中的租户 ID；如果用户上下文和显式租户不一致，SDK 会拒绝创建 Client。

兼容行为

• 未运行在支持多租户变量的 PaaS 环境，且用户对象没有 tenant_id 时，SDK 保持旧行为，不添加租户头。
• 旧的 X-Bkapi-Tenant-Id 输入会被规范化为 X-Bk-Tenant-Id，出站请求只发送标准头。
• 单次 API 调用不能用自定义 headers 覆盖 Client 已解析出的租户；需要切换租户时应重新创建 Client。

三个 Header 的职责相互独立：

使用自定义 Headers

对于需要 token 认证的接口，需要在 headers 中传入 token：

from bkflow.client.core import get_client_by_user
from bkflow.config.default import API_TOKEN_HEADER_KEY

client = get_client_by_user("admin")

# 调用需要 token 的接口
template = client.bkflow.fetch_template(
    path_params={"template_id": 100},
    headers={API_TOKEN_HEADER_KEY: "your-token-value"}
)

# 传入自定义 headers
result = client.bkflow.list_templates(
    path_params={"space_id": 1},
    headers={
        "X-Request-ID": "req-12345",
        "X-Client-Version": "1.0.0",
    }
)

⚠️ 重要注意事项

1. 配置要求

• 必填配置：

  • BKFLOW_SDK_APIGW_HOST：BKFlow API 网关地址，必须正确配置
  • APP_CODE 和 SECRET_KEY：蓝鲸应用认证信息，必须配置

• 可选配置：

  • BKFLOW_SDK_DEFAULT_SPACE_ID：默认空间ID，如果不配置 BKFLOW_SDK_SPACE_TRANSFORMER 则必须配置此选项
  • BKFLOW_SDK_SPACE_TRANSFORMER：可执行函数路径，格式为 module.path.function_name 或 module.path.ClassName.method_name，用于根据 scope_type 和 scope_value 动态获取 space_id。该函数接收两个参数：scope_type 和 scope_value，返回 space_id。如果配置了此选项，将优先使用此函数获取 space_id，否则使用 BKFLOW_SDK_DEFAULT_SPACE_ID
  • BKFLOW_SDK_APIGW_HEADERS_GENERATOR：可执行函数路径，格式为 module.path.function_name 或 module.path.ClassName.method_name，用于根据 request 动态获取 headers。该函数接收一个参数：request（Django request 实例），返回一个 headers 字典。如果配置了此选项，get_redirect_client_with_auth 将优先使用此函数获取 headers，否则使用默认的 headers 生成逻辑（从 cookies 和 META 中获取认证信息）
  • BK_APIGW_STAGE_NAME：API 网关环境，默认为 prod

配置方式说明：

• 所有以 BKFLOW_SDK 前缀开头的配置项支持两种声明方式：
  1. 直接声明：在 settings.py 中直接声明 BKFLOW_SDK_XXX = value
  2. 字典配置：在 BKFLOW_SDK 字典中配置 BKFLOW_SDK = {"BKFLOW_SDK_XXX": value}
• 如果同时使用两种方式配置相同的配置项，BKFLOW_SDK 字典中的配置优先级更高
• 两种方式可以混合使用，推荐使用直接声明方式，配置更简洁直观

2. Token 认证

部分接口需要 token 认证（标记为 token_required=True），调用这些接口时必须在 headers 中传入 BKFLOW-TOKEN：

from bkflow.config.default import API_TOKEN_HEADER_KEY

# 需要 token 的接口示例
client.bkflow.fetch_template(
    path_params={"template_id": 100},
    headers={API_TOKEN_HEADER_KEY: "your-token"}
)

3. 流程调试接口

调试接口用于平台方嵌入 BKFlow 流程调试页面或在本地后端转发调试能力。调用前应由平台方后端通过 BKFlow apply_token 申请 resource_type=TEMPLATE 或 SCOPE、permission_type=MOCK 的 token，并在请求头中传入 BKFLOW-TOKEN。

Python Client 示例：

from bkflow.config.default import API_TOKEN_HEADER_KEY

headers = {API_TOKEN_HEADER_KEY: "your-mock-token"}

context = client.bkflow.debug_context(
    space_id=1,
    template_id=100,
    headers=headers,
)

run_result = client.bkflow.debug_global_run(
    space_id=1,
    template_id=100,
    inputs={"${input}": "value"},
    headers=headers,
)

step_result = client.bkflow.debug_step_run(
    space_id=1,
    template_id=100,
    node_id="node1",
    mode="mock",
    mock_outputs={"k": "v"},
    headers=headers,
)

Django interface app 也提供对应本地转发接口：

所有调试接口都需要传入 space_id 和 template_id。

4. Headers 合并规则

• BaseAPIClient：调用时传入的 headers 会覆盖客户端实例的 headers
• RequestAPIClient：客户端实例的 headers 会覆盖调用时传入的 headers
• 租户头例外：Client 已解析出的 X-Bk-Tenant-Id 不允许被单次 API 调用覆盖

5. 错误处理

SDK 会自动处理 API 调用异常，返回统一的错误格式：

{
    "result": False,
    "message": "错误信息",
    "data": None
}

6. 用户认证

使用 get_client_by_user() 时，需要确保：

• 用户已通过蓝鲸认证
• 已正确配置 bkoauth 或使用 bk_ticket/bk_token 进行认证
• 全租户应用的用户对象能够提供可信的 tenant_id；无用户上下文时显式传入 tenant_id

📋 版本历史

查看 release.md 了解版本更新历史。

当前版本

• 最新版本 - 功能更新
  • ✨ 多租户支持：根据用户、显式参数和 PaaS 环境变量统一传递租户上下文
  • ✨ 新增功能：支持在 API 调用时传入自定义 headers
  • 🐛 Bug 修复：修复了 token 验证逻辑
  • ✅ 测试完善：添加了自定义 headers 功能的完整测试用例

📄 许可证

本项目采用 MIT 许可证 - 详见 LICENSE 文件

🔗 相关链接

• 蓝鲸智云官网
• 项目主页
• 问题反馈

💬 支持

如果您在使用过程中遇到问题，可以通过以下方式获取帮助：

• 提交 Issue
• 查看项目文档
• 加入蓝鲸社区讨论

Copyright © 2022 THL A29 Limited, a Tencent company. All Rights Reserved.