Databricks自定义MCP服务器

custom_mcp/
├── app.py              # MCPサーバーのメインアプリケーション
├── main.py             # Uvicornサーバーのエントリーポイント
├── app.yaml            # Databricks Apps設定ファイル
├── databricks.yml      # Databricksバンドル設定
├── requirements.txt    # Python依存関係
└── README.md           # このファイル

python3 -m venv .venv
source .venv/bin/activate  # Windowsの場合: .venv\Scripts\activate

pip install -r requirements.txt

python main.py

curl http://localhost:8000/health

command: ["python", "main.py"]
env:
  - name: 'WAREHOUSE_ID'
    value: 'your-warehouse-id-here'

databricks apps deploy

databricks apps list

{
  "mcpServers": {
    "databricks-custom-mcp": {
      "url": "https://your-databricks-app-url.com",
      "transport": "http"
    }
  }
}

トヨタ自動車（7203）の株価情報を教えてください

SQLクエリ "SELECT * FROM catalog.schema.table LIMIT 10" を実行してください

# ❌ 間違った方法
@mcp.tool(name="tool_name", description="...")
def my_tool():
    pass

app.mount("/mcp", mcp_app)  # ❌ サブパスにマウント

# ✅ 正しい方法
@mcp.tool()
def my_tool():
    """ツールの説明をdocstringに記載"""
    pass

app.mount("/", mcp_app)  # ✅ ルートパスにマウント

# ✅ 正しい方法
mcp_app = mcp.streamable_http_app()
app = FastAPI(lifespan=lambda _: mcp.session_manager.run())
app.mount("/", mcp_app)

@mcp.tool()
def run_query_on_databricks(sql_query: str):
    # ... クエリ実行 ...
    df = pd.DataFrame(rows, columns=[...])
    return df.to_string()  # ✅ 文字列に変換

# ✅ 正しい方法
uvicorn.run(
    "app:app",  # インポートパスを文字列で指定
    host="0.0.0.0",
    port=8000,
    reload=True,
)

@mcp.tool()
def my_custom_tool(param1: str, param2: int) -> str:
    """このツールの説明をここに記載します。

    LLMがツールを選択する際に、この説明を参考にします。
    """
    # ツールの実装
    result = some_operation(param1, param2)
    return str(result)  # 文字列で返すことを推奨

@mcp.resource("data://{resource_id}")
def get_resource(resource_id: str) -> str:
    """動的リソースを取得"""
    return f"Resource content for {resource_id}"

databricks apps logs <app-name>