Python SDK

用 pip 安装官方 Python SDK，从 Busabase Cloud 或私有化部署的服务器读取你的第一批 Bases 和记录。

busabase 是 Busabase REST API 的官方 Python 客户端，适合在脚本、Notebook、数据管道和 Python 后端里使用。同一个客户端既能连 Busabase Cloud，也能连任意私有化部署的 Busabase 服务器；它和 CLI、TypeScript SDK 使用同一套服务地址、API 密钥和工作区设置。

软件包：PyPI 上的 busabase
源码与问题反馈：github.com/busabase/busabase-python-sdk（MIT 协议）
需要 Python 3.11 或更高版本，同步、异步调用都支持。

每个接口都有类型化的模型，由 Busabase 的 OpenAPI 文档生成。在此之上有一层很薄的手写封装：从环境变量读取配置、提供 bb.<分组>.<操作>(...) 形式的调用，并在请求失败时统一抛出 BusabaseError。

安装

pip install busabase

uv 项目也可以使用 uv add busabase。

快速开始

from busabase import Busabase

bb = Busabase(api_key="bb_...")  # or set BUSABASE_API_KEY
for base in bb.bases.list():
    print(base.slug, base.name)
record = bb.records.get(record_id="rec_...")
print(record.head_commit.payload.to_dict())  # the record's field values

先在 API 密钥页面创建一个密钥（这个示例用只读密钥就够了），再把 rec_... 换成一条你能看到的记录 id。运行后会逐行打印密钥可见的每个 Base 的 slug 和名称，然后打印这条记录的字段值。此示例只读取数据。

Busabase Cloud 还是私有化部署

	Busabase Cloud	私有化部署的服务器
`base_url` / `BUSABASE_BASE_URL`	`https://busabase.com`（默认值）	你的服务器地址，例如 `http://localhost:15419`
`api_key` / `BUSABASE_API_KEY`	必填，在 API 密钥中创建	本地服务器不需要
`space` / `BUSABASE_SPACE_ID`	密钥能访问多个工作区时需要填写	通常不需要

bb = Busabase()  # everything from the environment
bb = Busabase(base_url="http://localhost:15419")  # a local `busabase server`
bb = Busabase(api_key="bb_...", space="spc_...")  # a specific Cloud workspace

base_url 末尾带不带 /api/v1 都可以。

SDK 会读取环境变量，但不会自动加载 ~/.busabase/.env。如果想沿用通过 busabase-cli login 和 busabase-cli space use 保存的连接配置，请在运行 Python 之前导出这个文件：

set -a; . "$HOME/.busabase/.env"; set +a
python quickstart.py

仅 Cloud 提供的接口。 SDK 覆盖全部公开 API，其中少数几组只在 Busabase Cloud 上存在，连私有化部署的服务器会返回 404：system（health、meta）、users.me 和 agent_tasks。

写入：Change Requests

Busabase 中的每次写入都是一个 Change Request。密钥有写入权限时，它会在同一次调用里直接合并，并返回保存后的对象；没有写入权限时，它会进入收件箱，等待有人审核。可以这样判断拿到的是哪一种：

from busabase.generated.models import MaterializedRecord, PendingChangeRequest

result = bb.bases.create_change_request(
    base_id="bas_...",
    body={"fields": {"company": "Acme", "score": 7}, "message": "Add Acme — inbound lead"},
)
if isinstance(result, MaterializedRecord):
    print("merged:", result.id)
elif isinstance(result, PendingChangeRequest):
    print("waiting for review:", result.id)

不要传 autoMerge，让服务端根据密钥的权限自行决定。

调用接口

操作按 API 的结构分组（bases、records、nodes、docs、file_trees、change_requests、search、grep 等），名称使用 snake_case。用 dir(bb) 和 dir(bb.records) 可以列出可用的操作。

bb.records.list(base_id="bas_...", limit=20)
bb.nodes.get("nod_...")  # path parameters can be positional
bb.grep(body={"pattern": "acme", "sources": ["records", "nodes"]})
await bb.records.get.asyncio(record_id="rec_...")  # async

请求的 body 可以是普通的 dict，也可以是生成的模型。请求和响应结构见 API 参考。

错误处理

只要 HTTP 状态码大于等于 400，就会抛出 BusabaseError，其中带有服务端返回的错误信息：

from busabase import BusabaseError

try:
    bb.records.get(record_id="rec_missing")
except BusabaseError as error:
    print(error.status, error.code, error.message)  # 404 NOT_FOUND Record not found: rec_missing

code 表示服务端给出的结果类型（NOT_FOUND、FORBIDDEN、BAD_REQUEST、CONFLICT 等）。校验失败时，message 会逐个列出出错的字段，data 中保存原始的 issues。

API 密钥请保存在服务端环境中，不要写进要分享或提交到仓库的 Notebook。异步用法、超时、代理等更多配置见 SDK README。

其他接入方式：接入方式总览 · CLI · SDK · Python SDK · MCP · OpenAPI / REST · Agent Skills · Webhook