こちらのガイドに従って実際にカスタムMCPサーバーをDatabricksで動かしてみます。
自分で動作確認したバージョンはこちらです。構成をシンプルにしたり、東京証券取引所のデータを取得するように変更しています。
MCPサーバーの実装
import uvicorn
def main():
uvicorn.run(
"app:app",
host="0.0.0.0",
port=8000,
reload=True,
)
if __name__ == "__main__":
main()
import yfinance as yf
import requests
import pandas as pd
from io import StringIO
import databricks.sql
import os
from databricks.sdk.core import Config
from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI
# MCPサーバーを作成
mcp = FastMCP("Databricks Apps上のカスタムMCPサーバー")
# ツール 1
@mcp.tool()
def get_stock_info(symbol: str) -> str | None:
"""指定されたシンボルの株式情報を取得します。TSE（東京証券取引所）株式の時価総額、現在価格、52週レンジを取得します。"""
try:
info = yf.Ticker(f"{symbol}.T").fast_info
return (
f"\n{symbol}.Tの詳細:\n"
f"時価総額 = {info['marketCap']}\n"
f"現在価格 = {info['last_price']} 円\n"
f"年間最低価格 = {info['year_low']} 円\n"
f"年間最高価格 = {info['year_high']} 円\n"
)
except Exception as e:
print(f"{symbol}の取得エラー: {e}")
return None
# ツール 2
@mcp.tool()
def run_query_on_databricks(sql_query: str):
"""SQLクエリを受け取り、Databricks SQLウェアハウスでクエリを実行します。結果をpandas DataFrameとして返します。"""
cfg = Config()
warehouse_id = os.getenv("WAREHOUSE_ID")
host = cfg.host
http_path = f"/sql/1.0/warehouses/{warehouse_id}"
if not all([host, http_path]):
raise ValueError("必要なDatabricks接続環境変数が不足しています。")
with databricks.sql.connect(server_hostname=host, http_path=http_path, credentials_provider=lambda: cfg.authenticate) as conn, conn.cursor() as cur:
cur.execute(sql_query)
rows = cur.fetchall()
df = pd.DataFrame(rows, columns=[c[0] for c in cur.description])
# DataFrameを文字列形式で返す
return df.to_string()
# MCPアプリをHTTPストリーミングアプリに変換
mcp_app = mcp.streamable_http_app()
# FastAPIアプリを作成
app = FastAPI(
lifespan=lambda _: mcp.session_manager.run(),
)
# ルートエンドポイント（オプショナル）
@app.get("/health", include_in_schema=False)
async def health():
return {
"status": "healthy",
"name": "Databricks Apps上のカスタムMCPサーバー",
"version": "1.0.0",
"tools": [
"get_stock_info - 指定されたシンボルの株式情報を取得",
"run_query_on_databricks - Databricks SQLウェアハウスでクエリを実行"
]
}
# MCPアプリをルートパスにマウント
app.mount("/", mcp_app)
command: ["python", "main.py"]
env:
- name: 'WAREHOUSE_ID'
value: '<SQLウェアハウスのID>'
bundle:
name: custom-mcp-server
sync:
include:
- .build
artifacts:
default:
type: whl
path: .
build: uv build --wheel
resources:
apps:
custom-mcp-server:
name: "mcp-custom-server-bundles"
description: "Custom MCP Server on Databricks Apps"
source_code_path: ./.build
targets:
dev:
mode: development
default: true
fastmcp
fastapi
uvicorn
yfinance
pandas
databricks-sql-connector
databricks-sdk
python-dotenv
Databricks Appとしてデプロイ
ローカルで以下を実行していきます。認証を行います。
databricks configure
ローカルのコードをワークスペースに同期します。
DATABRICKS_USERNAME=$(databricks current-user me | jq -r .userName)
databricks sync . "/Users/$DATABRICKS_USERNAME/mcp-custom-server"
アプリを作成します。
databricks apps create mcp-taka
アプリにコードをデプロイします。
databricks apps deploy mcp-taka --source-code-path "/Workspace/Users/$DATABRICKS_USERNAME/mcp-custom-server/custom-server"
アプリの画面で動作していることを確認します。
https://mcp-taka-xxxx.aws.databricksapps.com/healthというURLでMCPサーバーの情報を確認できます。
テーブルに書き込みを行うので、書き込みを行うスキーマにアプリのサービスプリンシパルに権限を付与しておきます。
AI Playgroundで動作確認
AI Playgroundにアクセスします。モデルはClaude Sonnet 4を選択します。
ツールをクリックし、MCP Serversタブを開いてDatabricksアプリのMCPサーバーで上で動かしたアプリを選択します。
システムプロンプトを指定します。
あなたは株式市場分析をサポートするAIエージェントです。MCPサーバーを通じて様々なツールを利用できます。状況に応じて適切なツールを選択し、必要に応じてあなた自身の知識も活用してください。回答は簡潔で分かりやすくまとめてください。
エージェントに指示を行います。
TSE（東京証券取引所）に上場している時価総額上位5社の株式情報を教えてください。
最初はテーブルからデータを取り出そうとしますが、その後はget_stock_infoを使用して情報を取得しています。
結果を得ることができました。
次にテーブル作成と組み合わせます。
TSE（東京証券取引所）に上場している時価総額上位5社の株式情報を格納するため、まずDeltaテーブルの構造を作成してください。カタログ'takaakiyayoi_catalog'、スキーマ'agents'内に、株式データに必要なカラムを持つテーブルを作成してください。その後で、市場情報から上位5社の情報を取得し、作成したテーブルにデータをインサートしてください。
試行錯誤しながらもテーブルを作成してデータをインサートしてくれています。
テーブルに企業情報がインサートされました。
なお、アプリのサービスプリンシパルが作成したテーブルにはアクセス権がない場合がありますので、その場合には権限を付与ください。
