Unity カタログ関数を使用してエージェントツールを作成する

エージェントが、エージェントがパラメーターを指定する SQL クエリなど、あらかじめ定義されたロジックを実行する必要がある場合は、Unity Catalog の機能を使用してください。他のツールについては、まずMCPから始めましょう。

このガイドではUnity Catalog AIを使って小さなPython関数を作成し、テストし、選択したフレームワークを使ってエージェントに追加します。すでに関数がある場合は、ステップ1でライブラリをインストールしてステップ3に進みましょう。

Requirements

Python 3.10 以上を使用する Azure Databricks ノートブック。クラシックコンピュートでは、Databricks Runtime 15.0以上を使いましょう。
サーバーレスコンピュートがワークスペースで有効になっています。 Function Client は、ツールを実行するためにサーバーレスの汎用コンピューティング機能を利用しており、SQL ウェアハウスだけでは不十分です。
関数の親カタログとスキーマに関する USE CATALOG と USE SCHEMA。関数を作成するにはスキーマ上の CREATE FUNCTION が必要です。既存の関数を呼び出すには、その関数に対する EXECUTE が必要です。関数の権限を参照してください。
ツールコールをサポートするモデル。 LangGraphの例ではAzure Databricksモデルエンドポイントを使用しています。 OpenAI、Anthropic、LlamaIndexの例は、タブで説明されているようにプロバイダーAPI認証情報を使用しています。

ステップ1:ライブラリのインストール

ノートブックのセルでフレームワークのインストールコマンドを実行します:

フレームワーク	インストールコマンド
LangGraph	`%pip install --upgrade unitycatalog-langchain[databricks] databricks-langchain langgraph`
OpenAI	`%pip install --upgrade unitycatalog-openai[databricks]`
Anthropic	`%pip install --upgrade unitycatalog-anthropic[databricks]`
LlamaIndex	`%pip install --upgrade unitycatalog-llamaindex[databricks] llama-index-llms-openai`

パッケージのソースコードについては、 Unity CatalogのAIリポジトリをご覧ください。

その後、別のセルでPythonを再起動します。

dbutils.library.restartPython()

ステップ2:関数の作成とテスト

DatabricksFunctionClientを使って関数を作成・テストしてください。 my_catalogやmy_schemaを、関数を作成できる既存のカタログとスキーマに置き換えます。この例では add_numbers を作成します。すでに同じ名前が存在する場合は、別の名前を選択してください。

from unitycatalog.ai.core.databricks import DatabricksFunctionClient

client = DatabricksFunctionClient()
CATALOG = "my_catalog"
SCHEMA = "my_schema"


def add_numbers(number_1: float, number_2: float) -> float:
    """Add two numbers.

    Args:
        number_1: The first number to add.
        number_2: The second number to add.

    Returns:
        The sum of the two numbers.
    """
    return number_1 + number_2


client.create_python_function(func=add_numbers, catalog=CATALOG, schema=SCHEMA)

result = client.execute_function(
    function_name=f"{CATALOG}.{SCHEMA}.add_numbers",
    parameters={"number_1": 2.0, "number_2": 3.0},
)
if result.error:
    raise RuntimeError(result.error)
print(result.value)  # 5.0

ステップ3: エージェント内で関数を使う

function_name を関数の完全な名前に置き換え、ご利用のフレームワーク用のサンプルを実行してください。それぞれ UCFunctionToolkit を使って関数を直接呼び出します。

from unitycatalog.ai.core.databricks import DatabricksFunctionClient

function_name = "my_catalog.my_schema.add_numbers"
client = DatabricksFunctionClient()
question = "Use the tool to add 2.0 and 3.0."

LangGraph

この例では databricks-claude-sonnet-4-5 エンドポイントを用いています。ワークスペースで利用可能なツール呼び出しモデルに置き換えてください。

from databricks_langchain import ChatDatabricks, UCFunctionToolkit
from langgraph.prebuilt import create_react_agent

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = create_react_agent(
    ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
    tools=toolkit.tools,
)
result = agent.invoke({"messages": [{"role": "user", "content": question}]})
print(result["messages"][-1].content)

LangChain統合リファレンスを参照してください。

OpenAI

ノートブックのプロセスで OPENAI_API_KEY と OPENAI_MODEL 環境変数を設定してください。ツール呼び出しをサポートするモデルを使用し、APIキーをAzure Databricks シークレットに保存してください。

import os
from openai import OpenAI
from unitycatalog.ai.openai.toolkit import UCFunctionToolkit
from unitycatalog.ai.openai.utils import generate_tool_call_messages

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = OpenAI()
messages = [{"role": "user", "content": question}]

