AutoGPT Platform Block SDK 实战指南：用 ProviderBuilder 与 Block 基类构建高级功能块

本篇技术指南以 docs/platform/block-sdk-guide.md 为骨架，围绕 AutoGPT Platform 的 Block SDK 开发模式展开，讲解如何从零编写带 Provider 配置、输入/输出 Schema、OAuth/Webhook 支持、媒体文件处理与内建测试能力的可复用 Block。读完你将掌握 AutoGPT Platform（前端可编排、后端执行的自研 Agent 平台，代码位于 autogpt_platform/）中 Block 的完整开发闭环，并能在现有仓库中找到全部源码依据。

什么是 Block？Block SDK 解决了什么问题

在 AutoGPT Platform 中，Block 是工作流里执行具体任务的复用组件：它们可以对接外部服务、处理数据，或执行任意程序化操作。一个 Agent（可视化的"图"）由多个 Block 通过输入/输出引脚串联而成。

Block SDK 是面向 Block 开发者的统一 Python 开发入口。它并不只是某一个模块，而是一层"聚合导出层"——所有 Block 开发所需的基础类、凭据组件、成本追踪组件、Webhook 组件与工具函数，都统一从 backend.sdk 导入：

from backend.sdk import *  # 完整 re-export，见 backend/sdk/__init__.py

在 backend/sdk/init.py 中可以看到 SDK 的完整导出清单，主要包括：

• 核心 Block 体系：Block、BlockCategory、BlockOutput、BlockSchema、BlockSchemaInput、BlockSchemaOutput、BlockType；
• Schema 与模型组件：SchemaField、Credentials、CredentialsMetaInput、APIKeyCredentials、OAuth2Credentials、UserPasswordCredentials；
• 成本体系：BlockCost、BlockCostType、block_usage_cost；
• 集成组件：ProviderName、BaseWebhooksManager、BaseOAuthHandler、Webhook；
• 工具函数：store_media_file、Requests、json、TextFormatter、TruncatedLogger；
• SDK 新组件：ProviderBuilder、Provider、AutoRegistry、BlockConfiguration、cost。

也就是说，Block 开发者几乎不需要关心各子模块的内部组织，只需面向 backend.sdk 这一稳定出口编程。

基础结构：三个步骤搭建一个 Block

一个基于 SDK 的 Block 通常由Provider 配置与Block 类实现两部分组成。先通过一个最小可用 API key Provider 示例建立整体认识。

1. 创建 Provider 配置（_config.py）

用 ProviderBuilder 声明你的服务商（Provider）身份与认证方式：

from backend.sdk import BlockCostType, ProviderBuilder

my_provider = (
    ProviderBuilder("my_provider")
    .with_api_key("MY_PROVIDER_API_KEY", "My Provider API Key")
    .with_base_cost(1, BlockCostType.RUN)
    .build()
)

.with_api_key(env_var_name, title) 会做两件事（见 builder.py）：

1. 把 "api_key" 加入该 Provider 支持的认证类型集合；
2. 通过 AutoRegistry.register_api_key(name, env_var) 注册环境变量到 Provider 的映射，并若环境变量中已存在该 key，自动注册一条 id 为 f"{name}-default" 的默认 API key 凭据。

2. 创建 Block 类（my_block.py）

实现类时声明 Input / Output 两个 Pydantic 风格的 Schema 类，并实现 async run() 方法：

import uuid
from backend.sdk import (
    APIKeyCredentials,
    Block,
    BlockCategory,
    BlockOutput,
    BlockSchemaInput,
    BlockSchemaOutput,
    CredentialsMetaInput,
    SchemaField,
)
from ._config import my_provider


