Connect a Model Context Protocol (MCP) server with Databricks | Mediumの翻訳です。
本書は著者が手動で翻訳したものであり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
著者: Jai Behl, Sr. Solutions Architect
MCPによるインテリジェントなお天気アシスタントの構築
このチュートリアルでは、予報と警報に関するクエリーにレスポンスできるお天気アシスタントを作成するために、Model Context Protocol (MCP)サーバーを実装し、(クライアントとしての)DatabricksとMeta Llama3.3 70bに接続する方法を探索します。お天気ツールとMCPサーバーのセットアップから、DatabricksのLLMサービスとやり取りを行うクライアントを構築に至る拗ねてのプロセスをウォークスルーします。
Model Context Protocol (MCP)とは？
MCPは、アプリケーションがLLMにコンテキストをどのように提供するのかを標準化する、Anthropicによってリリースされたオープンプロトコルです。MCPをAIアプリケーションの標準化されたポートと考えましょう - 様々なデータソースやツールにAIモデルが接続するための一貫性のある手段を提供します。
ハイレベルにおいては、MCPは双方向の通信を用いたクライアント/サーバーのアーキテクチャを実装します。サーバーは、LLMが利用できるツールをホストし、クラインとはユーザー、LLM、ツール間の会話をオーケストレーションします。
なぜMCPがゲームチェンジャーなのか？
MCPはAIアプリケーションの開発にいくつかのキーとなるメリットをもたらします:
- 関心の分離: ツール開発者は、コアのAIシステムから独立して機能を作成、更新できます。
- モジュール性: システム規模の変更なしに、新たなツールをインクリメンタルに追加できます。
- 標準化: 一貫性のあるインタフェースは、異なるコンポーネントにおけるインテグレーションの作業が少なくて済むことを意味します。
- スケーラビリティ: 要求に基づいてコンポーネントを独立してスケールさせることができます。
Databricksのデータインテリジェンスプラットフォームを構築している企業においては、MCPは背後のモデルを修正することなしにLLMの能力を拡張するクリーンな手段を提供します。
アーキテクチャとフロー
こちらが、実装するアーキテクチャとなります:
- ユーザーは天気に関するクエリーを送信します(「サンフランシスコの天気は？」など)
- MCPクライアントはこれをDatabricksのOpenAI互換の基盤モデルAPIと利用可能なツールに転送します
- Meta Llamaが使用するツールを決定し、ツール呼び出しを行います
- MCPクライアントはMCPサーバーにツール呼び出しを引き渡します
- MCPサーバーはツールを実行し、結果を返却します
- クライアントは結果を整形し、ユーザーに表示します
実装に踏み込みましょう！
お天気MCPサーバー
このサーバーの実装は、こちらで共有されている指示にシンプルに従ったものです: https://modelcontextprotocol.io/quickstart/server
MCPサーバー(weather.py)は2つの主要なツールを提供します:
- get_forecast: 特定の緯度/経度の天気予報を取得します
- get_alerts: USの州でアクティブな天気警報を取得します
こちらがコアコンポーネントとなります:
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("weather")
# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
# Initialize HTTP client
http = httpx.AsyncClient(timeout=10.0)
このサーバーは、リアルタイムの気候データを取得するために、National Weather Service APIにリクエストを行います。予報ツールを見てみましょう:
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""Get weather forecast for a location.
Args:
latitude: Latitude of the location
longitude: Longitude of the location
"""
# First get the forecast grid endpoint
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
# Get the forecast URL from the points response
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
# Format the periods into a readable forecast
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # Only show next 5 periods
forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)
同様に警報ツールを見てみましょう:
@mcp.tool()
async def get_alerts(state: str) -> str:
"""Get weather alerts for a US state.
Args:
state: Two-letter US state code (e.g. CA, NY)
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
MCPサーバーは以下によってローカルで実行します:
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')
これによって、標準入力/出力を通じて通信を行うようにサーバーをセットアプし、同じマシンにある我々のクライアントと容易に接続できるようになります。
DatabricksによるMCPクライアントの構築
それではエキサイティングな部分です - DatabricksのMeta LlamaモデルにMCPサーバーを接続します！我々のクライアント(connect.py)は以下を担当します:
- MCPサーバーに接続します
- DatabricksのAPIが理解できるようにMCPツールを変換します
- DatabricksのOpenAI互換インタフェースを通じてMeta Llama3.3 80bを呼び出します
- ツール呼び出しとレスポンスを処理します
Databricks接続のセットアップ
我々のクライアントのコアはMCPClientクラスです:
class MCPClient:
def __init__(self, llm_endpoint_name: str = "databricks-meta-llama-3-3-70b-instruct"):
# Initialize session and client objects
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
# Initialize Databricks client and get OpenAI-compatible client
self.workspace = WorkspaceClient()
self.llm_endpoint_name = llm_endpoint_name
self.openai_client = self.workspace.serving_endpoints.get_open_ai_client()
print(f"Initialized OpenAI-compatible client for {llm_endpoint_name}")
ここでは、以下を行うためにDatabricks SDKを使用しています:
- Databricksワークスペースに接続するWorkspaceClientを作成
- 特定のLLMエンドポイント(デフォルトはMeta Llama 3.3)のOpenAI互換のクライアントの取得
このアプローチでは、Databricksのモデルサービングの統合APIを活用しており、馴染みのあるOpenAIインタフェースパターンを活用して、Meta Llamaモデルにアクセスすることができます。
MCPサーバーへの接続
クライアントは我々のMCPサーバーへの接続を確立します:
async def connect_to_server(self, server_script_path: str):
"""Connect to an MCP server"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")
command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# List available tools
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
# Display tool descriptions
print("\nAvailable Tools:")
for tool in tools:
print(f"- {tool.name}: {tool.description}")
この関数は以下を行います:
- ファイルの拡張子に基づいてどのようにサーバースクリプトを実行するのかを決定します
- サーバーと通信するためにstdio通信チャネルをセットアップします
- MCPクライアントセッションを初期化します
- サーバーで利用できるツールを一覧します
Meta LLamaのためにMCPツールを変換
我々の実装の重要な部分は、Databricksのクライアントが期待するOpenAI関数フォーマットにMCPツールを変換するところです:
def _convert_schema_to_openai_format(self, tool):
"""Convert MCP tool schema to OpenAI function format"""
# Extract the input schema from the tool
schema = tool.inputSchema
# Create the function definition
function_def = {
"name": tool.name,
"description": tool.description,
"parameters": schema
}
# Return in the OpenAI tool format
return {
"type": "function",
"function": function_def
}
この変換レイヤーによって、MCPのツール表現とDatabricksのAPIを通じてMeta Llamaによって期待される関数呼び出しフォーマットの間の橋渡しを行うことができます。
Meta Llama 3.3 70bによるクエリーの処理
クライアントの心臓は以下を行うprocess_queryメソッドです:
- ユーザーのクエリーを受け取ります
- クエリーをMeta Llamaと利用可能なツールに送信します
- モデルが行う任意のツール呼び出しを処理します
- 整形された結果を返却します
async def process_query(self, query: str) -> str:
"""Process a query using Meta Llama and available tools"""
if not self.session:
print("Error: Not connected to a server")
return "Not connected to server."
# Initialize conversation with user query
messages = [
{"role": "user", "content": query}
]
# Get available tools and convert to OpenAI format
response = await self.session.list_tools()
available_tools = [self._convert_schema_to_openai_format(tool) for tool in response.tools]
final_text = []
try:
# Make initial LLM API call
print("Sending query to LLM with tools...")
llm_response = self.openai_client.chat.completions.create(
model=self.llm_endpoint_name,
messages=messages,
tools=available_tools,
max_tokens=1000
)
# Get the assistant's response
assistant_message = llm_response.choices[0].message
# Add any content from the assistant to our output
if assistant_message.content:
final_text.append(assistant_message.content)
# Check for tool calls
if hasattr(assistant_message, 'tool_calls') and assistant_message.tool_calls:
# Process each tool call
for tool_call in assistant_message.tool_calls:
# ... tool call processing ...
Meta Llamaが使用するツールを決定すると、引数を抽出して、MCPサーバーへの呼び出しを行います:
# Execute tool call
try:
final_text.append(f"\n[Calling tool: {tool_name} with arguments: {tool_args}]")
result = await self.session.call_tool(tool_name, tool_args)
# Extract the raw text from the result
raw_result = str(result.content)
# If it's a TextContent object, extract just the text part
if 'TextContent' in raw_result and "text='" in raw_result:
# Find the text content between quotes
text_start = raw_result.index("text='") + 6
text_end = raw_result.rindex("'")
if text_start < text_end:
cleaned_result = raw_result[text_start:text_end]
# Unescape newlines and tabs
cleaned_result = cleaned_result.replace('\\n', '\n').replace('\\t', '\t')
final_text.append(f"\nForecast for {tool_args.get('latitude', '')},{tool_args.get('longitude', '')}:\n{cleaned_result}")
else:
final_text.append(f"\nResult: {raw_result}")
else:
final_text.append(f"\nResult: {raw_result}")
インタラクティブなチャットループ
最後に、我々のクライアントはユーザーのためにインタラクティブなチャットインタフェースを提供します:
async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Weather Client Started!")
print("You can ask questions about weather. For example:")
print("- What's the weather forecast for San Francisco?")
print("- Are there any weather alerts in CA?")
print("- Tell me the forecast for latitude 37.7749 and longitude -122.4194")
print("(Type 'quit' to exit)")
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
if query.strip():
print("\nProcessing query...")
response = await self.process_query(query)
print("\nResponse:")
print(response)
except Exception as e:
print(f"\nError: {str(e)}")
そして、こちらがmainエントリーポイントです:
async def main():
# Default to weather.py in the current directory if no path provided
server_script_path = sys.argv[1] if len(sys.argv) > 1 else "weather.py"
# Optional second argument for the LLM endpoint name
llm_endpoint_name = sys.argv[2] if len(sys.argv) > 2 else "databricks-meta-llama-3-3-70b-instruct"
client = MCPClient(llm_endpoint_name)
try:
print(f"Connecting to server: {server_script_path}")
await client.connect_to_server(server_script_path)
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())
これによって、サーバースクリプトのパスとLLMエンドポイント名の両方を指定することで、我々のお天気アシスタントとやり取りを行うシンプルなCLIを提供します。
結果
我々のソリューションは信頼性を保証するためにフローの簡素化にフォーカスしました:
- Meta Llamaで初回のクエリーを解析します
- MCPサーバーを通じて任意のツール呼び出しを実行します
- 結果を整形し、ユーザーに結果を表示します
このアプローチは、生計の問題につながる可能性のある複雑な会話のチェーンを回しつつも、有用なお天気アシスタントを提供します。
まとめと次のステップ
このチュートリアルでは、以下を用いた自然言語お天気アシスタントの構築方法を示しました:
- ツールオーケストレーションのためのModel Context Protocol(MCP)
- Meta LlamaモデルにアクセスするためのDatabricks SDK
- リアルタイム気象データのためのNational Weather Service API
このアーキテクチャは、モジュール化され拡張可能なAIアプリケーションの作成におけるMCPのパワーをデモンストレーションします。ツールの実装をLLMとのやり取りから分離することで、新機能で容易に拡張可能なシステムを作成しました。
更なる機能拡張には以下のようなものが含まれるでしょう:
- 更なるお天気ツール(過去のデータ、お天気マップなど)の追加
- 個別のサービスとしてMCPサーバーをデプロイ
DatabricksのLLMの機能とMCPのツールフレームワークの組み合わせは、現実世界のデータソースとやり取りできる洗練されたAIアシスタントの構築におけるパワフルな基盤を提供します。
自分で試しましょう
このお天気アシスタントを実行するには:
- Databricks SDKがインストールされていることを確認します
- あなたのDatabricksワークスペースの資格情報をセットアップします
- お天気サーバーを実行します
- 別のターミナルでクライアントを実行します
お好きな場所の天気を質問し、Databricks上のLlama3.3 70bが正確で最新の情報を提供するために、どのようにツールを使うのかをみてみましょう