Host Model Context Protocol (MCP) Servers on Databricks | Mediumの翻訳です。
本書は著者が手動で翻訳したものであり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
著者: Jai Behl (Sr. Solutions Architect @ Databricks)
AIエージェントを構築しているほとんどのチームは最終的には同じ壁に直面します: 自分たちのエージェントを既存のシステムやワークフローに接続する必要がありますが、すべてのインテグレーションがカスタムのエンジニアリングのプロジェクトになってしまいます。モデルコンテキストプロトコル(MCP)は、エージェントが外部のツールやデータソースを特定し、やり取りを行う標準的な方法を提供することで、これを解決します。
Databricksでは、Unity Catalogの関数、Vector Search、GenieスペースのマネージドMCPサーバーを提供していますが、あなたの企業特有のツールやワークフローを公開するためにカスタムのMCPサーバーを必要とすることがあるでしょう。Databricks Appsはこれらのカスタムサーバーをホスティングするための素晴らしいプラットフォームとなります。
なぜカスタムMCPサーバーが重要なのか
すべての企業は、構築や改善に数年を要するドメイン固有のツールを所有しています。ネットワーク監視システム、財務分析ツール、カスタマーサポートワークフローのいずれであろうと、これらのシステムにはAIエージェントにとって重要な組織の知識が含まれています。これらの能力を再構成したり、エージェントごとにワンオフのインテグレーションを作成するのではなく、MCPによって標準化されたインタフェースを通じて一度にこれらを公開することができます。
実際にどのようなものなのかを以下に示します。ネットワークの健康状態を監視する運用ツールがあるものとしましょう:
- 障害ステータス監視: リージョンにまたがる現時点のネットワーク障害のチェック
- ネットワークメトリクス: リージョンごとのリアルタイムのパフォーマンスメトリクスの収集
- 障害の報告: ネットワーク問題のチケットの作成
ツールごとにカスタムのエージェントインテグレーションを構築するのではなく、これら3つの機能全てを公開する一つのMCPサーバーを構築します。
自分のMCPサーバーを構築する
MCP Python SDKによって、サーバーの構築はわかりやすいものとなっています。任意のPython Webフレームワークを使うことができます - 公式のカスタムサーバーの例では、多くのチームがすでに慣れ親しんでいるFastMCPとFastAPIを使っています。このガイドでは、軽量でMCPサーバーとうまく連携するStarletteアプローチを説明します。
サーバーのセットアップと構造
コアのMCPサーバーのセットアップからスタートします:
import contextlib
import logging
from collections.abc import AsyncIterator
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount, Route
import json
# Set up logging for debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("your-mcp-server")
# Create the MCP server
app = Server("your-custom-mcp-server")
ツールハンドラーの実装
あなたのMCPサーバーのコアとなるのはツールをハンドリングするロジックです。アクセスのロギングとセッション管理にコンテキストを使用します:
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""Handle tool calls for your domain-specific operations."""
ctx = app.request_context
if name == "check-system-status":
return await handle_system_status(arguments, ctx)
elif name == "get-metrics":
return await handle_metrics(arguments, ctx)
elif name == "create-ticket":
return await handle_ticket_creation(arguments, ctx)
else:
raise ValueError(f"Unknown tool: {name}")
async def handle_system_status(arguments: dict, ctx) -> list[types.TextContent]:
"""Example tool implementation with proper logging and error handling."""
region = arguments.get("region")
# Log the operation for debugging
await ctx.session.send_log_message(
level="info",
data=f"Checking system status for region: {region}...",
logger="your-mcp-server",
related_request_id=ctx.request_id,
)
try:
# Your actual tool logic here - call APIs, query databases, etc.
# Simulate some async operation
await asyncio.sleep(0.1)
response = {
"timestamp": datetime.now().isoformat(),
"region": region or "All regions",
"status": "operational",
"uptime": "99.9%"
}
await ctx.session.send_log_message(
level="info",
data="System status retrieved successfully",
logger="your-mcp-server",
related_request_id=ctx.request_id,
)
return [types.TextContent(type="text", text=json.dumps(response, indent=2))]
except Exception as e:
await ctx.session.send_log_message(
level="error",
data=f"Error checking system status: {str(e)}",
logger="your-mcp-server",
related_request_id=ctx.request_id,
)
raise
ツールスキーマの定義
エージェントがツールをどのように使うのかを理解できるように、詳細なスキーマと説明文を用いてツールを定義します:
@app.list_tools()
async def list_tools() -> list[types.Tool]:
"""Define the tools your server exposes."""
return [
types.Tool(
name="check-system-status",
description="Check the current operational status of systems by region",
inputSchema={
"type": "object",
"properties": {
"region": {
"type": "string",
"description": "Specific region to check (optional). Examples: 'us-west', 'eu-central'"
}
}
}
),
types.Tool(
name="create-ticket",
description="Create a support ticket for system issues",
inputSchema={
"type": "object",
"required": ["issue_type", "description"],
"properties": {
"issue_type": {
"type": "string",
"enum": ["outage", "performance", "security", "maintenance"],
"description": "Type of issue being reported"
},
"region": {
"type": "string",
"description": "Region where the issue is occurring"
},
"description": {
"type": "string",
"description": "Detailed description of the issue"
}
}
}
)
]
Starletteとの連携
適切なセッション管理と共に全てを繋ぎ合わせます:
# Create the session manager
session_manager = StreamableHTTPSessionManager(
app=app,
event_store=None, # Use None for stateless operation
stateless=True,
)
async def handle_streamable_http(scope, receive, send):
await session_manager.handle_request(scope, receive, send)
@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
async with session_manager.run():
logger.info("MCP server started!")
try:
yield
finally:
logger.info("MCP server stopped!")
# Create the Starlette app with routes
starlette_app = Starlette(
debug=False,
routes=[
Route("/", homepage), # Optional demo interface
Mount("/api/mcp", app=handle_streamable_http) # MCP endpoint
],
lifespan=lifespan,
)
重要な実装のポイント
- 常に非同期を用いましょう: MCPエージェントはブロックされないオペレーションを想定しています
- ロギングを搭載しましょう: デバッグのためにログメッセージを送信するためにコンテキストセッションを用いましょう
- エラーを適切に処理しましょう: try/catchブロックでツールロジックをラッピングしましょう
- 構造化されたJSONを返却しましょう: エージェントがレスポンスを解析するので、一貫性を保ちましょう
- 詳細なスキーマを用いましょう: 適切なツールの説明文やパラメータのスキーマは、エージェントが適切にツールを使用する助けとなります
コンテキストオブジェクト(ctx)は、本番運用におけるエージェントとのやり取りのデバッグで重要となるセッション管理、ロギングへのアクセスを提供します。
Databricks Appsへのデプロイ
Databricks Appsはインフラストラクチャの複雑性の面倒を見るので、あなたはツールにフォーカスすることができます。デプロイメントプロセスは驚くほどに簡単です:
設定ファイル
2つのファイルが必要となります:
app.yaml:
command:
- "python"
- "mcp_server.py"
env:
- name: "DATABRICKS_APP_PORT"
value: "8000"
runtime: python_3.10
requirements.txt:
mcp>=1.9.0
starlette>=0.27.0
uvicorn>=0.22.0
python-dotenv>=1.0.0
デプロイメントスクリプト
基本的なデプロイメントスクリプトは面倒な部分に対応します:
#!/bin/bash
APP_NAME="your-mcp-server"
WORKSPACE_PATH="/Workspace/Users/$(databricks current-user me --output json | jq -r '.userName')/mcp_server"
# Sync source code to workspace
databricks sync . "$WORKSPACE_PATH"
# Create and deploy the app
databricks apps create "$APP_NAME"
databricks apps deploy "$APP_NAME" --source-code-path "$WORKSPACE_PATH"
一度このコードを実行すればライブになります。このスクリプトはコードを同期し、アプリを作成し、デプロイします。
認証処理はおまかせ
ここがDatabricks Appsが真に輝く場所です: 認証処理は組み込まれています。OAuthフローを実装したり、APIキーを管理したり、最初からユーザー認証を実装する必要はありません。
MCPサーバーをデプロイする際には:
- 認証はあなたのワークスペースの設定を継承します
- Unity Catalogの権限が自動的に強制されます
- デフォルトでセキュリティが設定されます
- セキュアなエンドポイントを通じてすぐにサーバーにアクセスできるようになります
これは、認証のインフラストラクチャの管理ではなくツールの構築にフォーカスできることを意味します。
MCPサーバーのテスト
あなたのカスタムMCPサーバーをテストするベストな方法は、Databricks AI Playgroundを用いることです。Playgroundには、ツールとしてDatabricks AppsにホストされたカスタムMCPサーバーの追加をネイティブでサポートしており、エージェントとのインタラクションのプロトタイピングと検証を即座に行う手段を提供します。
エージェントのテストのためにDatabricks AI PlaygroundにカスタムMCPサーバーを追加
なぜPlaygroundでテストするのか
Playgroundでのテストにはいくつかのメリットがあります:
- 本当のエージェントとのインタラクション: LLMがあなたのツールをどのように呼び出し、レスポンスを解釈するのかを正確に確認
- パラメータの検証: あなたのツールのスキーマと必要とするパラメータを用いて問題をクイックに特定
- レスポンスのテスト: あなたのツールの出力がエージェントで利用できるように適切に構造化されていることを検証
- 繰り返しの改善: エージェントがどのようにツールを使用するのかに基づいて、ツールの説明文やスキーマを調整
様々なLLMがあなたのツールとどのように連携するのかを確認するために、隣り合わせに複数のモデルを比較することができ、より良いツールの説明文の記述やエッジケースの対応の助けとなります。
なぜこのアプローチがうまくいくのか
Databricks AppでカスタムMCPサーバーを構築することで、いくつかの運用上のスイートスポットをヒットすることができます:
1つのサーバー、多数のエージェント
あなたのMCPサーバーを構築すると、MCPと互換性のある全てのエージェントから利用できるようになります:
- Databricks Agent Frameworkアプリケーション
- Claude Desktopや他のAnthropickのツール
- MCP Python SDKで構築されたカスタムエージェント
- サードパーティのエージェントプラットフォーム
エージェントごとにカスタムのインテグレーションを開発する必要はありません。
実装のポイント
- 最大2-3のツールからスタートしましょう: 一度に全てのAPI界面を公開しようとしないでください。最も重要なツールをピックアップし、そこから繰り返しましょう。
- 全てを非同期で: 定常的にasync/awaitパターンを使いましょう。同期的なAPI呼び出しを待つことがないことに、あなたのエージェントは感謝することでしょう。
- 意味のある説明文を: LLMがツールを選択する必要があるので、説明文はここでは本当に重要です。
- デプロイメントのスクリプトを作成しましょう: 初日からデプロイメントスクリプトを記述しましょう。開発過程では頻繁にデプロイメントを行うことになります。
まとめ
MCPは、AIエージェントを構築する全てのチームが最終的に直面するインテグレーションの複雑性を解決します。それぞれのエージェントとツールの組み合わせごとにカスタムのインテグレーションではなく、一度構築したらどこでも使うことができます。
Databricks Appsによって、これらのサーバーのホスティングは些細なことになります - 認証、セキュリティ、インフラストラクチャはお任せください。あなたのチームはデプロイメントの複雑性の管理ではなく、適切なツールの公開にフォーカスすることができます。
この組み合わせはパワフルなものです: エンタープライズレベルのインフラストラクチャに下支えされた標準化エージェントインタフェース。エージェントが使う必要のあるドメイン固有のツールをお持ちであれば、このアプローチを探索する価値があると言えます。