class MyBlock(Block):
    class Input(BlockSchemaInput):
        credentials: CredentialsMetaInput = my_provider.credentials_field(
            description="API credentials for My Provider"
        )
        query: str = SchemaField(description="The query to process")
        limit: int = SchemaField(
            description="Number of results",
            default=10,
            ge=1,
            le=100,
        )
        advanced_option: str = SchemaField(
            description="Advanced setting",
            default="",
            advanced=True,
        )

    class Output(BlockSchemaOutput):
        results: list = SchemaField(description="List of results")
        count: int = SchemaField(description="Total count")

    def __init__(self):
        super().__init__(
            id=str(uuid.uuid4()),
            description="Brief description of what this block does",
            categories={BlockCategory.SEARCH},
            input_schema=self.Input,
            output_schema=self.Output,
        )

    async def run(
        self,
        input_data: Input,
        *,
        credentials: APIKeyCredentials,
        **kwargs
    ) -> BlockOutput:
        try:
            results = await self.process_data(
                input_data.query,
                input_data.limit,
                credentials
            )

            yield "results", results
            yield "count", len(results)

        except Exception as e:
            yield "error", str(e)

    async def process_data(self, query, limit, credentials):
        # 实际业务逻辑：调用外部 API、处理数据等
        return []

3. 在 __init__.py 中导出

在 Block 所在包（provider 目录）的 __init__.py 中导出 Block 实例。SDK 的 AutoRegistry 会自动收集并注册 Provider 与 Block（ProviderBuilder.build() 内部即调用 AutoRegistry.register_provider(provider)，见 builder.py）。

Input Schema 字段说明

• credentials：用 my_provider.credentials_field(description=...) 声明该字段，SDK 会在其 JSON Schema 上写入 credentials_provider 与 credentials_types 信息，前端据此渲染对应的凭据连接控件（源码见 provider.py）；
• query：带描述字符串的普通输入；
• limit：带校验约束的整型字段（ge=1 表示 ≥1，le=100 表示 ≤100）；
• advanced_option：标记 advanced=True 后，该字段在基础版 UI 中被折叠隐藏，只有展开"高级选项"才可见。

Output Schema 字段说明

• results：Block 输出的结果列表；
• count：结果总数；
• error 输出引脚由 BlockSchemaOutput 基类预定义，无需自行声明。源码中 BlockSchemaOutput 内置了 error: str = SchemaField(...)（见 blocks/_base.py），保证所有 Block 的错误行为一致，执行器能统一接收 error 输出。

Block 初始化参数

• id：用 uuid.uuid4() 生成全局唯一 ID（仓库要求每个 Block 有稳定唯一 ID，便于持久化与引用）；
• description：一句话说明该 Block 的功能；
• categories：从 BlockCategory 枚举中选择分类，用于前台可发现性与筛选。完整枚举见 blocks/_base.py，包括 AI、SOCIAL、TEXT、SEARCH、INPUT、OUTPUT、LOGIC、COMMUNICATION、DEVELOPER_TOOLS、DATA、HARDWARE、AGENT、CRM、SAFETY、PRODUCTIVITY、ISSUE_TRACKING、MULTIMEDIA、MARKETING 等，每个分类在枚举值里都附带了用途说明文案；
• input_schema / output_schema：绑定上面定义的 Input / Output 类。

run() 方法语义

• 将核心业务逻辑抽到 process_data() 之类的辅助方法中，run() 只负责编排与输出；
• 通过 credentials.api_key.get_secret_value() 获取明文 API key（凭据字段均为 pydantic SecretStr，必须显式调用 get_secret_value() 才能取到真实值）；
• 使用 yield 逐个产出输出引脚的数据，方法返回类型为 BlockOutput。

核心组件详解：ProviderBuilder、Schema 与 Block

ProviderBuilder 的能力全集

ProviderBuilder 提供流式（fluent）链式 API，各方法对应能力如下（全部方法定义见 builder.py）：

