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が正確で最新の情報を提供するために、どのようにツールを使うのかをみてみましょう

はじめてのDatabricks

はじめてのDatabricks

Databricks無料トライアル

Databricks無料トライアル