Databricksで管理される MCP サーバーのメタ問題
プレビュー
この機能は パブリック プレビュー段階です。
Databricksで管理される MCP サーバーを使用するAIエージェントを構築する場合は、 _metaを使用して、結果の制限、検索フィルタ、 SQLウェアハウスの選択などのツールの動作を制御します。 これにより、エージェントが動的に生成するクエリを柔軟に保ちながら、構成を事前に設定できます。
_metaパラメーターは、公式のMCP 仕様の一部です。
ツール呼び出しの引数と_metaパラメーター
Databricksで管理される MCP サーバーは、次の 2 つの方法で問題を処理します。
- ツール呼び出しの引数 : LLM通常、ユーザー入力に基づいて動的に生成されるという問題
_meta問題 : 動作を決定論的に制御するためにエージェント コードで事前設定できる構成の問題
DBSQL MCP サーバー_metaの問題
DBSQL MCP サーバーでは、次の_meta問題がサポートされています。
パラメーター名 | Type | 説明 |
|---|---|---|
|
| クエリの実行に使用するSQLウェアハウスの ID。 例： 指定しない場合、システムはリソースと権限に基づいてウェアハウスを自動的に選択します。 |
例: DBSQLクエリ用のSQLウェアハウスを指定する
この例では、 warehouse_id _meta問題を使用して、公式Python MCP SDKを使用して DBSQL MCP サーバーからクエリを実行するSQLウェアハウスを指定する方法を示します。
このシナリオでは、次のことを行います。
- システムに自動的に選択させるのではなく、クエリの実行に特定のSQLウェアハウスを使用する
- クエリを専用ウェアハウスにルーティングすることで一貫したパフォーマンスを確保
この例を実行するには、マネージド MCP 開発用の Python 環境をセットアップします。
コード例を展開する
SQLウェアハウス ID を確認するには、 SQLウェアハウスへの接続」を参照してください。
# 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
"name": "execute_sql",
# Dynamic arguments - typically provided by your AI agent
"arguments": {
"query": "SELECT * FROM my_catalog.my_schema.my_table LIMIT 10"
},
# Meta parameters - specify which warehouse to use
"_meta": {
"warehouse_id": "a1b2c3d4e5f67890" # 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())
トレンド検索 MCP サーバー_meta問題
次の_metaパラメーターが、Vector Searchでサポートされています。
パラメーター名 | Type | 説明 |
|---|---|---|
|
| 検索結果で返される列名のコンマ区切りリスト。 例： 指定しない場合は、すべての列（「__」で始まる内部列を除く）が返されます。 |
|
| 再ランキング モデルが再スコアリングに使用するコンテンツを持つ列名のコンマ区切りリスト。リランカーはこのコンテンツを使用してすべての検索結果のスコアを再調整し、関連性を向上させます。 例： 指定しない場合は再ランク付けは実行されません。 |
|
| 検索に適用するフィルターを含む JSON 文字列。有効な JSON である必要があります。 例： 指定しない場合はフィルターは適用されません。 |
|
| 返される結果に類似度スコアを含めるかどうか。 サポートされている値: デフォルト： |
|
| 返される結果の数。 例： |
|
| 結果を取得するために使用する検索アルゴリズム。 サポートされている値: デフォルト： |
|
| 結果をフィルタリングするための類似度スコアの最小しきい値。このしきい値を下回るスコアの結果は除外されます。 例： 指定しない場合は、スコア フィルタリングは適用されません。 |
これらの問題の詳細については、「一斉検索Python SDKドキュメント」を参照してください。
例: ベクトル検索取得の最大結果とフィルターを構成する
この例では、公式Python MCP SDKを使用してAIエージェントからの動的クエリを許可しながら、 _meta使用して追跡検索の動作を制御する方法を示します。
このシナリオでは、次のことを行います。
- 応答時間を一定に保つために、検索結果を常に 3 項目に制限します。
- 関連性を確保するため、最近のドキュメント（2024年1月1日以降に更新されたもの）のみを検索します。
- ハイブリッド検索を使用すると、純粋なベクトル検索よりも精度が向上します。
- 特定の列（ID、テキスト、メタデータ）のみを返す
- 結果に類似度スコアを含める
- 類似度スコアが0.5未満の結果を除外する
- 関連性を高めるためにテキストとタイトルの列の再ランク付けを使用する
この例を実行するには、マネージド MCP 開発用の 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_vector_search_tool_call_with_meta():
# Initialize Databricks workspace client for authentication
workspace_client = WorkspaceClient()
# Construct the MCP server URL for your specific catalog and schema
# Replace <workspace-hostname>, YOUR_CATALOG, and YOUR_SCHEMA with your values
mcp_server_url = "https://<workspace-hostname>/api/2.0/mcp/vector-search/YOUR_CATALOG/YOUR_SCHEMA"
# 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 both dynamic and preset parameters
request = CallToolRequest(
method="tools/call",
params={
# Tool name follows the pattern: CATALOG__SCHEMA__INDEX_NAME
"name": "YOUR_CATALOG__YOUR_SCHEMA__YOUR_INDEX_NAME",
# Dynamic arguments - typically provided by your AI agent or user input
"arguments": {
"query": "How do I reset my password?" # This comes from your agent
},
# Meta parameters - preset configuration to control search behavior
"_meta": {
"num_results": "3", # Limit to 3 results for consistent performance
"filters": '{"updated_after": "2024-01-01"}', # JSON string for date filtering
"query_type": "HYBRID", # Use hybrid search for better relevance
"columns": "id,text,metadata", # Return only specific columns
"score_threshold": "0.5", # Filter out results with similarity score < 0.5
"include_score": "true", # Include similarity scores in results
"columns_to_rerank": "text,title" # Use reranker on these columns for better quality
}
}
)
# 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_vector_search_tool_call_with_meta())