方法	作用	备注
.with_api_key(env_var, title)	注册 API key 认证	环境变量存在时自动生成默认凭据（builder.py）
.with_managed_api_key()	声明 API key 认证但不生成默认凭据	用于密钥由 ManagedCredentialProvider 按用户托管的场景，避免组织级 key 泄漏为"用户凭据"
.with_api_key_from_settings(settings_attr, title)	从平台 Settings 中复用既有 key	读取 Settings().secrets 对应属性
.with_oauth(handler, scopes, client_id_env_var, client_secret_env_var)	添加 OAuth 2.0 认证	两个 env var 缺一不可，缺失时仅记录 warning 且不启用 OAuth（见 builder.py）；不传时默认取 {provider}_CLIENT_ID / {provider}_CLIENT_SECRET（大写）
.with_user_password(username_env_var, password_env_var, title)	添加用户名/密码认证	两个环境变量都存在时才注册默认凭据
.with_webhook_manager(manager_class)	为 Provider 注册 Webhook 管理器	传入 BaseWebhooksManager 子类
.with_base_cost(amount, cost_type, cost_divisor=1)	设置该 Provider 下所有 Block 的基础计费	cost_divisor 仅作用于 SECOND/ITEMS；TOKENS 走按模型单价表，忽略 divisor（见 builder.py）
.with_api_client(factory)	注册 API 客户端工厂	之后可用 provider.get_api(credentials) 获取带凭据的客户端
.with_error_handler(handler)	注册 Provider 专属错误翻译器	handler: Callable[[Exception], str]，将异常转换为可读消息
.with_description(desc)	设置一句话描述	通过 GET /integrations/providers 暴露给前端展示
.with_supported_auth_types(*types)	额外声明可用的认证类型	通常 OAuth/API key/user_password 已自动加入集合，仅"认证逻辑位于 builder 链之外"时才需要调用
.with_config(**kwargs)	附加自定义配置	可通过 provider.get_config(key, default) 读取

.build() 会构造一个 Provider 对象（构造与默认凭据/测试凭据逻辑见 provider.py），并自动向 AutoRegistry 注册。

真实仓库中的 Provider 配置

下面是仓库里三种代表性 Provider 的完整配置，可作为高保真模板：

Firecrawl（纯 API key + 成本核算） —— backend/blocks/firecrawl/_config.py：

from backend.sdk import BlockCostType, ProviderBuilder

# Firecrawl 以其自有 credits 计费（1 credit ≈ $0.001）
firecrawl = (
    ProviderBuilder("firecrawl")
    .with_description("Web scraping and crawling")
    .with_api_key("FIRECRAWL_API_KEY", "Firecrawl API Key")
    .with_base_cost(1000, BlockCostType.COST_USD)  # 1000 平台积分 ≈ $1
    .build()
)

Linear（OAuth + API key 双认证） —— backend/blocks/linear/_config.py：

from backend.sdk import BlockCostType, ProviderBuilder
from ._oauth import LinearOAuthHandler

linear = (
    ProviderBuilder("linear")
    .with_description("Issues and project tracking")
    .with_api_key(env_var_name="LINEAR_API_KEY", title="Linear API Key")
    .with_base_cost(1, BlockCostType.RUN)
    .with_oauth(
        LinearOAuthHandler,
        scopes=[LinearScope.READ, LinearScope.WRITE,
                LinearScope.ISSUES_CREATE, LinearScope.COMMENTS_CREATE],
        client_id_env_var="LINEAR_CLIENT_ID",
        client_secret_env_var="LINEAR_CLIENT_SECRET",
    )
    .build()
)

Linear 的 _config.py 还把 OAuth scope 定义成了 LinearScope(str, Enum)，包含 read、write、issues:create、comments:create、timeSchedule:write、admin，便于在多个 Block 间复用与校验。

Exa（API key + Webhook） —— backend/blocks/exa/_config.py：

from backend.sdk import BlockCostType, ProviderBuilder
from ._webhook import ExaWebhookManager

exa = (
    ProviderBuilder("exa")
    .with_description("Neural web search")
    .with_api_key("EXA_API_KEY", "Exa API Key")
    .with_webhook_manager(ExaWebhookManager)
    .with_base_cost(100, BlockCostType.COST_USD)
    .build()
)

