💡
Databricks MCP 서버에 뛰어들기 전에 Apidog를 확인해 보세요. 이는 API 테스트 및 통합을 간소화하기 위해 설계된 무료 도구입니다. Apidog의 직관적인 인터페이스를 통해 API 워크플로를 쉽게 디버깅하고 최적화할 수 있어 개발 프로세스를 간소화하고 귀중한 시간을 절약할 수 있습니다. API를 구축하든 문제를 해결하든 Apidog는 워크플로를 향상시키는 데 필요한 모든 것을 갖추고 있습니다.
모델 컨텍스트 프로토콜(MCP)은 응용 프로그램이 대형 언어 모델(LLM)에 컨텍스트를 제공하는 방법을 표준화하기 위해 Anthropic에 의해 처음 개발된 개방형 프로토콜입니다. Databricks MCP 서버는 Databricks의 강력한 데이터 분석 플랫폼과 AI 모델 간의 다리 역할을 하여 LLM이 프로그램 방식으로 Databricks 환경과 상호 작용할 수 있도록 합니다. 이 통합은 AI 도우미가 SQL 쿼리를 실행하고, 작업 실행의 세부 정보를 나열 및 가져오며, 직접적인 인간 개입 없이 Databricks 계정 내의 데이터에 접근할 수 있게 합니다.
본 포괄적인 가이드에서는 향상된 AI 기반 데이터 작업 및 분석을 위한 Databricks MCP 서버를 설정하고 구성하며 활용하는 과정을 안내합니다.
모델 컨텍스트 프로토콜(MCP) 이해하기
Databricks 특정 구현에 뛰어들기 전에 MCP가 무엇이며 왜 중요한지 이해하는 것이 중요합니다:
MCP는 양방향 통신을 포함하는 클라이언트/서버 아키텍처를 구현합니다:
- 서버는 LLM이 사용할 수 있는 도구를 호스팅합니다.
- 클라이언트는 사용자, LLM 및 도구 간의 대화를 조정합니다.
MCP가 가치 있는 주요 이점은 다음과 같습니다:
- 관심사의 분리: 도구 개발자는 기능을 독립적으로 생성 및 업데이트할 수 있습니다.
- 모듈성: 새로운 도구는 시스템 전체의 변경 없이 점진적으로 추가될 수 있습니다.
- 표준화: 일관된 인터페이스는 구성 요소 간의 통합 작업을 줄여줍니다.
- 확장성: 구성 요소는 수요에 따라 독립적으로 확장할 수 있습니다.
Databricks MCP 서버를 실행하기 위한 전제 조건
Databricks MCP 서버를 설정하기 전에 다음 사항이 준비되어 있는지 확인하세요:
- 적절한 권한이 있는 Databricks 계정
- 시스템에 설치된 Python 3.8 이상
- 인증이 구성된 Databricks CLI
- Databricks 작업 공간 개념에 대한 기본 이해
- SQL 및 Databricks 작업에 대한 친숙함
단계 1: 환경 설정
먼저 필수 패키지를 설치하여 환경을 설정해 보겠습니다:
# 가상 환경 생성
python -m venv databricks-mcp-env
source databricks-mcp-env/bin/activate # Windows의 경우: databricks-mcp-env\\\\Scripts\\\\activate
# 필요한 패키지 설치
pip install databricks-sdk mcp-server httpx fastapi uvicorn
단계 2: Databricks MCP 서버 생성
다음 코드 구조로 databricks_mcp.py라는 새 파일을 만듭니다:
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
# FastMCP 서버 초기화
mcp = FastMCP("databricks")
# Databricks 클라이언트 초기화
workspace = WorkspaceClient()
단계 3: Databricks를 위한 핵심 MCP 도구 구현
이제 LLM이 Databricks 환경과 상호 작용할 수 있도록 해주는 핵심 도구를 구현해 보겠습니다:
@mcp.tool()
async def run_sql_query(query: str, warehouse_id: Optional[str] = None) -> Dict[str, Any]:
"""Databricks SQL 창고에서 SQL 쿼리 실행.
Args:
query: 실행할 SQL 쿼리
warehouse_id: 사용할 SQL 창고의 선택적 ID
Returns:
쿼리 결과를 포함하는 딕셔너리
"""
# 보안을 위한 쿼리 유효성 검사
if not query or any(keyword in query.lower() for keyword in ["drop", "delete", "truncate", "alter"]):
return {"error": "파괴적인 SQL 작업은 허용되지 않습니다."}
try:
# 쿼리 실행
statement = workspace.statement_execution.execute_statement(
warehouse_id=warehouse_id,
statement=query,
wait_timeout=60
)
# 결과 처리
if statement.status.state == StatementState.SUCCEEDED:
# 결과를 더 유용한 형식으로 변환
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)}
다음으로, 사용 가능한 데이터베이스를 나열하는 도구를 추가하겠습니다:
@mcp.tool()
async def list_databases() -> List[str]:
"""Databricks 작업 공간에 있는 모든 사용 가능한 데이터베이스 나열.
Returns:
데이터베이스 이름 목록
"""
try:
# 모든 데이터베이스를 가져오는 쿼리 실행
result = await run_sql_query("SHOW DATABASES")
if result.get("success"):
# 결과에서 데이터베이스 이름 추출
return [row.get("databaseName") for row in result.get("rows", [])]
else:
return [f"데이터베이스 나열 중 오류: {result.get('error')}"]
except Exception as e:
return [f"오류: {str(e)}"]
이제 테이블 스키마를 가져오는 도구를 구현해 보겠습니다:
@mcp.tool()
async def get_table_schema(database_name: str, table_name: str) -> Dict[str, Any]:
"""특정 테이블의 스키마 가져오기.
Args:
database_name: 데이터베이스 이름
table_name: 테이블 이름
Returns:
테이블 스키마 정보를 포함하는 딕셔너리
"""
try:
# 테이블 스키마를 가져오기 위한 쿼리 실행
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)}
단계 4: 작업 관리 도구 추가
Databricks 작업을 관리하고 쿼리할 도구를 추가해 보겠습니다:
@mcp.tool()
async def list_jobs(limit: int = 100) -> List[Dict[str, Any]]:
"""Databricks 작업 공간에 있는 작업 나열.
Args:
limit: 반환할 최대 작업 수
Returns:
작업 세부 사항 목록
"""
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]:
"""특정 작업의 최근 실행 가져오기.
Args:
job_id: 작업 ID
limit: 반환할 최대 실행 수
Returns:
작업 실행 세부 사항을 포함하는 딕셔너리
"""
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)}
단계 5: 클러스터 관리 도구 구현
이제 Databricks 클러스터와 상호 작용할 도구를 추가해 보겠습니다:
@mcp.tool()
async def list_clusters() -> List[Dict[str, Any]]:
"""Databricks 작업 공간의 모든 클러스터 나열.
Returns:
클러스터 세부 사항 목록
"""
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)}]
단계 6: MCP 서버 실행
마지막으로, MCP 서버를 실행하는 코드를 추가합니다:
if __name__ == "__main__":
# Claude Desktop 호환성을 위한 stdio 전송으로 서버 실행
# 또는 웹 기반 클라이언트를 위한 HTTP로 실행
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--http":
# HTTP 서버로 실행
mcp.run(transport='http', host="127.0.0.1", port=8000)
else:
# Claude Desktop의 경우 stdio 기본값으로 설정
mcp.run(transport='stdio')
단계 7: MCP 클라이언트 구성
Claude Desktop과 같은 MCP 클라이언트와 함께 Databricks MCP 서버를 사용하려면 구성 파일을 생성해야 합니다:
{
"mcpServers": {
"databricks": {
"command": "python",
"args": ["databricks_mcp.py"]
}
}
}
HTTP 전송을 사용하려면, 서버를 별도의 프로세스에서 실행하고 클라이언트를 그에 맞게 구성할 수 있습니다:
{
"mcpServers": {
"databricks": {
"url": "<http://localhost:8000>"
}
}
}
단계 8: 클라이언트 애플리케이션 구축 (선택 사항)
좀 더 맞춤화된 경험을 위해 Databricks MCP 서버를 활용하는 클라이언트 애플리케이션을 구축할 수 있습니다. 다음은 MCP 서버와 Databricks의 Meta Llama 모델에 연결되는 Python 기반 클라이언트의 예입니다:
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"):
# 세션 및 클라이언트 객체 초기화
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
# Databricks 클라이언트 초기화
self.workspace = WorkspaceClient()
self.llm_endpoint_name = llm_endpoint_name
self.openai_client = self.workspace.serving_endpoints.get_open_ai_client()
print(f"{llm_endpoint_name}에 대한 OpenAI 호환 클라이언트 초기화 완료")
async def connect_to_server(self, server_script_path: str):
"""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()
# 사용 가능한 도구 목록
response = await self.session.list_tools()
tools = response.tools
print("\\\\nDatabricks MCP 서버에 도구 연결됨:", [tool.name for tool in tools])
# 도구 설명 표시
print("\\\\n사용 가능한 Databricks 도구:")
for tool in tools:
print(f"- {tool.name}: {tool.description}")
# 쿼리 처리 등을 위한 추가 메서드 추가
async def main():
client = DatabricksMCPClient()
try:
await client.connect_to_server("databricks_mcp.py")
# 대화형 루프 또는 기타 기능 추가
finally:
await client.exit_stack.aclose()
if __name__ == "__main__":
asyncio.run(main())
Databricks MCP 서버 사용을 위한 모범 사례
- 보안 우선: 항상 접근 제어를 적용하고 파괴적인 작업을 방지하기 위해 SQL 쿼리를 검증하세요.
- 오류 처리: 서버 및 클라이언트 애플리케이션 모두에서 체계적인 오류 처리를 구현하세요.
- 비율 제한: 과도한 API 호출을 방지하기 위해 비율 제한을 추가하세요.
- 로그 기록: 디버깅 및 감사 목적으로 포괄적인 로그 기록을 구현하세요.
- 페이지네이션: 대규모 결과 세트의 경우 메모리 문제를 피하기 위해 페이지네이션을 구현하세요.
- 보안 자격 증명: 자격 증명을 하드코딩하는 대신 환경 변수 또는 안전한 자격 증명 저장소를 사용하세요.
- 도구 문서화: 각 도구의 명확하고 구체적인 설명을 작성하여 LLM이 올바르게 사용할 수 있도록 도와주세요.
고급 기능
델타 라이브 테이블에 대한 지원 추가
@mcp.tool()
async def list_delta_pipelines() -> List[Dict[str, Any]]:
"""작업 공간의 모든 Delta Live Table 파이프라인 나열.
Returns:
파이프라인 세부 사항 목록
"""
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)}]
노트북 관리 추가
@mcp.tool()
async def list_notebooks(path: str = "/") -> List[Dict[str, Any]]:
"""작업 공간의 특정 경로에 있는 노트북 나열.
Args:
path: 노트북을 나열할 작업 공간 경로
Returns:
노트북 세부 사항 목록
"""
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)}]
결론
Databricks MCP 서버는 AI 모델과 귀사의 데이터 인프라 간의 격차를 해소할 수 있는 강력한 방법을 제공합니다. 이 포괄적인 가이드를 따르면 다음을 배울 수 있습니다:
- Databricks MCP 서버를 설정하고 구성합니다.
- SQL 쿼리 실행 및 데이터베이스 탐색을 위한 핵심 도구를 구현합니다.
- 작업 및 클러스터 관리 기능을 추가합니다.
- 서버와 상호 작용하는 MCP 클라이언트를 구성합니다.
- 선택적으로 사용자 지정 클라이언트 애플리케이션을 구축합니다.
이 통합은 AI 도우미가 Databricks의 강력한 데이터 처리 능력을 직접 활용할 수 있게 하여 보다 정교한 데이터 분석, 모니터링 및 자동화를 가능하게 합니다. 모델 컨텍스트 프로토콜 생태계가 계속 발전함에 따라 귀하의 Databricks MCP 서버는 귀사의 특정 요구 사항에 맞춰 추가 도구와 기능으로 확장될 수 있습니다.
이 통합을 수용함으로써 더 깊은 통찰력과 데이터 분석 워크플로를 위해 보다 효과적인 지원을 제공할 수 있는 보다 지능적이고 데이터 인식 AI 시스템을 구축하는 중요한 단계를 밟고 있습니다.