Skip to content

sdk.md

Latest commit

 

History

History
205 lines (150 loc) · 8.89 KB

sdk.md

File metadata and controls

205 lines (150 loc) · 8.89 KB
Raw

快速接入 SDK

Caploom SDK 的用途是：在旧业务外增加一个适配文件，业务规则保留在原系统， 同一份能力可以被人工代码、CLI、Web 和可选 Agent 调用。

这是一层位于既有 Capability Runtime 上的 Python API，不是新的 Agent 框架。 当前 SDK 是原路线图 Phase 2 中的开发者体验切片，不代表已经完成全部 Phase 2。

1. 先运行一个不需要 Agent 的例子

在仓库根目录执行：

uv sync --extra capabilities --frozen
uv run --extra capabilities python examples/sdk_quickstart.py

输出：

{"sku":"SKU-001","quantity":12}

examples/legacy_inventory.py 是标准库实现的独立库存 API； examples/sdk_quickstart.py 是它的接入文件。库存示例是第二个虚构业务域， 不是已经在真实企业完成部署的证明。

核心用法如下，完整模型和业务对象在可运行示例中：

from caploom.sdk import Capability, Provider, Runtime

# 输入、输出均为已有的 Pydantic BaseModel；写操作分类必须显式指定。
GET_STOCK = Capability(
    "inventory.available@1",
    description="Query available stock without modifying inventory",
    input=Query,
    output=Stock,
    side_effect="none",
)

provider = Provider("example.legacy-inventory", service=old_business)
provider.expose(GET_STOCK, lambda old, q: Stock(sku=q.sku, quantity=old.available(q.sku)))

async with Runtime([provider], bindings={GET_STOCK.id: provider.name}) as runtime:
    client = runtime.client(caller="web", allowed=[GET_STOCK.id])
    stock: Stock = await client.invoke(GET_STOCK, Query(sku="SKU-001"))

普通接入文件不需要使用 Kernel、Fiber、Registry、Invoker 或 PluginContext。 输入输出模型建议使用 ConfigDict(strict=True, extra="forbid")。SDK 复用 Pydantic 和原有 Invoker，仍会严格验证实际调用，不信任模型或客户端声称合法。 当前沿用 JSON 对象契约；复杂日期、二进制和任意对象应先显式映射为 JSON 字段。

2. 旧业务已经有契约，不必重写

contract = Capability.from_spec(existing_spec, input=ExistingInput, output=ExistingOutput)
provider.expose(contract, adapter)

from_spec 比较模型生成的输入输出 Schema 与既有 Spec；不匹配立即报错。 返回值保留原 Spec，而不是注册一份近似的新定义。

现有客服 Demo 使用这个路径：domain.py 和 contracts.py 保持原样， 仅将 providers.py 的重复注册代码换成 SDK Builder。Agent 的 Tool Schema 仍来自原始七项 Capability；没有另一套 Agent 专用业务契约。

3. 资源归谁所有，必须明确

借用旧对象： Provider(name, service=existing_service) 不调用旧对象的 close()。企业应用继续负责对象寿命；SDK 停止不等于关闭整个旧业务。

插件独占资源： 使用每次调用都会创建新上下文管理器的 factory。

from contextlib import asynccontextmanager


@asynccontextmanager
async def connection():
    client = await open_business_connection()
    try:
        yield client
    finally:
        await client.close()


provider = Provider("company.orders", resource=connection)

每次启用创建新资源。卸载时先禁止新调用，等待已接收调用完成，再关闭资源。 setup 部分失败也通过已有 Effect 机制清理。不要把一个已经进入的上下文管理器 实例重复传入，也不要把必须在同一 Task 退出的 TaskGroup/取消作用域作为跨生命周期 资源。资源 factory 应管理连接等可在后续生命周期操作中关闭的资源。

高级组合： ServiceRef("legacy.orders", OrderService) 表示借用声明过的内部 Service，生成的 Plugin 会声明该硬依赖。它不允许直接依赖某个业务 Provider 实例。 现有 Demo 的旧业务根对象使用这一方式，由原应用持有。

4. 同步和异步旧方法

expose 的回调是 (service, request) -> OutputModel，也可以是异步函数。

默认 execution="inline" 用于异步回调或确定很快、不会阻塞的同步函数。 它不偷偷改变旧对象的线程归属。阻塞 I/O 必须明确指定：

provider.expose(contract, blocking_adapter, execution="thread")

