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

Unity カタログ関数を使用して、カスタム ロジックを実行し、言語生成を超えて LLM の機能を拡張する特定のタスクを実行する AI エージェント ツールを作成します。

Unity カタログ関数と MCP サーバーを使用する場合

Databricks では、クエリが事前にわかっており、エージェントがパラメーターを提供する場合に、構造化データ取得ツール専用のエージェント ツールとして Unity カタログ関数を使用することをお勧めします。 「エージェントを構造化データに接続する」を参照してください。

他のほとんどのユース ケースでは、Databricks は MCP サーバーを推奨するか、エージェント コードでロジックを直接定義して、実行を高速化し、ユーザーごとの認証をサポートし、柔軟性を高めます。

要求事項

UNITY カタログ関数を AI エージェント ツールとして作成して使用するには、次のものが必要です。

• Databricks ランタイム: Databricks Runtime 15.0 以降を使用する
• Python バージョン: Python 3.10 以降をインストールします

Unity カタログ関数を実行するには:

• 運用環境で AI エージェント ツールとして Unity カタログ機能を実行するには、ワークスペースでサーバーレス コンピューティングを有効にする必要があります。 サーバーレス コンピューティング要件を参照してください。
  • Python関数のローカル モードの実行では、サーバーレス汎用コンピューティングを実行する必要はありません。ただし、ローカル モードは開発とテストのみを目的としています。

Unity カタログ関数を作成するには:

• Databricks ワークスペース クライアントまたは SQL 本文ステートメントを使用して関数を作成するには、ワークスペースでサーバーレス 汎用コンピューティングを有効にする必要があります。
  • Python関数は、サーバーレス コンピューティングなしで作成できます。

Unity カタログ関数ツールを作成する

次の手順では、Unity カタログ関数を作成してテストする方法を示します。 Databricks ノートブックで次のコードを実行します。

依存関係のインストール

[databricks]を追加して Unity カタログ AI パッケージをインストールします。

# Install Unity Catalog AI integration packages with the Databricks extra
%pip install unitycatalog-ai[databricks]

dbutils.library.restartPython()

Databricks 関数クライアントを初期化する

Databricks 関数クライアントを初期化します。これは、Databricks で Unity カタログ関数を作成、管理、および実行するための特殊なインターフェイスです。

from unitycatalog.ai.core.databricks import DatabricksFunctionClient

client = DatabricksFunctionClient()

ツールのロジックを定義する

Unity カタログ ツールは、実際には Unity カタログのユーザー定義関数 (UDF) です。 Unity カタログ ツールを定義すると、Unity カタログに関数が登録されます。 Unity カタログ UDF の詳細については、 Unity カタログのユーザー定義関数 (UDF) に関するページを参照してください。

Warnung

エージェント ツールで任意のコードを実行すると、エージェントがアクセスできる機密情報や個人情報が公開される可能性があります。 お客様は、信頼できるコードのみを実行し、ガードレールと適切なアクセス許可を構成して、意図しないデータへのアクセスを防ぐ責任を負います。

Unity カタログ関数は、次の 2 つの API のいずれかを使用して作成できます。

• create_python_function は、呼び出し可能なPythonを受け入れます。
• create_function は、SQL 本文の作成関数ステートメントを受け入れます。 Python関数の作成を参照してください。

create_python_function API を使用して関数を作成します。

Unity カタログ関数データ モデルに対して呼び出し可能なPythonを認識できるようにするには、関数が次の要件を満たしている必要があります。

• Type ヒント: 関数シグネチャは、有効なPython型ヒントを定義する必要があります。 名前付き引数と戻り値の両方に、型が定義されている必要があります。
• 変数引数を使用しないでください。*args や **kwargs などの変数引数はサポートされていません。 すべての引数を明示的に定義する必要があります。
• Type 互換性: SQL では、すべてのPython型がサポートされているわけではありません。 Spark でサポートされるデータ型を参照してください。
• 説明ドキュメント文字列: Unity カタログ関数ツールキットは、docstring から重要な情報を読み取り、解析、抽出します。
  • Docstring は 、Google docstring 構文に従って書式設定する必要があります。
  • LLM が関数を使用する方法とタイミングを理解するのに役立つ、関数とその引数の明確な説明を記述します。
• 依存関係のインポート: ライブラリは関数の本体内にインポートする必要があります。 関数外のインポートは、ツールの実行時に解決されません。

次のコード スニペットでは、create_python_function を使用して、Python呼び出し可能な add_numbersを登録します。

CATALOG = "my_catalog"
SCHEMA = "my_schema"

