Cara Menggunakan Server MCP Databricks

💡

Sebelum menyelami Databricks MCP Server, lihat Apidog—alat gratis yang dirancang untuk menyederhanakan pengujian dan integrasi API. Dengan antarmuka Apidog yang intuitif, Anda dapat dengan mudah melakukan debug dan mengoptimalkan alur kerja API Anda, merampingkan proses pengembangan, dan menghemat waktu berharga Anda. Baik Anda membuat API atau memecahkan masalah, Apidog memiliki semua yang Anda butuhkan untuk meningkatkan alur kerja Anda.

button

Model Context Protocol (MCP) adalah protokol terbuka yang awalnya dikembangkan oleh Anthropic yang menstandarisasi bagaimana aplikasi menyediakan konteks ke Large Language Models (LLM). Databricks MCP Server bertindak sebagai jembatan antara platform analitik data Databricks yang kuat dan model AI, memungkinkan LLM untuk berinteraksi dengan lingkungan Databricks Anda secara terprogram. Integrasi ini memungkinkan asisten AI untuk menjalankan kueri SQL, membuat daftar dan mendapatkan detail eksekusi pekerjaan, dan mengakses data di dalam akun Databricks Anda tanpa intervensi manusia secara langsung.

Dalam panduan komprehensif ini, kita akan membahas proses pengaturan, konfigurasi, dan pemanfaatan Databricks MCP Server untuk operasi dan analitik data berbasis AI yang ditingkatkan.

Memahami Model Context Protocol (MCP)

Sebelum menyelami implementasi khusus Databricks, penting untuk memahami apa itu MCP dan mengapa itu penting:

MCP mengimplementasikan arsitektur klien/server dengan komunikasi dua arah:

Server menghosting alat yang dapat digunakan LLM
Klien mengatur percakapan antara pengguna, LLM, dan alat

Manfaat utama yang membuat MCP berharga meliputi:

Pemisahan Perhatian: Pengembang alat dapat membuat dan memperbarui kemampuan secara independen
Modularitas: Alat baru dapat ditambahkan secara bertahap tanpa perubahan di seluruh sistem
Standardisasi: Antarmuka yang konsisten berarti lebih sedikit pekerjaan integrasi di seluruh komponen
Skalabilitas: Komponen dapat diskalakan secara independen berdasarkan permintaan

Prasyarat untuk Menjalankan Databricks MCP Server

Sebelum menyiapkan Databricks MCP Server, pastikan Anda memiliki:

Akun Databricks dengan izin yang sesuai
Python 3.8+ terinstal di sistem Anda
Databricks CLI dikonfigurasi dengan autentikasi
Pemahaman dasar tentang konsep ruang kerja Databricks
Keakraban dengan SQL dan pekerjaan Databricks

Langkah 1: Menyiapkan Lingkungan

Pertama, mari kita siapkan lingkungan kita dengan menginstal paket yang diperlukan:

# Buat lingkungan virtual
python -m venv databricks-mcp-env
source databricks-mcp-env/bin/activate  # Di Windows: databricks-mcp-env\\\\Scripts\\\\activate

# Instal paket yang diperlukan
pip install databricks-sdk mcp-server httpx fastapi uvicorn

Langkah 2: Membuat Databricks MCP Server

Buat file baru bernama databricks_mcp.py dengan struktur kode berikut:

from typing import Any, Dict, List, Optional
from mcp.server.fastmcp import FastMCP
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.sql import ExecuteStatementRequest, StatementState

# Inisialisasi server FastMCP
mcp = FastMCP("databricks")

# Inisialisasi klien Databricks
workspace = WorkspaceClient()

Langkah 3: Mengimplementasikan Alat MCP Inti untuk Databricks

Sekarang, mari kita implementasikan alat inti yang akan memungkinkan LLM untuk berinteraksi dengan lingkungan Databricks Anda:

