Claude Agent SDKとDatabricks Foundation Model APIsで作るカスタムエージェント

こんにちは、VポイントマーケティングAIエンジニアの三浦諒一です。

あっという間に桜の時期が過ぎて、いつの間にか気温が高くなってきました。外で体を動かすときなんかはもうTシャツだけで十分なくらいです。

アプリケーション開発にClaude CodeやGithub Copilot、Codexなどを活用しているのですが、それ以外にもドキュメントを書いてもらったり調査をしたりといった用途でもこれらのエージェントがとても便利だと感じるようになりました。

ただこういったエージェントは便利な反面扱う側も気を使わなければいけないことがあります。使うツールに制限をかけるとか、動作するための専用環境を用意するなどです。こういったハードルを軽減しながらこれらのエージェントのメリットを誰もが受けられるようにしたいな、と考えていました。

そんな中、Claude Agent SDKというライブラリがあることを知り、これを使えば自分が実現したいと思っていたことが実現出来そうな可能性を感じました。

今回の記事はClaude Agent SDKの使い方を中心にまとめてみました。バックエンドのLLMをDatabricksでホストされたものを利用することで今後社内展開がしやすくなるように意識した内容になっています。

Claude Agent SDKとは

Claude Agent SDKとはAnthropicが公開しているカスタムエージェントを開発するためのSDKです。

code.claude.com

Agent SDKを使うと、Claude Codeでも採用されている以下の特徴を持ったエージェントを開発することが出来るようになります。

• Claude Codeと同様のToolが使用出来る
• Toolの実行結果と評価、再実行を繰り返すAgentic Loop
• Contextの管理

たとえばAgent SDKでは以下のToolをすぐに利用することが出来ます。

Tool	説明
Read	作業ディレクトリ内の任意のファイルを読み取る
Write	新しいファイルを作成する
Edit	既存のファイルに対してピンポイントな編集を行う
Bash	ターミナルコマンド、スクリプト、git操作を実行する
Monitor	バックグラウンドで実行中のスクリプトを監視し、各出力行をイベントとして処理する
Glob	パターン指定（例: **/*.ts, src/**/*.py）でファイルを検索する
Grep	正規表現を使ってファイル内容を検索する
WebSearch	最新の情報を取得するために Web検索を行う
WebFetch	Webページの内容を取得し、解析する
AskUserQuestion	ユーザーに対して、選択肢付きの確認質問を行う

さらにClaude Codeと同様にAgent Skillsも利用することが出来ます。Agent Skillsを使うと定型的なワークフローをエージェントに実行させることが出来るため、エージェント自体の構造はシンプルに保ちながらAgent Skillsによってエージェントの能力を拡張していく、といったことが可能になります。

Databricks Foundation Model APIs

Databricks Foundation Model APIsはDatabricksでホストされた基盤モデルをAPIで利用できるサービスです。Claude Sonnet 4.6やClaude Opus 4.6といったAnthropicのモデルも利用することが出来ます。

Databricks Foundation Model APIsを利用することのメリットとしてはDatabricksのユーザーごとに使用状況やリミットを掛けられるAI Gatewayと組み合わせられる点が挙げられます。またDatabricksの標準OAuth認証を利用してアクセスすることが出来るため、アクセストークンをアプリケーションで管理する必要が無い点もメリットです。

実はClaude Codeは内部で呼び出す基盤モデルを変更することが出来るのですが、Databricks Foundation Model APIsで提供されているモデルを利用することも出来ます。そして今回取り上げるAgent SDKでも同様にDatabricksのFoundation Model APIsを使うことが出来ます。

実装方法

環境

Agent SDKはPythonとTypeScriptで利用出来、今回はPythonを採用しました。Pythonのプロジェクトの環境はuvを使って行いました。

pyproject.tomlは以下のような内容で作成しました。

[project]
name = "claude-agent-sdk-playground"
version = "0.1.0"
description = "Playground for the Claude Agent SDK"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
  "databricks-sdk==0.102.0",
  "claude-agent-sdk==0.1.56",
  "python-dotenv==1.2.2",
]
...

環境変数は以下をセットします。ANTHROPIC_AUTH_TOKENとANTHROPIC_BASE_URLはコードの中でセットするようにしました。

環境変数	説明
ANTHROPIC_BASE_URL	https://{databricks-host-name}/serving-endpoints/anthropic
ANTHROPIC_AUTH_TOKEN	Databricks Foundation Model APIsを使うためのToken
CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS	1に設定することで、Anthropic固有のヘッダーとベータフィールドを削除し、Claude Codeで外部LLMが利用可能になります
DATABRICKS_CONFIG_PROFILE	接続に使用するDatabricksのProfile
ANTHROPIC_MODEL	基盤モデル名(databricks-claude-sonnet-4-5 etc)

ANTHROPIC_AUTH_TOKENとANTHROPIC_BASE_URLの取得処理

基盤モデルへの接続に必要となるANTHROPIC_AUTH_TOKENとANTHROPIC_BASE_URLは以下の処理で取得するようにしました。DATABRICKS_CONFIG_PROFILEには事前に作成済みのDatabricks構成プロファイル名を指定します。

from databricks.sdk import WorkspaceClient
from dotenv import load_dotenv

