Databricks一斉検索 MCP サーバーのメタ争奪
備考
ベータ版
この機能はベータ版です。
Databricksで管理された一連検索 MCP サーバーを使用するAIエージェントを構築する場合は、 _meta問題を使用して、返される結果の最大数や検索フィルターの適用などのツールの動作を決定的に制御します。
主な利点は、エージェントが動的に生成できるように実際の検索クエリを柔軟に保ちながら、検索動作（結果の制限やフィルターなど）を事前に設定できることです。
_meta問題は、公式のMCP 仕様の一部です。
ツール呼び出しの引数と_meta問題
引き続き検索 MCP サーバーは次の 2 つの方法で問題を処理します。
- ツール呼び出し引数 : 通常、ユーザー入力に基づいてLLMによって動的に生成される問題
_meta問題 : 検索動作を決定的に制御するためにエージェント コードで事前設定できる構成の問題
サポートされている_meta問題
次の_meta問題が、一斉検索でサポートされています。
パラメーター名 | Type | 説明 |
|---|---|---|
|
| 返される結果の数。 デフォルト: 5 |
|
| 検索に適用するフィルターを含む JSON 文字列。有効な JSON 文字列形式である必要があります。 例： |
|
| 結果を取得するために使用する検索アルゴリズム。 サポートされている値: デフォルト： |
これらの問題に関する詳細情報については、「一斉検索APIドキュメント」を参照してください。
例: ベクトル検索取得の最大結果とフィルターを構成する
この例では、 _metaを使用して、 AIエージェントからの動的クエリを許可しながら、待機検索の動作を制御する方法を示します。 公式の Python MCP SDK を使用します。
このシナリオでは、次のことを行います。
- 応答時間を一定に保つために、検索結果を常に 3 項目に制限します。
- 関連性を確保するため、最近のドキュメント（2024年1月1日以降に更新されたもの）のみを検索します。
- ハイブリッド検索を使用すると、純粋なベクトル検索よりも精度が向上します。
この例を実行するには、マネージド MCP 開発用の Python 環境をセットアップします。
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
}
}
)
# 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())