线程模式复用 asyncio.to_thread，服务必须允许从工作线程访问。线程不能被安全 强杀，因此取消或超时后仍等待它完成，再传播取消/超时；等待可能超过声明的超时。 这保证资源不会先被关闭，但不保证业务写入被撤销。具有副作用的远程调用必须 使用业务幂等键和事后状态查询，SDK 不自动重试。CPU 密集任务和不可信代码不适合 此模式，应在后续进程/远程 Adapter 中处理。

5. 错误、确认和 Provider 选择不做隐式推断

def map_error(error: Exception) -> CapabilityError | None:
    if isinstance(error, LegacyNotFound):
        return CapabilityError("NOT_FOUND", "Record not found")
    return None


provider = Provider("company.orders", service=old_business, error_mapper=map_error)

只翻译经过审查的领域错误，不应默认公开 str(error)。普通未知错误由 Invoker 转换为 PROVIDER_FAILED，取消和控制流异常不当作业务错误吞掉。

写契约必须指定 side_effect="write"，调用时显式 confirmed=True。这仅是调用 确认边界，不是用户身份认证或生产授权。退款规则、事务和幂等仍属于企业业务。

即使只有一个 Provider，也必须显式 Binding：

await runtime.bind({contract.id: "company.orders-v2"})

SDK 没有 expose-all，不根据函数名猜读写，不按最后注册顺序选择 Provider。

6. 生命周期和测试

Runtime 是一个应用拥有的单次 lifespan：启动后可切换已注册插件，关闭后创建新的 Runtime，而不是复活旧客户端。低层 Kernel 原有的重复 stop/start API 仍然保留。

runtime = Runtime([v1], bindings={contract.id: v1.name})
runtime.register(v2, enabled=False)  # 启动前登记，不执行 setup
async with runtime:
    client = runtime.client(caller="cli", allowed=[contract.id])
    await runtime.disable(v1.name)
    await runtime.enable(v2.name)
    await runtime.bind({contract.id: v2.name})

启用不等于绑定；停用时旧 Binding 被撤销，恢复后由调用者明确选择。 普通调用可并发，生命周期操作拒绝重叠或从同一在途调用 Task 中发起，避免自我等待。 回调不要创建另一个 Task 或线程来等待本 Runtime 的关闭操作；这类跨 Task 人为循环 仍须由应用避免，SDK 不声称能检测任意异步等待图。 SDK 的 disable/stop 是必须完成的清理操作：取消请求会等清理结束后再传播。 这比低层 Kernel 可取消排空并恢复运行的操作更保守；高级宿主仍可使用低层接口。 snapshot() 返回 Kernel 快照；capabilities()、traces() 返回脱离内部对象的 诊断数据。原始 Plugin 可通过 runtime.register(plugin, enabled=False) 接入， 因此 Agent 不需要获得特殊内核权限。

测试不必再次手工装配 Kernel：

from caploom.sdk.testing import RuntimeTest

test = RuntimeTest([provider], bindings={contract.id: provider.name})
async with test as runtime:
    client = runtime.client(caller="test", allowed=[contract.id])
    result = await client.invoke(contract, request)
test.assert_clean()

测试工具检查受 Caploom 管理的 Fiber、Effect、Service、Provider 和 Binding， 不声称能检测业务代码私自创建的所有线程、文件或连接。资源自身关闭状态仍应断言。

7. 兼容性与分发

SDK 随同一个 caploom distribution 分发，使用 capabilities extra，未新增 Web 或 Agent 的强制依赖。基础 import caploom 仍只需要标准库。 当前项目尚未承诺公共 PyPI 包名或稳定大版本；使用本地构建 Wheel 验证：

uv build
# 在另一个虚拟环境中，安装本地构建产物及 capabilities extra。
uv pip install './dist/caploom-0.1.0-py3-none-any.whl[capabilities]'

SDK API 当前是 0.1 阶段，使用方应固定测试过的版本或提交。将来不兼容 API 更改需要迁移说明和回归验证。Capability 的 @major 独立于包版本；破坏业务 契约不能只换 Provider 名称而沿用旧主版本。

Code 负责实现、Manifest 负责声明、Profile 负责组合的原设计仍保留。SDK 不替代 现有 Profile，也不把运行时 enable/disable 描述成 pip 安装。可选的 caploom.plugins 现在支持已安装 wheel 的 Entry Point 和 plugin.toml 检查；caploom.adapters 提供 显式 HTTP Provider 与 MCP 出口，详见 互通接入说明。这些都在 SDK 外围。 TypeScript Client、MCP 输入适配、模板 CLI 和外部开发者接入时间实验仍需继续验证， 不能把当前 SDK 与协议适配切片标记为整个 Phase 2 已完成。