Python SDK

公式 Python SDK を pip でインストールし、Busabase Cloud またはセルフホストのサーバーから最初の Bases とレコードを読み取ります。

busabase は Busabase REST API の公式 Python クライアントです。スクリプト、ノートブック、データパイプライン、Python のバックエンドから利用できます。1 つのクライアントで Busabase Cloud にもセルフホストの Busabase サーバーにも接続でき、CLI や TypeScript SDK と同じ URL、API キー、ワークスペースの設定を使います。

パッケージ：PyPI の busabase
ソースコードと Issue：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 キーはサーバーの環境変数に保存し、共有したりリポジトリにコミットしたりするノートブックには書かないでください。非同期の使い方、タイムアウト、プロキシなどの設定は SDK の README を参照してください。

その他の接続方法：連携方法の一覧 · CLI · SDK · Python SDK · MCP · OpenAPI / REST · Agent Skills · Webhook