@mcp.tool()
async def run_sql_query(query: str, warehouse_id: Optional[str] = None) -> Dict[str, Any]:
    """Jalankan kueri SQL di gudang Databricks SQL.

    Args:
        query: Kueri SQL untuk dieksekusi
        warehouse_id: ID opsional dari gudang SQL yang akan digunakan

    Returns:
        Dictionary yang berisi hasil kueri
    """
    # Validasi kueri untuk keamanan
    if not query or any(keyword in query.lower() for keyword in ["drop", "delete", "truncate", "alter"]):
        return {"error": "Operasi SQL yang berpotensi merusak tidak diizinkan"}

    try:
        # Eksekusi kueri
        statement = workspace.statement_execution.execute_statement(
            warehouse_id=warehouse_id,
            statement=query,
            wait_timeout=60
        )

        # Proses hasil
        if statement.status.state == StatementState.SUCCEEDED:
            # Konversi hasil ke format yang lebih mudah digunakan
            columns = [col.name for col in statement.manifest.schema.columns]
            rows = []
            for chunk in workspace.statement_execution.get_statement_result_chunks(statement.statement_id):
                for row_data in chunk.data:
                    rows.append(dict(zip(columns, row_data)))

            return {
                "success": True,
                "columns": columns,
                "rows": rows,
                "row_count": len(rows)
            }
        else:
            return {
                "success": False,
                "state": statement.status.state,
                "error": statement.status.error
            }
    except Exception as e:
        return {"success": False, "error": str(e)}

Selanjutnya, mari kita tambahkan alat untuk membuat daftar database yang tersedia:

@mcp.tool()
async def list_databases() -> List[str]:
    """Buat daftar semua database yang tersedia di ruang kerja Databricks.

    Returns:
        Daftar nama database
    """
    try:
        # Jalankan kueri untuk mendapatkan semua database
        result = await run_sql_query("SHOW DATABASES")
        if result.get("success"):
            # Ekstrak nama database dari hasil
            return [row.get("databaseName") for row in result.get("rows", [])]
        else:
            return [f"Error membuat daftar database: {result.get('error')}"]
    except Exception as e:
        return [f"Error: {str(e)}"]

Sekarang, mari kita implementasikan alat untuk mendapatkan skema tabel:

@mcp.tool()
async def get_table_schema(database_name: str, table_name: str) -> Dict[str, Any]:
    """Dapatkan skema tabel tertentu.

    Args:
        database_name: Nama database
        table_name: Nama tabel

    Returns:
        Dictionary yang berisi informasi skema tabel
    """
    try:
        # Jalankan kueri untuk mendapatkan skema tabel
        result = await run_sql_query(f"DESCRIBE {database_name}.{table_name}")
        if result.get("success"):
            return {
                "database": database_name,
                "table": table_name,
                "columns": result.get("rows", [])
            }
        else:
            return {"error": result.get("error")}
    except Exception as e:
        return {"error": str(e)}

Langkah 4: Menambahkan Alat Manajemen Pekerjaan

Mari kita tambahkan alat untuk mengelola dan membuat kueri pekerjaan Databricks:

@mcp.tool()
async def list_jobs(limit: int = 100) -> List[Dict[str, Any]]:
    """Buat daftar pekerjaan di ruang kerja Databricks.

    Args:
        limit: Jumlah maksimum pekerjaan yang akan dikembalikan

    Returns:
        Daftar detail pekerjaan
    """
    try:
        jobs = workspace.jobs.list(limit=limit)
        result = []
        for job in jobs:
            result.append({
                "job_id": job.job_id,
                "name": job.settings.name,
                "created_by": job.created_by,
                "created_time": job.created_time
            })
        return result
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
async def get_job_runs(job_id: int, limit: int = 10) -> Dict[str, Any]:
    """Dapatkan eksekusi terbaru untuk pekerjaan tertentu.

    Args:
        job_id: ID pekerjaan
        limit: Jumlah maksimum eksekusi yang akan dikembalikan

    Returns:
        Dictionary yang berisi detail eksekusi pekerjaan
    """
    try:
        runs = workspace.jobs.list_runs(job_id=job_id, limit=limit)

        result = {
            "job_id": job_id,
            "runs": []
        }

        for run in runs.runs:
            run_info = {
                "run_id": run.run_id,
                "state": run.state.life_cycle_state,
                "result_state": run.state.result_state,
                "start_time": run.start_time,
                "setup_duration": run.setup_duration,
                "execution_duration": run.execution_duration,
                "cleanup_duration": run.cleanup_duration
            }
            result["runs"].append(run_info)

        return result
    except Exception as e:
        return {"error": str(e)}

