Databricks SQL

備考

プレビュー

この機能は パブリック プレビュー段階です。

Databricks SQL MCPサーバーは、エージェントがAI生成したSQLをUnity Catalogテーブルに対してランしてデータの読み取りと書き込みを行うことを可能にするDatabricks マネージドMCPサーバーであり、アクセスはUnity Catalogの権限によって管理されます。これを使用して、AIコーディングツールからデータパイプラインを構築します。クエリーは非同期でランされます。エージェントがツールを呼び出してクエリーを起動し、応答が完了するまでポーリングを行います。

URL パターン	OAuthスコープ
https://<workspace-hostname>/api/2.0/mcp/sql	sql

URL パターン	OAuthスコープ
https://<workspace-hostname>/api/2.0/mcp/sql	sql

_metaパラメーター

_meta パラメーターとは、ツール呼び出し時に LLM が動的に生成するのではなく、MCP サーバーの動作を決定論的に設定するためにエージェントコード内で事前設定する構成値のことです。Databricks SQL MCP サーバーでは、次の_metaパラメーターがサポートされています。

パラメーター名	Type	説明
warehouse_id	str	クエリーの実行に使用するSQLウェアハウスのID。 例： "a1b2c3d4e5f67890" 指定されていない場合、システムはリソースとアクセス許可に基づいてwarehouseを自動的に選択します。

パラメーター名	Type	説明
warehouse_id	str	クエリーの実行に使用するSQLウェアハウスのID。 例： "a1b2c3d4e5f67890" 指定されていない場合、システムはリソースとアクセス許可に基づいてwarehouseを自動的に選択します。

例：Databricks SQLクエリー用のSQL Warehouseを指定する

この例では、warehouse_id _metaパラメーターを使用して、公式の Python MCP SDK を使用して Databricks SQL MCP サーバーからクエリーを実行する SQL Warehouse を指定する方法を示します。 [[ ## completed ##]]

このシナリオでは、次のことを行います：

• システムによる自動選択ではなく、クエリー実行用に特定のSQLウェアハウスを使用します
• クエリーを専用のwarehouseにルーティングして、一貫したパフォーマンスを確認する

この例をランするには、マネージド MCP 開発用に Python 環境をセットアップしてください：

SQLウェアハウス ID を確認するには、「SQLウェアハウスへの接続」を参照してください。

Python

# Import required libraries for MCP client and Databricks authentication
import asyncio
from databricks.sdk import WorkspaceClient
from databricks_mcp.oauth_provider import DatabricksOAuthClientProvider
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession
from mcp.types import CallToolRequest, CallToolResult

async def run_dbsql_tool_call_with_meta():
    # Initialize Databricks workspace client for authentication
    workspace_client = WorkspaceClient()

    # Construct the MCP server URL for DBSQL
    # Replace <workspace-hostname> with your workspace hostname
    mcp_server_url = "https://<workspace-hostname>/api/2.0/mcp/sql"

    # Establish connection to the MCP server with OAuth authentication
    async with streamablehttp_client(
        url=mcp_server_url,
        auth=DatabricksOAuthClientProvider(workspace_client),
    ) as (read_stream, write_stream, _):

        # Create an MCP session for making tool calls
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the session before making requests
            await session.initialize()

            # Create the tool call request with warehouse_id in _meta
            request = CallToolRequest(
                method="tools/call",
                params={
                    # Tool name for executing SQL queries
                    &quot;name&quot;: &quot;execute_sql&quot;,

                    # Dynamic arguments - typically provided by your AI agent
                    &quot;arguments&quot;: {
                        &quot;query&quot;: &quot;SELECT * FROM my_catalog.my_schema.my_table LIMIT 10&quot;
                    },

                    # Meta parameters - specify which warehouse to use
                    &quot;_meta&quot;: {
                        &quot;warehouse_id&quot;: &quot;a1b2c3d4e5f67890&quot;  # Your SQL warehouse ID
                    }
                }
            )

            # Send the request and get the response
            response = await session.send_request(request, CallToolResult)
            return response

# Execute the async function and get results
response = asyncio.run(run_dbsql_tool_call_with_meta())