load_dotenv("../.env")

def get_workspace_client() -> WorkspaceClient:
    profile = os.getenv("DATABRICKS_CONFIG_PROFILE")

    if profile:
        return WorkspaceClient(profile=profile)

    return WorkspaceClient()

def set_anthropic_auth_token_from_databricks() -> None:
    w = get_workspace_client()

    token = w.api_client._cfg.oauth_token()
    host = w.config.host
    if not token or not getattr(token, "access_token", None):
        raise RuntimeError("Failed to retrieve Databricks OAuth access token")

    os.environ["ANTHROPIC_AUTH_TOKEN"] = token.access_token
    os.environ["ANTHROPIC_BASE_URL"]=f"{host}/serving-endpoints/anthropic"

1回だけクエリを実行する

1回だけクエリを実行するシンプルな用途であれば以下のような実装になります。

import asyncio
import argparse
import os

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage

async def main():
    set_anthropic_auth_token_from_databricks()
    parser = argparse.ArgumentParser(
        description="A simple sample for using the Claude Agent SDK to query a Databricks-hosted Claude model."
    )

    parser.add_argument(
        "--prompt",
        type=str,
        help="The prompt to send to the Claude model.",
    )

    args = parser.parse_args()
    async for message in query(
        prompt=args.prompt,
        options=ClaudeAgentOptions(
            setting_sources=["project"],
            allowed_tools=["Read", "Edit", "Bash", "Write"],
        ),   
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(f"block_test:{block.text}"))
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print("Final result:", message.result)

asyncio.run(main())

次のように実行することが出来ます。

uv run python main.py --profile "このプロジェクトのREADME.md書いて。"

print(f"block_test:{block.text}")によって途中経過を以下のように表示することが出来ます。UIを実装する時はこのメッセージを表示させることで作業の経過状況をユーザーに知らせることが出来そうです。

lock_test:README.mdを作成する前に、プロジェクトの内容を確認させてください。
block_test:プロジェクトの内容を確認しました。これはDatabricks上でホストされているClaude AIモデルをClaude Agent SDK経由で呼び出すPythonプロジェクトですね。README.mdを作成します。
block_test:README.mdを作成しました！
...

このコードのポイントの1つが以下の部分です。

async for message in query(
    prompt=args.prompt,
    options=ClaudeAgentOptions(
        setting_sources=["project"],
        allowed_tools=["Read", "Edit", "Bash", "Write"],
    ),   
):
    ...

setting_sourcesはエージェントが実行時に参照する設定ファイルの読み込み元を指定しています。"user", "project", "local"を指定することが出来て、この設定に基づいてCLAUDE.mdやAgent Skillsのファイルをエージェントに参照させることが可能です。

allowed_toolsは許可済みのToolを指定します。ここに指定したToolはユーザーの承認なしに実行されます。ここに指定されていないToolは実行前にユーザーの承認が必要になります。

セッション内で複数のやり取りを行う場合

先ほどの例は1回限りのクエリを実行する場合を取り扱いました。定期実行する処理などはこの書き方が向いていますが、チャットアプリなどユーザーと同一セッション内で複数のやり取りをしたい場合は別の方法で実現出来ます。

ClaudeSDKClientを使うと内部でセッションIDを生成してセッション内のやり取りをコンテキストに保持しながら入力を処理できるようになります。実装例は以下のようになります。

async def print_response(client: ClaudeSDKClient) -> None:
    """メッセージ表示用の関数(複数個所で呼び出すので関数化)"""
    async for message in client.receive_response():
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text") and block.text:
                    print(block.text, end="", flush=True)

        if isinstance(message, ResultMessage):
            print()
            print(f"\nFinal result: {message.result}")


async def main() -> None:
    set_anthropic_auth_token_from_databricks()
    args = parse_args()

    options = ClaudeAgentOptions(
        setting_sources=["project"],
        allowed_tools=["Read", "Edit", "Bash","Write"],
    )

    async with ClaudeSDKClient(options=options) as client:
        print("Interactive session started. Type 'exit' or 'quit' to end.")

        if args.prompt:
            await client.query(args.prompt)
            await print_response(client)

        while True:
            """exit or endが入力されたら終了する"""
            user_input = input("\nYou> ").strip()
            if user_input.lower() in {"exit", "quit"}:
                break
            if not user_input:
                continue

            await client.query(user_input)
            await print_response(client)

Agent Skillsを使う

最後にAgent Skillsを使う場合を取り扱います。Agent Skillsはもしプロジェクト内に配置するのであれば.claude/skillsに格納しておきます。そのうえでClaudeAgentOptionsを以下のように設定します。

options = ClaudeAgentOptions(
    setting_sources=["project"],
    allowed_tools=["Skill", "Read", "Edit", "Bash","Write"],
)

エージェントが自動的にプロジェクト内のSkillを利用して作業をしてくれるようになります。

まとめ

ということで今回はClaude Agent SDKを使ってClaude Codeで使われている技術を活用したカスタムエージェントを作る方法についてまとめてみました。また、基盤モデルとしてDatabricks Foundation Model APIsを利用する方法も調べることが出来ました。今後はDatabricks Appsでカスタムエージェントを稼働し、社内展開できるか調べてみたいと思っています。