for _ in range(10):
    response = model_client.chat.completions.create(
        model=os.environ["OPENAI_MODEL"],
        messages=messages,
        tools=toolkit.tools,
    )
    if not response.choices[0].message.tool_calls:
        print(response.choices[0].message.content)
        break
    messages.extend(generate_tool_call_messages(response=response, client=client))
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

ヘルパーは要求された関数を実行し、次のモデル呼び出しのために結果をフォーマットします。 OpenAI 統合リファレンスを参照してください。

Anthropic

ノートブックプロセス用に、ANTHROPIC_API_KEY と ANTHROPIC_MODEL の環境変数を設定してください。ツール呼び出しをサポートするモデルを使用し、API キーをAzure Databricks シークレットに保存してください。

import os
from anthropic import Anthropic
from unitycatalog.ai.anthropic.toolkit import UCFunctionToolkit
from unitycatalog.ai.anthropic.utils import generate_tool_call_messages

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = Anthropic()
messages = [{"role": "user", "content": question}]

for _ in range(10):
    response = model_client.messages.create(
        model=os.environ["ANTHROPIC_MODEL"],
        max_tokens=1024,
        tools=toolkit.tools,
        messages=messages,
    )
    if response.stop_reason != "tool_use":
        print("\n".join(block.text for block in response.content if block.type == "text"))
        break
    messages.extend(
        generate_tool_call_messages(
            response=response, client=client, conversation_history=[],
        )
    )
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

ヘルパーは要求された関数を実行します。次のモデルコールの前にそのメッセージ群を会話に追加してください。 Anthropic integration リファレンスを参照してください。

LlamaIndex

この例はOpenAIモデルを使用しています。ノートブックのプロセスに OPENAI_API_KEY と OPENAI_MODEL の環境変数を設定してください。ツール呼び出しをサポートするモデルを使用し、API キーをAzure Databricks シークレットに保存してください。

import os
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
from unitycatalog.ai.llama_index.toolkit import UCFunctionToolkit

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = FunctionAgent(
    tools=toolkit.tools,
    llm=OpenAI(model=os.environ["OPENAI_MODEL"]),
)
result = await agent.run(user_msg=question)
print(result)

await対応のノートセルで最後の2行を実行してください。 LlamaIndex統合リファレンスを参照してください。

エージェントは add_numbers を呼び出し、5 を含む応答を返します。既存の関数を使用する場合は、question をその関数がサポートするタスクに変更してください。

展開と認証については、「カスタムエージェントの構築」および「エージェント認証」を参照してください。

ローカル開発

信頼されたPython関数をテストするには、クライアントをexecution_mode="local"で初期化し、フレームワークのツールキットに渡します。

client = DatabricksFunctionClient(execution_mode="local")
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)

ローカルモードはUnity Catalogから関数を取得し、ローカルサブプロセスで実行します。 SQL関数はサポートしていません。本番環境ではデフォルトのサーバーレスモードを使用し、信頼されたコードのみを実行してください。実行モードを参照してください。

タイムアウト、メモリ制限、サーバーレス設定については、ローカル実行制限および Function Client環境変数を参照してください。

workspace MCPエンドポイント (レガシー) を使用

既存のMCP統合の場合、このLangGraphの例はレガシーワークスペースエンドポイントを通じて同じ関数に接続されています。 Managed MCP Servers のプレビューを有効にし、その関数へのアクセス権を持つ ID を使用します。

ノートブックのセルで %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" を実行し、Python を再起動します。 my_catalog/my_schema/add_numbersを自分の関数のパスに置き換えてください。

from databricks.sdk import WorkspaceClient
from databricks_langchain import (
    ChatDatabricks,
    DatabricksMCPServer,
    DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
function_path = "my_catalog/my_schema/add_numbers"
mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="uc-functions",
        url=f"{workspace_client.config.host}/api/2.0/mcp/functions/{function_path}",
        workspace_client=workspace_client,
    ),
])
agent = create_react_agent(
    ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
    tools=await mcp_client.get_tools(),
    prompt=lambda state: convert_to_openai_messages(state["messages"]),
)
result = await agent.ainvoke({
    "messages": [{"role": "user", "content": "Use the tool to add 2.0 and 3.0."}],
})
print(result["messages"][-1].content)

デプロイメントについては、エージェント認証を参照してください。 http_requestで外部APIを呼び出す関数については、Legacy REST APIツールを参照してください。

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-08