注意 Exa 的 ExaSearchBlock 等约 45 个共享该 Provider 配置的 Block 会把 API 返回的 cost_dollars.total 写入 NodeExecutionStats.provider_cost，因此这里按 COST_USD 计费，100 积分/美元、约 $0.01/积分。

BlockCostType 计费类型

BlockCostType 枚举（blocks/_base.py）支持六种计费口径：

类型	含义
RUN	每次运行固定 cost_amount 积分
BYTE	按输入数据字节数 × cost_amount 积分
SECOND	每 cost_divisor 秒运行时长计 cost_amount 积分
ITEMS	每 cost_divisor 个条目（来自执行统计）计 cost_amount 积分
COST_USD	按执行统计中 provider_cost（美元）× cost_amount 积分
TOKENS	按 (model, provider) 单价表计费，见 TOKEN_COST

例如文档建议的 .with_base_cost(1, BlockCostType.SECOND, cost_divisor=10) 即每 10 秒运行时长收 1 积分。

高级特性之一：为 Block 内建测试

SDK 的 Block 基类支持在初始化时直接声明 test_input / test_output / test_mock，让平台自动执行"输入 → mock 业务方法 → 校验输出"的单元测试，无需编写额外断言逻辑：

def __init__(self):
    super().__init__(
        # ... 其它配置 ...
        test_input={
            "query": "test query",
            "limit": 5,
            "credentials": {
                "provider": "my_provider",
                "id": str(uuid.uuid4()),
                "type": "api_key"
            }
        },
        test_output=[
            ("results", ["result1", "result2"]),
            ("count", 2)
        ],
        test_mock={
            "process_data": lambda *args, **kwargs: ["result1", "result2"]
        }
    )

• test_input：模拟图执行时注入该 Block 的输入字典，credentials 中给出 provider 名、凭据 id 与类型即可；
• test_output：期望的输出引脚序列 [(output_name, expected_value), ...]；
• test_mock：用 lambda 替换被测方法（这里是 process_data），使测试不真正访问外部网络。

在实际仓库中，Linear 等 Provider 会进一步把 mock 凭据定义为模块级常量，例如 TEST_CREDENTIALS_API_KEY / TEST_CREDENTIALS_INPUT_API_KEY（见 linear/_config.py），供 test_input 引用；若开发时不在 builder 中声明测试配置，Provider.get_test_credentials() 会根据首个支持的认证类型自动生成 mock 凭据（OAuth 生成带 mock token/scope 的 OAuth2Credentials，API key 生成 APIKeyCredentials，见 provider.py）。

高级特性之二：OAuth 与 Webhook 支持

编写 OAuth Handler（_oauth.py）

继承 BaseOAuthHandler，实现授权 URL 生成与授权码换 token 两个核心钩子：

from backend.integrations.oauth.base import BaseOAuthHandler

class MyProviderOAuthHandler(BaseOAuthHandler):
    PROVIDER_NAME = "my_provider"

    def _get_authorization_url(self, scopes: list[str], state: str) -> str:
        # 生成授权页 URL（拼接 client_id、redirect_uri、scopes、state 等）
        pass

    def _exchange_code_for_token(self, code: str, scopes: list[str]) -> dict:
        # 用授权码向 Provider 换取 access_token / refresh_token
        pass

仓库内可以参考 oauth 目录 下 Google、GitHub、Notion、Discord、Reddit、Todoist、Twitter、Linear 等成熟实现。注意：OAuth 是否能被启用还取决于部署环境中是否同时配置了 client_id_env_var 与 client_secret_env_var 指向的两个环境变量（见上表与 builder.py），未配置时平台会打印 warning 并退回只支持其它认证方式。

编写 Webhook Manager（_webhook.py）

继承 BaseWebhooksManager，核心是实现事件校验：

from backend.integrations.webhooks._base import BaseWebhooksManager

class MyProviderWebhookManager(BaseWebhooksManager):
    PROVIDER_NAME = "my_provider"

    async def validate_event(self, event: dict) -> bool:
        # 校验事件签名/结构，通过返回 True
        pass