Langkah 5: Mengimplementasikan Alat Manajemen Klaster

Sekarang mari kita tambahkan alat untuk berinteraksi dengan klaster Databricks:

@mcp.tool()
async def list_clusters() -> List[Dict[str, Any]]:
    """Buat daftar semua klaster di ruang kerja Databricks.

    Returns:
        Daftar detail klaster
    """
    try:
        clusters = workspace.clusters.list()
        result = []

        for cluster in clusters:
            cluster_info = {
                "cluster_id": cluster.cluster_id,
                "cluster_name": cluster.cluster_name,
                "state": cluster.state,
                "creator_username": cluster.creator_user_name,
                "spark_version": cluster.spark_version,
                "node_type_id": cluster.node_type_id
            }
            result.append(cluster_info)

        return result
    except Exception as e:
        return [{"error": str(e)}]

Langkah 6: Menjalankan MCP Server

Terakhir, tambahkan kode untuk menjalankan MCP server:

if __name__ == "__main__":
    # Jalankan server dengan transportasi stdio untuk kompatibilitas Claude Desktop
    # atau HTTP untuk klien berbasis web
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--http":
        # Jalankan sebagai server HTTP
        mcp.run(transport='http', host="127.0.0.1", port=8000)
    else:
        # Default ke stdio untuk Claude Desktop
        mcp.run(transport='stdio')

Langkah 7: Mengonfigurasi Klien MCP

Untuk menggunakan Databricks MCP Server dengan klien MCP seperti Claude Desktop, Anda perlu membuat konfigurasi:

{
  "mcpServers": {
    "databricks": {
      "command": "python",
      "args": ["databricks_mcp.py"]
    }
  }
}

Jika Anda lebih suka menggunakan transportasi HTTP, Anda dapat menjalankan server dalam proses terpisah dan mengonfigurasi klien yang sesuai:

{
  "mcpServers": {
    "databricks": {
      "url": "<http://localhost:8000>"
    }
  }
}

Langkah 8: Membangun Aplikasi Klien (Opsional)

Untuk pengalaman yang lebih disesuaikan, Anda dapat membangun aplikasi klien yang memanfaatkan Databricks MCP Server. Berikut adalah contoh klien berbasis Python yang terhubung ke server MCP dan model Meta Llama Databricks:

import asyncio
from contextlib import AsyncExitStack
from typing import Optional, Dict, Any

from databricks.sdk import WorkspaceClient
from mcp.client import ClientSession, stdio_client
from mcp.stdio import StdioServerParameters

