Python API

简介

开发者会在 Scriptlet Block 中调用 SDK，与 OOCANA 工作流引擎交互。本文档主要介绍常用 API 及其使用方式。

提示

如果你的 block 需要调用 Fusion 托管能力，应该使用 oomol-fusion-sdk，并通过 token=await context.oomol_token() 和 base_url=context.fusion_api_url 初始化。

可继续参考 Fusion SDK 概览 和 Fusion SDK Python 文档。

快速开始

在 Scriptlet Block 中使用 SDK 时，需要导入 Context 对象，并通过该对象调用相关 API。

Scriptlet 模板

下面是 Scriptlet 的标准模板函数。你需要在函数体内编写执行代码，函数的输入输出类型也可以按需自定义，并支持导入第三方库。 如果要向后续 block 传递数据，像普通函数一样返回一个字典即可。OOCANA 会把第一层 key 视为对应的输出 handle，value 则是这个 handle 的输出值。

from oocana import Context

def main(params: dict, context: Context):

    # 此处编写需要执行的代码
    # 例如，向下游传递一个 output 数据，值为 None。
    return { "output": None }

核心 API

    def finish(self, *, result: Dict[str, Any] | None = None, error: str | None = None):
        """
        结束当前块的执行，并将结果发送给 oocana。
        如果 error 不为 None，则以错误状态结束当前块。
        如果 result 不为 None，则以指定结果结束当前块。
        如果 error 和 result 均为 None，则正常结束当前块且不返回结果。
        """

    def output(self, key: str, value: Any):
        """
        将数据输出到下游块。

        key: str，输出的键名，需在块的输出定义（handle 字段）中声明
        value: Any，输出的值
        """

    def outputs(self, outputs: Dict[str, Any]):
        """
        批量将数据输出到下游块。

        outputs: Dict[str, Any]，输出的键值对，键需在块的输出定义（handle 字段）中声明
        """

    # 启用数据预览功能
    def preview(self, payload: PreviewPayload):
    # 获取会话临时目录路径，可用于存储临时文件
    def session_dir(self) -> str:
    # 报告执行进度（0-100），内置节流机制，每300ms触发一次
    def report_progress(self, progress: float | int):
    # 获取当前 OOMOL token，用于调用 Fusion 等 OOMOL 自有服务
    async def oomol_token(self) -> str:
    # 当前运行时的 Fusion API base URL
    fusion_api_url: str

Context.preview 类型

context.preview 是一个核心 API，用于预览各种类型的数据。该 API 支持几乎所有常见数据格式的预览。

PreviewPayload 类型定义

函数签名：context.preview

    def preview(self, payload: PreviewPayload):
    # 仅支持 Python
    payload = self.__dataframe(payload)
    # 仅支持 Python
    payload = self.__matplotlib(payload)

    self.__mainframe.report(
        self.block_info,
        {
            "type": "BlockPreview",
            "payload": payload,
        },
    )

参数类型：PreviewPayload

     PreviewPayload: TypeAlias = Union[
        TablePreviewPayload,
        TextPreviewPayload,
        JSONPreviewPayload,
        ImagePreviewPayload,
        MediaPreviewPayload,
        DataFrame,
        PandasPreviewPayload,
        DefaultPreviewPayload
    ]

    class TablePreviewData(TypedDict):
        columns: List[str | int | float]
        rows: List[List[str | int | float | bool]]
        row_count: int | None

    class TablePreviewPayload(TypedDict):
        type: Literal['table']
        data: TablePreviewData | Any

    class TextPreviewPayload(TypedDict):
        type: Literal["text"]
        data: Any

    class JSONPreviewPayload(TypedDict):
        type: Literal["json"]
        data: Any

    class ImagePreviewPayload(TypedDict):
        type: Literal['image']
        data: str | List[str]

    class MediaPreviewPayload(TypedDict):
        type: Literal["image", 'video', 'audio', 'markdown', "iframe", "html"]
        data: str

    class PandasPreviewPayload(TypedDict):
        type: Literal['table']
        data: DataFrame

    class DefaultPreviewPayload:
        type: str
        data: Any