仓库内已有多个可参考的实现，例如 exa/_webhook.py（Exa 的 webhook 管理器）以及 webhooks 目录 下的 GitHub、Telegram、Compass、Slant3D 等。SDK 还同时导出 BlockWebhookConfig、BlockManualWebhookConfig 与 ManualWebhookManagerBase，支持配置化（平台管理）与手动（用户自填 URL）两类 webhook 接入方式。

高级特性之三：媒体文件处理（store_media_file）

当 Block 需要处理图片、视频、文档等媒体文件时，不要直接拼本地路径或手动转 base64，而应统一调用 store_media_file()，按用途选择 return_format：

from backend.data.execution import ExecutionContext
from backend.util.file import store_media_file
from backend.util.type import MediaFileType

async def run(
    self,
    input_data: Input,
    *,
    execution_context: ExecutionContext,
    **kwargs,
):
    # 场景一：需要本地文件路径给 ffmpeg / MoviePy / PIL 等工具处理
    local_path = await store_media_file(
        file=input_data.video,
        execution_context=execution_context,
        return_format="for_local_processing",
    )

    # 场景二：需要把内容发给 Replicate、OpenAI 等外部 API（base64 data URI）
    image_b64 = await store_media_file(
        file=input_data.image,
        execution_context=execution_context,
        return_format="for_external_api",
    )

    # 场景三：把结果返回给用户/下一个 Block（自动适配上下文）
    result = await store_media_file(
        file=generated_url,
        execution_context=execution_context,
        return_format="for_block_output",
    )
    yield "image_url", result

三种 return_format 的适用语义总结：

返回格式	使用场景	返回内容
"for_local_processing"	本地工具（ffmpeg、MoviePy、PIL）需要文件路径	本地文件路径
"for_external_api"	发送给外部 API（Replicate、OpenAI 等）	base64 data URI
"for_block_output"	输出一律用这个	自动选择最佳格式

特别强调 "for_block_output" 的"自动适配"行为：当 Block 运行在 CoPilot 场景时输出 workspace:// 引用，运行在图中时则输出 data URI 数据——凡是作为 Block 输出的媒体，永远选择 for_block_output，让平台根据执行上下文决定存储与传输方式。相关实现位于 backend/util/file.py（store_media_file）与 backend/util/type.py（MediaFileType）。

常见开发模式

用 SDK Requests 发起 API 请求

SDK 统一导出了异步 HTTP 客户端 Requests（来自 backend.util.request），避免直接管理会话：

from backend.sdk import Requests

async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs):
    headers = {
        "Authorization": f"Bearer {credentials.api_key.get_secret_value()}",
        "Content-Type": "application/json"
    }

    response = await Requests().post(
        "https://api.example.com/endpoint",
        headers=headers,
        json={"query": input_data.query}
    )

    data = response.json()
    yield "results", data.get("results", [])

同一 Block 支持多种认证类型

当 Provider 同时注册了 OAuth 与 API key（如上面的 Linear），run() 的 credentials 参数会按类型注入，可用 isinstance 分支处理：

from backend.sdk import APIKeyCredentials, OAuth2Credentials

async def run(
    self,
    input_data: Input,
    *,
    credentials: OAuth2Credentials | APIKeyCredentials,
    **kwargs
):
    if isinstance(credentials, OAuth2Credentials):
        # OAuth：取访问令牌
        token = credentials.access_token.get_secret_value()
    else:
        # API key：取密钥
        token = credentials.api_key.get_secret_value()

各凭据类型的取值入口：

• OAuth2Credentials：credentials.access_token.get_secret_value()（另有 refresh_token、username、scopes 等字段）；
• APIKeyCredentials：credentials.api_key.get_secret_value()；
• UserPasswordCredentials：credentials.username.get_secret_value() 与 credentials.password.get_secret_value()。

错误处理规范