def add_numbers(number_1: float, number_2: float) -> float:
  """
  A function that accepts two floating point numbers adds them,
  and returns the resulting sum as a float.

  Args:
    number_1 (float): The first of the two numbers to add.
    number_2 (float): The second of the two numbers to add.

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

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

関数をテストする

関数をテストして、期待どおりに動作することを確認します。 execute_function API で完全修飾関数名を指定して、関数を実行します。

result = client.execute_function(
  function_name=f"{CATALOG}.{SCHEMA}.add_numbers",
  parameters={"number_1": 36939.0, "number_2": 8922.4}
)

result.value # OUTPUT: '45861.4'

Unity カタログ関数をエージェントに追加する

Unity カタログ関数を作成してテストしたら、次のいずれかの方法を選択してエージェントに追加します。

MCP の使用 (推奨)

MCP の使用 (推奨)

Databricks では、MCP サーバーを使用して Unity カタログ関数をエージェントに追加することをお勧めします。 MCP アプローチを使用すると、ツールの自動検出と組み込みの認証サポートとの統合が簡単になります。

Unity カタログ関数のマネージド MCP URL は、 https://<workspace-hostname>/api/2.0/mcp/functions/{catalog}/{schema}です。 必要に応じて、 /{function_name}を追加することで、特定の関数を指定できます。

次の例では、MCP を使用してエージェントを Unity カタログ関数に接続する方法を示します。 <catalog>と<schema>を関数の場所に置き換えます。

OpenAI Agents SDK (アプリ)

from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer

workspace_client = WorkspaceClient()

async with McpServer.from_uc_function(
    catalog="<catalog>",
    schema="<schema>",
    workspace_client=workspace_client,
    name="uc-functions",
) as uc_server:
    agent = Agent(
        name="Tool-using agent",
        instructions="You are a helpful assistant. Use the available tools to answer questions.",
        model="databricks-claude-sonnet-4-5",
        mcp_servers=[uc_server],
    )
    result = await Runner.run(agent, "Look up customer info for Acme Corp")
    print(result.final_output)

databricks.ymlの Unity Catalog 関数へのアクセス権をアプリに付与します。

resources:
  apps:
    my_agent_app:
      resources:
        - name: 'my_uc_function'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<function-name>'
            securable_type: 'FUNCTION'
            permission: 'EXECUTE'

LangGraph (アプリ)

from databricks.sdk import WorkspaceClient
from databricks_langchain import ChatDatabricks, DatabricksMCPServer, DatabricksMultiServerMCPClient
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
host = workspace_client.config.host

mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="uc-functions",
        url=f"{host}/api/2.0/mcp/functions/<catalog>/<schema>",
        workspace_client=workspace_client,
    ),
])

async with mcp_client:
    tools = await mcp_client.get_tools()
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=tools,
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Look up customer info for Acme Corp"}]}
    )
    print(result["messages"][-1].content)

databricks.ymlの Unity Catalog 関数へのアクセス権をアプリに付与します。

resources:
  apps:
    my_agent_app:
      resources:
        - name: 'my_uc_function'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<function-name>'
            securable_type: 'FUNCTION'
            permission: 'EXECUTE'

Model Serving

from databricks.sdk import WorkspaceClient
from databricks_mcp import DatabricksMCPClient
import mlflow

workspace_client = WorkspaceClient()
host = workspace_client.config.host

# Connect to the UC functions MCP server
mcp_client = DatabricksMCPClient(
    server_url=f"{host}/api/2.0/mcp/functions/<catalog>/<schema>",
    workspace_client=workspace_client,
)

# List available tools
tools = mcp_client.list_tools()

# Log the agent with the required resources for deployment
mlflow.pyfunc.log_model(
    "agent",
    python_model=my_agent,
    resources=mcp_client.get_databricks_resources(),
)

エージェントをデプロイするには、「 生成 AI アプリケーション用のエージェントをデプロイする (モデル サービス)」を参照してください。 MCP リソースを使用したエージェントのログ記録の詳細については、 Databricks マネージド MCP サーバーの使用に関するページを参照してください。

UCFunctionToolkit の使用

UCFunctionToolkit の使用

この例では LangChain を使用していますが、他のライブラリにも同様の方法を適用できます。 Unity カタログ ツールとサード パーティの生成 AI フレームワークの統合に関する説明を参照してください。

追加の依存関係をインストールする

UCFunctionToolkit 用の LangChain 統合パッケージをインストールします。

%pip install unitycatalog-langchain[databricks]

# Install the Databricks LangChain integration package
%pip install databricks-langchain

dbutils.library.restartPython()

UCFunctionToolKit を使用して関数をラップする

UCFunctionToolkitを使用して関数をラップし、エージェント作成ライブラリからアクセスできるようにします。 このツールキットは、さまざまな Gen AI ライブラリ間の一貫性を確保し、レトリバーの自動トレースなどの便利な機能を追加します。

from databricks_langchain import UCFunctionToolkit

# Create a toolkit with the Unity Catalog function
func_name = f"{CATALOG}.{SCHEMA}.add_numbers"
toolkit = UCFunctionToolkit(function_names=[func_name])

tools = toolkit.tools

エージェントでツールを使用する

toolsの UCFunctionToolkit プロパティを使用して、ツールを LangChain エージェントに追加します。

注

この例では、LangChain を使用します。 ただし、Unity カタログ ツールは、LlamaIndex、OpenAI、Anthropic などの他のフレームワークと統合できます。 Unity カタログ ツールとサード パーティの生成 AI フレームワークの統合に関する説明を参照してください。

この例では、簡単にするために LangChain AgentExecutor API を使用して単純なエージェントを作成します。 運用ワークロードの場合は、「AI エージェントの作成」に示されているエージェント作成ワークフロー を使用し、Databricks Apps にデプロイします。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.prompts import ChatPromptTemplate
from databricks_langchain import (
  ChatDatabricks,
  UCFunctionToolkit,
)
import mlflow

# Initialize the LLM (optional: replace with your LLM of choice)
LLM_ENDPOINT_NAME = "databricks-meta-llama-3-3-70b-instruct"
llm = ChatDatabricks(endpoint=LLM_ENDPOINT_NAME, temperature=0.1)

# Define the prompt
prompt = ChatPromptTemplate.from_messages(
  [
    (
      "system",
      "You are a helpful assistant. Make sure to use tools for additional functionality.",
    ),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
  ]
)

# Enable automatic tracing
mlflow.langchain.autolog()

# Define the agent, specifying the tools from the toolkit above
agent = create_tool_calling_agent(llm, tools, prompt)

# Create the agent executor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor.invoke({"input": "What is 36939.0 + 8922.4?"})

明確なドキュメントを使用してツール呼び出しを改善する

適切なドキュメントは、エージェントが各ツールを使用するタイミングと方法を把握するのに役立ちます。 ツールを文書化するには、次のベスト プラクティスに従います。

• Unity カタログ関数の場合は、 COMMENT 句を使用してツールの機能とパラメーターを記述します。
• 予想される入力と出力を明確に定義します。
• エージェントや人間が使いやすいツールを作成するためのわかりやすい説明を記述します。

例: 有効なツールのドキュメント

次の例は、構造化テーブルに対してクエリを実行するツールの明確な COMMENT 文字列を示しています。

CREATE OR REPLACE FUNCTION main.default.lookup_customer_info(
  customer_name STRING COMMENT 'Name of the customer whose info to look up.'
)
RETURNS STRING
COMMENT 'Returns metadata about a specific customer including their email and ID.'
RETURN SELECT CONCAT(
    'Customer ID: ', customer_id, ', ',
    'Customer Email: ', customer_email
  )
  FROM main.default.customer_data
  WHERE customer_name = customer_name
  LIMIT 1;

例: 無効なツールのドキュメント

次の例では、重要な詳細が不足しているため、エージェントがツールを効果的に使用することが困難になります。

CREATE OR REPLACE FUNCTION main.default.lookup_customer_info(
  customer_name STRING COMMENT 'Name of the customer.'
)
RETURNS STRING
COMMENT 'Returns info about a customer.'
RETURN SELECT CONCAT(
    'Customer ID: ', customer_id, ', ',
    'Customer Email: ', customer_email
  )
  FROM main.default.customer_data
  WHERE customer_name = customer_name
  LIMIT 1;

サーバーレス モードまたはローカル モードを使用して関数を実行する

Gen AI サービスがツール呼び出しが必要であると判断すると、統合パッケージ (UCFunctionToolkit インスタンス) によって DatabricksFunctionClient.execute_function API が実行されます。

execute_function呼び出しでは、サーバーレスまたはローカルの 2 つの実行モードで関数を実行できます。 このモードでは、関数を実行するリソースが決まります。

運用環境のサーバーレス モード

サーバーレス モードは、UNITY カタログ関数を AI エージェント ツールとして実行する場合に、運用環境のユース ケースに既定で推奨されるオプションです。 このモードでは、サーバーレス 汎用コンピューティング (Spark Connect サーバーレス) を使用してリモートで関数を実行します。 Lakeguard は、エージェントのプロセスを安全に保ち、任意のコードをローカルで実行するリスクを防ぐことができます。

注

AI エージェント ツールとして実行される Unity カタログ関数には、サーバーレス SQL ウェアハウスではなく、サーバーレス 汎用コンピューティング (Spark Connect サーバーレス) が必要です。 サーバーレス 汎用コンピューティングなしでツールを実行しようとすると、 PERMISSION_DENIED: Cannot access Spark Connectなどのエラーが発生します。

# Defaults to serverless if `execution_mode` is not specified
client = DatabricksFunctionClient(execution_mode="serverless")

エージェントが サーバーレス モードでツールの実行を要求すると、次の処理が行われます。

1. DatabricksFunctionClientは、定義がローカルにキャッシュされていない場合に関数定義を取得する要求を Unity カタログに送信します。
2. DatabricksFunctionClientは、関数定義を抽出し、パラメーターの名前と型を検証します。
3. DatabricksFunctionClientは、実行を UDF としてサーバーレス 汎用コンピューティングに送信します。

開発用のローカル モード

ローカル モードでは、サーバーレス 汎用コンピューティングPython要求を行う代わりに、ローカル サブプロセスで関数が実行されます。 これにより、ローカル スタック トレースを提供することで、ツール呼び出しのトラブルシューティングをより効果的に行うことができます。 Unity カタログ関数Python開発およびデバッグ用に設計されています。

エージェントが ローカル モードでツールの実行を要求すると、 DatabricksFunctionClient は次の処理を行います。

1. 定義がローカルにキャッシュされていない場合は、関数定義を取得する要求を Unity カタログに送信します。
2. Python呼び出し可能な定義を抽出し、呼び出し可能な定義をローカルにキャッシュし、パラメーターの名前と型を検証します。
3. 指定されたパラメーターを使用し、タイムアウト保護付きの制限されたサブプロセス内で、呼び出し可能な関数を呼び出します。

# Defaults to serverless if `execution_mode` is not specified
client = DatabricksFunctionClient(execution_mode="local")

"local" モードで実行すると、次の機能が提供されます。

• CPU 時間制限: 過剰な計算負荷を防ぐために、呼び出し可能な実行の合計 CPU ランタイムを制限します。

  CPU 時間制限は、実際の CPU 使用率に基づいており、ウォール クロック時間ではありません。 システム スケジューリングと同時実行プロセスにより、CPU 時間は実世界のシナリオで壁時計時間を超える可能性があります。

• メモリ制限: プロセスに割り当てられた仮想メモリを制限します。

• タイムアウト保護: 関数を実行するための合計ウォール クロック タイムアウトを適用します。

環境変数を使用してこれらの制限をカスタマイズします (詳細を参照)。

ローカル モードの制限事項

• Python 関数のみ: SQL ベースの関数はローカル モードではサポートされていません。
• 信頼されていないコードのセキュリティに関する考慮事項: ローカル モードでは、プロセスの分離のためにサブプロセスで関数が実行されますが、AI システムによって生成された任意のコードを実行すると、潜在的なセキュリティ リスクがあります。 これは主に、レビューされていない動的に生成されたPythonコードを関数が実行する場合に問題になります。
• ライブラリのバージョンの違い: ライブラリのバージョンは、サーバーレス環境とローカル実行環境で異なる場合があり、関数の動作が異なる可能性があります。

環境変数

次の環境変数を使用して、 DatabricksFunctionClient で関数を実行する方法を構成します。

環境変数	既定値	説明
EXECUTOR_MAX_CPU_TIME_LIMIT	10 秒	最大許容 CPU 実行時間 (ローカル モードのみ)。
EXECUTOR_MAX_MEMORY_LIMIT	100 MB（メガバイト）	プロセスの最大許容仮想メモリ割り当て (ローカル モードのみ)。
EXECUTOR_TIMEOUT	20 秒	ウォール クロックの最大合計時間 (ローカル モードのみ)。
UCAI_DATABRICKS_SESSION_RETRY_MAX_ATTEMPTS	5	トークンの有効期限が切れた場合にセッション クライアントの更新を再試行する最大試行回数。
UCAI_DATABRICKS_SERVERLESS_EXECUTION_RESULT_ROW_LIMIT	100	サーバーレス コンピューティングと databricks-connectを使用して関数を実行するときに返す最大行数。

サンプルノートブック

次のノートブックでは、Unity カタログ関数を使用して外部サービスに接続する AI エージェント ツールを作成する方法を示します。

Slack メッセージング エージェント ツール

ノートブックを入手

Microsoft Graph API エージェント ツール

ノートブックを入手

Azure AI Search エージェント ツール

ノートブックを入手

次のステップ

• プログラムを使用して Unity カタログ ツールをエージェントに追加します。 「AI エージェントを作成して Databricks Apps にデプロイする」を参照してください。

• AI Playground UI を使用してエージェントに Unity カタログ ツールを追加します。 「 作業の開始: コードなしで LLM にクエリを実行し、AI エージェントをプロトタイプ作成する」を参照してください。

• Function Client を使用して Unity カタログ関数を管理します。 Unity カタログのドキュメントを参照する - 関数クライアント

• エージェントを外部サービスに接続する すべての方法の概要をご覧ください。