Preview 使用示例

  context.preview({
    # 支持的类型: "image" | "video" | "audio" | "markdown" | "iframe" | "html" | "json" | "text" | `text/${string}` | "table"
    "type": "video",
    # 数据可以是: 文件路径、base64 编码或 pandas DataFrame
    "data": "https://www.w3schools.com/html/mov_bbb.mp4"
  })

提示

以下功能仅支持 Python

    import pandas as pd

    df = pd.DataFrame({
        'A': [1, 2, 3],
        'B': [4, 5, 6],
        'C': [7, 8, 9]
    })
     context.preview(df)

    import matplotlib.pyplot as plt
    import numpy as np

    x = np.linspace(0, 10, 100)
    y = np.sin(x)
    plt.plot(x, y)
    plt.show()

    import plotly.graph_objects as go

    fig = go.Figure(data=[go.Bar(y=[2, 3, 1])])
    fig.show()

Context 类型定义

OOCANA SDK 提供的完整 API 类型描述，帮助开发者准确理解和使用这些接口。

class OOMOL_LLM_ENV(TypedDict):
    base_url: str
    base_url_v1: str
    api_key: str
    models: list[str]

class HostInfo(TypedDict):
    gpu_vendor: str
    gpu_renderer: str

class Context:
    def __init__(self, inputs: dict[str, Any], blockInfo: BlockInfo, mainframe: Mainframe, store, outputs, session_dir: str, tmp_dir: str, package_name: str, pkg_dir: str) -> None: ...
    @property
    def logger(self) -> logging.Logger: ...
    @property
    def session_dir(self) -> str: ...
    @property
    def tmp_dir(self) -> str: ...
    @property
    def tmp_pkg_dir(self) -> str: ...
    @property
    def pkg_dir(self) -> str: ...
    @property
    def keepAlive(self): ...
    @property
    def inputs(self): ...
    @property
    def session_id(self): ...
    @property
    def job_id(self): ...
    @property
    def job_info(self) -> JobDict: ...
    @property
    def block_info(self) -> BlockDict: ...
    @property
    def node_id(self) -> str: ...
    @property
    def oomol_llm_env(self) -> OOMOL_LLM_ENV: ...
    @property
    def host_info(self) -> HostInfo: ...
    @property
    def host_endpoint(self) -> str | None: ...
    @property
    def fusion_api_url(self) -> str: ...
    async def oomol_token(self) -> str: ...
    def output(self, key: str, value: Any): ...
        """
        将数据输出到下游块

        key: str，输出的键名，需在块的输出定义（handle 字段）中声明
        value: Any，输出的值
        """
    def outputs(self, outputs: dict[str, Any]): ...
        """
        批量将数据输出到下游块

        outputs: Dict[str, Any]，输出的键值对，键需在块的输出定义（handle 字段）中声明
        """
    def finish(self, *, result: dict[str, Any] | None = None, error: str | None = None): ...
        """
        结束当前块的执行，并将结果发送给 oocana。
        如果 error 不为 None，则以错误状态结束当前块。
        如果 result 不为 None，则以指定结果结束当前块。
        如果 error 和 result 均为 None，则正常结束当前块且不返回结果。
        """
    def send_message(self, payload) -> None: ...
    def preview(self, payload: PreviewPayload): ...
    def report_progress(self, progress: float | int): ...
        """报告进度

        此 API 用于报告块的执行进度，仅影响 UI 进度条，不影响实际进度。
        此 API 内置节流机制，最小间隔为 0.3 秒。
        第一次调用时会立即上报进度，之后每隔节流周期结束时才会上报一次。

        |       0.25 秒        |   0.2 秒  |
        第一次调用    第二次调用    第三次调用  第四五六七次调用
        |                     |          |          | | | |
        | -------- 0.3 秒 -------- | -------- 0.3 秒 -------- |
        上报                  上报                    上报
        :param float | int progress: 块的进度值，取值范围应为 [0, 100]。
        """
    def report_log(self, line: str, stdio: str = 'stdout'): ...
    def log_json(self, payload) -> None: ...
    def send_warning(self, warning: str): ...
    def send_error(self, error: str): ...
    def error(self, error: str): ...