• 面向用户可修复的校验性/参数性错误应抛 BlockInputError（例如请求参数非法），运行时/依赖服务类错误抛 BlockExecutionError，二者均从 backend.util.exceptions 导入。文档强调：它们继承自 ValueError，因此执行器会把它们当作"用户可修正"的错误分类处理，而不是视为系统故障；
• 更细的错误分类策略请参见 docs/platform/new_blocks.md；
• 在 _base.py 的 Block 执行路径（blocks/_base.py）中可以看到大量 raise BlockExecutionError(...) / raise BlockInputError(...) 的真实用法，SDK 兜底逻辑会捕获 Exception 并把消息写到 error 输出引脚，保证图执行不会因单个 Block 异常而崩溃且能获得结构化错误。

文件组织规范

文档推荐的 Block 目录布局（仓库中所有 provider 目录都遵循该模式）：

backend/blocks/my_provider/          # 仓库实际位置：autogpt_platform/backend/backend/blocks/my_provider/
├── __init__.py          # 导出 Block 实例（AutoRegistry 自动注册）
├── _config.py           # Provider 配置（ProviderBuilder）
├── _oauth.py            # OAuth handler（可选）
├── _webhook.py          # Webhook 管理器（可选）
├── _api.py              # API 客户端封装（可选）
├── models.py            # 数据模型（可选）
└── my_block.py          # Block 实现

上表中的 /backend/blocks/ 对应仓库内实际目录 autogpt_platform/backend/backend/blocks/，其下每个子目录即一个 Provider 的 Block 集。

测试你的 Block

Block 的自动化测试集中在 backend/blocks/test/test_block.py（仓库路径 backend/blocks/test/test_block.py），该测试会遍历所有已注册 Block，用每个 Block 声明的 test_input / test_mock 执行并比对 test_output。在 backend 的 Poetry 环境中运行：

# 运行全部 Block 测试
poetry run pytest backend/blocks/test/test_block.py -xvs

# 只测特定 Block（占位符换成你的类名，如 test_available_blocks[MyBlock]）
poetry run pytest 'backend/blocks/test/test_block.py::test_available_blocks[MyBlock]' -xvs

提示：未配置 test_mock 的 Block 会真实调用外部服务并可能产生费用或网络依赖，因此务必为所有涉及外部调用的 Block 提供 mock 测试配置，这也是集成测试能稳定通过的前提。

集成 Checklist

按以下顺序核对，即可完成一个新 Block 的交付：

• [ ] 在 _config.py 中用 ProviderBuilder 创建 Provider 配置
• [ ] 实现带 Input/Output Schema 的 Block 类
• [ ] 用 uuid.uuid4() 生成唯一 Block ID
• [ ] 选择恰当的 BlockCategory 分类（提升可发现性）
• [ ] 实现 async run() 方法（复杂逻辑抽到辅助方法）
• [ ] 优雅处理错误（区分 BlockInputError / BlockExecutionError）
• [ ] 添加 test_input / test_output / test_mock 测试配置
• [ ] 在 __init__.py 中导出 Block
• [ ] 用 pytest backend/blocks/test/test_block.py 验证测试
• [ ] 在文档中说明任何特殊要求（如所需环境变量、scope 清单）

仓库参考实例

以下三个 Provider 分别是文档重点推荐的三种模式范例：

• 纯 API key：backend/blocks/firecrawl/ —— Firecrawl 爬虫类 Block，认证简单、计费清晰（COST_USD），并示范了 _api.py 与块拆分；
• OAuth + API key 双认证：backend/blocks/linear/ —— Linear 项目管理 Block，_config.py + _oauth.py + models.py + _api.py 布局完整，还演示了 scope 枚举与 mock 测试凭据常量；
• Webhook 支持：backend/blocks/exa/ —— Exa 神经搜索 Block 家族（约 45 个共享 Provider 配置），含 _webhook.py 与成本回写实现。

研究这些目录，可以快速理解 Provider 配置、Schema 声明、异步执行、凭据注入与成本统计之间是如何协同的，从而在自己的 Block 中复刻这些模式。