class DatabricksMCPClient:
    def __init__(self, llm_endpoint_name: str = "databricks-llama3-70b-instruct"):
        # Inisialisasi objek sesi dan klien
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

        # Inisialisasi klien Databricks
        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}")

    async def connect_to_server(self, server_script_path: str):
        """Terhubung ke server Databricks MCP"""
        server_params = StdioServerParameters(
            command="python",
            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()

        # Buat daftar alat yang tersedia
        response = await self.session.list_tools()
        tools = response.tools
        print("\\\\nTerhubung ke server Databricks MCP dengan alat:", [tool.name for tool in tools])

        # Tampilkan deskripsi alat
        print("\\\\nAlat Databricks yang Tersedia:")
        for tool in tools:
            print(f"- {tool.name}: {tool.description}")

    # Tambahkan metode tambahan untuk memproses kueri, dll.

async def main():
    client = DatabricksMCPClient()
    try:
        await client.connect_to_server("databricks_mcp.py")
        # Tambahkan loop interaktif atau fungsionalitas lainnya
    finally:
        await client.exit_stack.aclose()

if __name__ == "__main__":
    asyncio.run(main())

Praktik Terbaik untuk Menggunakan Databricks MCP Server

Keamanan Utama: Selalu terapkan kontrol akses dan validasi kueri SQL untuk mencegah operasi yang merusak.
Penanganan Kesalahan: Implementasikan penanganan kesalahan yang kuat di aplikasi server dan klien.
Pembatasan Laju: Tambahkan pembatasan laju untuk mencegah panggilan API yang berlebihan ke Databricks.
Pencatatan: Implementasikan pencatatan komprehensif untuk tujuan debugging dan audit.
Pagination: Untuk set hasil yang besar, implementasikan pagination untuk menghindari masalah memori.
Kredensial Aman: Gunakan variabel lingkungan atau penyimpanan kredensial aman daripada hardcoding kredensial.
Dokumentasi Alat: Tulis deskripsi yang jelas dan terperinci untuk setiap alat untuk membantu LLM menggunakannya dengan benar.

Fitur Tingkat Lanjut

Menambahkan Dukungan untuk Delta Live Tables

@mcp.tool()
async def list_delta_pipelines() -> List[Dict[str, Any]]:
    """Buat daftar semua pipeline Delta Live Table di ruang kerja.

    Returns:
        Daftar detail pipeline
    """
    try:
        pipelines = workspace.pipelines.list()
        result = []

        for pipeline in pipelines:
            pipeline_info = {
                "pipeline_id": pipeline.pipeline_id,
                "name": pipeline.name,
                "state": pipeline.state,
                "creator": pipeline.creator,
                "target": pipeline.target
            }
            result.append(pipeline_info)

        return result
    except Exception as e:
        return [{"error": str(e)}]

Menambahkan Manajemen Notebook

@mcp.tool()
async def list_notebooks(path: str = "/") -> List[Dict[str, Any]]:
    """Buat daftar notebook di jalur yang ditentukan di ruang kerja.

    Args:
        path: Jalur ruang kerja untuk membuat daftar notebook dari

    Returns:
        Daftar detail notebook
    """
    try:
        objects = workspace.workspace.list(path)
        notebooks = []

        for obj in objects:
            if obj.object_type == "NOTEBOOK":
                notebooks.append({
                    "path": obj.path,
                    "name": obj.path.split("/")[-1],
                    "language": obj.language
                })

        return notebooks
    except Exception as e:
        return [{"error": str(e)}]

Kesimpulan

Databricks MCP Server menyediakan cara yang ampuh untuk menjembatani kesenjangan antara model AI dan infrastruktur data organisasi Anda. Dengan mengikuti panduan komprehensif ini, Anda telah mempelajari cara:

Menyiapkan dan mengonfigurasi Databricks MCP Server
Mengimplementasikan alat inti untuk eksekusi kueri SQL dan eksplorasi database
Menambahkan kemampuan manajemen pekerjaan dan klaster
Mengonfigurasi klien MCP untuk berinteraksi dengan server Anda
Secara opsional membangun aplikasi klien khusus

Integrasi ini memberdayakan asisten AI Anda untuk secara langsung memanfaatkan kemampuan pemrosesan data Databricks yang kuat, memungkinkan analisis data, pemantauan, dan otomatisasi yang lebih canggih. Saat ekosistem Model Context Protocol terus berkembang, Databricks MCP Server Anda dapat diperluas dengan alat dan kemampuan tambahan untuk memenuhi kebutuhan spesifik organisasi Anda.

Dengan merangkul integrasi ini, Anda mengambil langkah signifikan menuju pembuatan sistem AI yang lebih cerdas dan sadar data yang dapat memberikan wawasan lebih dalam dan bantuan yang lebih efektif untuk alur kerja analitik data Anda.