Databricks 관리형 MCP 서버 사용

Important

이 기능은 베타 버전으로 제공됩니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. Azure Databricks 미리 보기 관리를 참조하세요.

Databricks 관리형 MCP 서버는 AI 에이전트를 Unity 카탈로그, Databricks Vector Search 인덱스, 지니 공간 및 사용자 지정 함수에 저장된 데이터에 연결하는 즉시 사용할 수 있는 서버입니다.

사용 가능한 관리되는 서버

MCP 서버 및 해당 엔드포인트 URL을 보려면 작업 영역 >에이전트>MCP 서버로 이동합니다.

에이전트 MCP 서버 탭

Databricks는 기본 제공되는 다음의 MCP 서버를 제공합니다.

MCP 서버	Description	URL 패턴
벡터 검색	벡터 검색 인덱스를 쿼리하여 관련 문서를 찾습니다. Databricks 관리 임베딩이 있는 인덱스만 지원됩니다.	`https://<workspace-hostname>/api/2.0/mcp/vector-search/{catalog}/{schema}`
지니 공간	지니 공간을 쿼리하여 자연어를 사용하여 구조화된 데이터를 분석합니다. 읽기 전용 작업을 위해 Claude 또는 ChatGPT와 같은 외부 AI 도우미를 데이터에 연결할 때 Genie MCP 서버를 사용합니다. 메모: Genie용 관리형 MCP 서버는 Genie를 MCP 도구로 호출합니다. 즉, Genie API를 호출할 때 기록이 전달되지 않습니다. 또는 다중 에이전트 시스템에서 Genie를 사용할 수 있습니다.	`https://<workspace-hostname>/api/2.0/mcp/genie/{genie_space_id}`
Unity 카탈로그 함수	Unity 카탈로그 함수를 사용하여 미리 정의된 SQL 쿼리를 실행합니다.	`https://<workspace-hostname>/api/2.0/mcp/functions/{catalog}/{schema}`
DBSQL	AI에서 생성된 SQL을 실행합니다. DBSQL MCP 서버를 사용하여 AI 코딩 도구(Claude Code, Cursor, Codex 등)를 사용하여 데이터 파이프라인을 작성합니다. 읽기 전용 데이터 검색 및 챗봇 통합의 경우 Genie를 대신 사용합니다.	`https://<workspace-hostname>/api/2.0/mcp/sql`

예제 시나리오

고객 지원에 도움이 되는 사용자 지정 에이전트를 고려합니다. 여러 관리형 MCP 서버에 연결할 수 있습니다.

벡터 검색: https://<workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
- 지원 티켓 및 설명서 검색
지니 공간: https://<workspace-hostname>/api/2.0/mcp/genie/{billing_space_id}
- 청구 데이터 및 고객 정보 쿼리
UC 함수: https://<workspace-hostname>/api/2.0/mcp/functions/prod/billing
- 계정 조회 및 업데이트에 대한 사용자 지정 함수 실행

이렇게 하면 에이전트가 구조화되지 않은 데이터(지원 티켓), 구조적 데이터(청구 테이블) 및 사용자 지정 비즈니스 논리에 액세스할 수 있습니다.

Notebook 예제: Databricks MCP 서버를 통해 에이전트를 구축하기

다음 Notebook에서는 Databricks 관리형 MCP 서버를 사용하여 MCP 도구를 호출하는 LangGraph 및 OpenAI 에이전트를 작성하는 방법을 보여 줍니다.

로컬 환경에서 에이전트 빌드

Databricks에서 MCP 서버에 연결하는 것은 다른 원격 MCP 서버에 연결하는 것과 비슷합니다. MCP Python SDK와 같은 표준 SDK를 사용하여 서버에 연결할 수 있습니다. 주요 차이점은 Databricks MCP 서버는 기본적으로 안전하며 클라이언트가 인증을 지정하도록 요구한다는 것입니다.

databricks-mcp Python 라이브러리는 사용자 지정 에이전트 코드에서 인증을 간소화하는 데 도움이 됩니다.

에이전트 코드를 개발하는 가장 간단한 방법은 로컬로 실행하고 작업 영역에 인증하는 것입니다. 다음 단계에 따라 Databricks MCP 서버에 연결하는 AI 에이전트를 빌드합니다.

환경 설정

OAuth를 사용하여 작업 영역에 인증합니다. 로컬 터미널에서 다음 명령을 실행합니다.
```
databricks auth login --host https://<your-workspace-hostname>
```
메시지가 표시되면 프로필 이름을 생성하고 나중에 참고하십시오. 기본 프로필 이름은 DEFAULT입니다.

Python 3.12 이상이 있는 로컬 환경이 있는지 확인하고 종속성을 설치합니다.

pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"

로컬 환경 연결 테스트

이 코드 조각을 실행하려면 작업 영역에서 서버리스 컴퓨팅을 사용하도록 설정해야 합니다.

Unity 카탈로그 도구를 나열하고 기본 제공 Python 코드 인터프리터 도구를 실행하여 MCP 서버에 대한 연결의 유효성을 검사합니다.

다음 코드를 실행하여 MCP 서버에 대한 연결의 유효성을 검사합니다.

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

# TODO: Update to the Databricks CLI profile name you specified when
# configuring authentication to the workspace.
databricks_cli_profile = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    databricks_cli_profile != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set databricks_cli_profile to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=databricks_cli_profile)
workspace_hostname = workspace_client.config.host
mcp_server_url = f"{workspace_hostname}/api/2.0/mcp/functions/system/ai"

# This code uses the Unity Catalog functions MCP server to expose built-in
# AI tools under `system.ai`, like the `system.ai.python_exec` code interpreter tool
def test_connect_to_server():
    mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
    tools = mcp_client.list_tools()

    print(
        f"Discovered tools {[t.name for t in tools]} "
        f"from MCP server {mcp_server_url}"
    )

    result = mcp_client.call_tool(
        "system__ai__python_exec", {"code": "print('Hello, world!')"}
    )
    print(
        f"Called system__ai__python_exec tool and got result "
        f"{result.content}"
    )


if __name__ == "__main__":
    test_connect_to_server()

에이전트 만들기

도구를 사용하는 기본 단일 턴 에이전트를 정의하려면 위의 코드를 기반으로 합니다. 에이전트 코드를 다음과 같은 mcp_agent.py파일로 로컬로 저장합니다.

 import json
 import uuid
 import asyncio
 from typing import Any, Callable, List
 from pydantic import BaseModel

 import mlflow
 from mlflow.pyfunc import ResponsesAgent
 from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse

 from databricks_mcp import DatabricksMCPClient
 from databricks.sdk import WorkspaceClient

 # 1) CONFIGURE YOUR ENDPOINTS/PROFILE
 LLM_ENDPOINT_NAME = "databricks-claude-sonnet-4-5"
 SYSTEM_PROMPT = "You are a helpful assistant."
 DATABRICKS_CLI_PROFILE = "YOUR_DATABRICKS_CLI_PROFILE"
 assert (
     DATABRICKS_CLI_PROFILE != "YOUR_DATABRICKS_CLI_PROFILE"
 ), "Set DATABRICKS_CLI_PROFILE to the Databricks CLI profile name you specified when configuring authentication to the workspace"
 workspace_client = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)
 host = workspace_client.config.host
 # Add more MCP server URLs here if desired, for example:
 # f"{host}/api/2.0/mcp/vector-search/prod/billing"
 # to include vector search indexes under the prod.billing schema, or
 # f"{host}/api/2.0/mcp/genie/<genie_space_id>"
 # to include a Genie space
 MANAGED_MCP_SERVER_URLS = [
     f"{host}/api/2.0/mcp/functions/system/ai",
 ]
 # Add Custom MCP Servers hosted on Databricks Apps
 CUSTOM_MCP_SERVER_URLS = []



 # 2) HELPER: convert between ResponsesAgent “message dict” and ChatCompletions format
 def _to_chat_messages(msg: dict[str, Any]) -> List[dict]:
     """
     Take a single ResponsesAgent‐style dict and turn it into one or more
     ChatCompletions‐compatible dict entries.
     """
     msg_type = msg.get("type")
     if msg_type == "function_call":
         return [
             {
                 "role": "assistant",
                 "content": None,
                 "tool_calls": [
                     {
                         "id": msg["call_id"],
                         "type": "function",
                         "function": {
                             "name": msg["name"],
                             "arguments": msg["arguments"],
                         },
                     }
                 ],
             }
         ]
     elif msg_type == "message" and isinstance(msg["content"], list):
         return [
             {
                 "role": "assistant" if msg["role"] == "assistant" else msg["role"],
                 "content": content["text"],
             }
             for content in msg["content"]
         ]
     elif msg_type == "function_call_output":
         return [
             {
                 "role": "tool",
                 "content": msg["output"],
                 "tool_call_id": msg["tool_call_id"],
             }
         ]
     else:
         # fallback for plain {"role": ..., "content": "..."} or similar
         return [
             {
                 k: v
                 for k, v in msg.items()
                 if k in ("role", "content", "name", "tool_calls", "tool_call_id")
             }
         ]


 # 3) “MCP SESSION” + TOOL‐INVOCATION LOGIC
 def _make_exec_fn(
     server_url: str, tool_name: str, ws: WorkspaceClient
 ) -> Callable[..., str]:
     def exec_fn(**kwargs):
         mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
         response = mcp_client.call_tool(tool_name, kwargs)
         return "".join([c.text for c in response.content])

     return exec_fn


 class ToolInfo(BaseModel):
     name: str
     spec: dict
     exec_fn: Callable


 def _fetch_tool_infos(ws: WorkspaceClient, server_url: str) -> List[ToolInfo]:
     print(f"Listing tools from MCP server {server_url}")
     infos: List[ToolInfo] = []
     mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
     mcp_tools = mcp_client.list_tools()
     for t in mcp_tools:
         schema = t.inputSchema.copy()
         if "properties" not in schema:
             schema["properties"] = {}
         spec = {
             "type": "function",
             "function": {
                 "name": t.name,
                 "description": t.description,
                 "parameters": schema,
             },
         }
         infos.append(
             ToolInfo(
                 name=t.name, spec=spec, exec_fn=_make_exec_fn(server_url, t.name, ws)
             )
         )
     return infos


 # 4) “SINGLE‐TURN” AGENT CLASS
 class SingleTurnMCPAgent(ResponsesAgent):
     def _call_llm(self, history: List[dict], ws: WorkspaceClient, tool_infos):
         """
         Send current history → LLM, returning the raw response dict.
         """
         client = ws.serving_endpoints.get_open_ai_client()
         flat_msgs = []
         for msg in history:
             flat_msgs.extend(_to_chat_messages(msg))
         return client.chat.completions.create(
             model=LLM_ENDPOINT_NAME,
             messages=flat_msgs,
             tools=[ti.spec for ti in tool_infos],
         )

     def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
         ws = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)

         # 1) build initial history: system + user
         history: List[dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
         for inp in request.input:
             history.append(inp.model_dump())

         # 2) call LLM once
         tool_infos = [
             tool_info
             for mcp_server_url in (MANAGED_MCP_SERVER_URLS + CUSTOM_MCP_SERVER_URLS)
             for tool_info in _fetch_tool_infos(ws, mcp_server_url)
         ]
         tools_dict = {tool_info.name: tool_info for tool_info in tool_infos}
         llm_resp = self._call_llm(history, ws, tool_infos)
         raw_choice = llm_resp.choices[0].message.to_dict()
         raw_choice["id"] = uuid.uuid4().hex
         history.append(raw_choice)

         tool_calls = raw_choice.get("tool_calls") or []
         if tool_calls:
             # (we only support a single tool in this “single‐turn” example)
             fc = tool_calls[0]
             name = fc["function"]["name"]
             args = json.loads(fc["function"]["arguments"])
             try:
                 tool_info = tools_dict[name]
                 result = tool_info.exec_fn(**args)
             except Exception as e:
                 result = f"Error invoking {name}: {e}"

             # 4) append the “tool” output
             history.append(
                 {
                     "type": "function_call_output",
                     "role": "tool",
                     "id": uuid.uuid4().hex,
                     "tool_call_id": fc["id"],
                     "output": result,
                 }
             )

             # 5) call LLM a second time and treat that reply as final
             followup = (
                 self._call_llm(history, ws, tool_infos=[]).choices[0].message.to_dict()
             )
             followup["id"] = uuid.uuid4().hex

             assistant_text = followup.get("content", "")
             return ResponsesAgentResponse(
                 output=[
                     {
                         "id": uuid.uuid4().hex,
                         "type": "message",
                         "role": "assistant",
                         "content": [{"type": "output_text", "text": assistant_text}],
                     }
                 ],
                 custom_outputs=request.custom_inputs,
             )

         # 6) if no tool_calls at all, return the assistant’s original reply
         assistant_text = raw_choice.get("content", "")
         return ResponsesAgentResponse(
             output=[
                 {
                     "id": uuid.uuid4().hex,
                     "type": "message",
                     "role": "assistant",
                     "content": [{"type": "output_text", "text": assistant_text}],
                 }
             ],
             custom_outputs=request.custom_inputs,
         )


 mlflow.models.set_model(SingleTurnMCPAgent())

 if __name__ == "__main__":
     req = ResponsesAgentRequest(
         input=[{"role": "user", "content": "What's the 100th Fibonacci number?"}]
     )
     resp = SingleTurnMCPAgent().predict(req)
     for item in resp.output:
         print(item)

에이전트를 배포하세요

관리형 MCP 서버에 연결하는 에이전트를 배포할 준비가 되면 생성 AI 애플리케이션에 대한 에이전트 배포를 참조하세요.

로깅 시 에이전트에 필요한 모든 리소스를 지정합니다. Databricks 리소스에 대한 인증 참조

예를 들어, 에이전트가 아래에 나열된 MCP 서버 URL을 사용하는 경우, prod.customer_support 및 prod.billing 스키마에서 모든 벡터 검색 인덱스를 지정해야 합니다. 또한 다음에서 모든 Unity 카탈로그 함수를 지정해야 합니다.prod.billing

https://<your-workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
https://<your-workspace-hostname>/api/2.0/mcp/vector-search/prod/billing
https://<your-workspace-hostname>/api/2.0/mcp/functions/prod/billing

에이전트가 Databricks의 MCP 서버에 연결하여 도구를 검색하고 실행하는 경우 이러한 MCP 서버에 필요한 리소스를 에이전트와 함께 기록합니다. Databricks는 이 프로세스를 간소화하기 위해 PyPI 패키지를 설치하는 databricks-mcp 것이 좋습니다.

특히 관리형 MCP 서버를 사용하는 경우 관리되는 MCP 서버에 필요한 리소스를 검색하는 데 사용할 databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) 수 있습니다. 에이전트가 Databricks 앱에서 호스트되는 사용자 지정 MCP 서버를 쿼리하는 경우 모델을 로깅할 때 서버를 리소스로 명시적으로 포함하여 권한 부여를 구성할 수 있습니다.

예를 들어 위에서 정의한 에이전트를 배포하려면 에이전트 코드 정의를 저장했다고 가정하고 다음 코드를 실행합니다.mcp_agent.py

import os
from databricks.sdk import WorkspaceClient
from databricks import agents
import mlflow
from mlflow.models.resources import DatabricksFunction, DatabricksServingEndpoint, DatabricksVectorSearchIndex
from mcp_agent import LLM_ENDPOINT_NAME
from databricks_mcp import DatabricksMCPClient

# TODO: Update this to your Databricks CLI profile name
databricks_cli_profile = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    databricks_cli_profile != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set databricks_cli_profile to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=databricks_cli_profile)


# Configure MLflow and the Databricks SDK to use your Databricks CLI profile
current_user = workspace_client.current_user.me().user_name
mlflow.set_tracking_uri(f"databricks://{databricks_cli_profile}")
mlflow.set_registry_uri(f"databricks-uc://{databricks_cli_profile}")
mlflow.set_experiment(f"/Users/{current_user}/databricks_docs_example_mcp_agent")
os.environ["DATABRICKS_CONFIG_PROFILE"] = databricks_cli_profile

MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]
# Log the agent defined in mcp_agent.py
here = os.path.dirname(os.path.abspath(__file__))
agent_script = os.path.join(here, "mcp_agent.py")
resources = [
    DatabricksServingEndpoint(endpoint_name=LLM_ENDPOINT_NAME),
    DatabricksFunction("system.ai.python_exec"),
    # --- Uncomment and edit the following lines to include custom mcp servers hosted on Databricks Apps ---
    # DatabricksApp(app_name="app-name")
]

for mcp_server_url in MANAGED_MCP_SERVER_URLS:
    mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
    resources.extend(mcp_client.get_databricks_resources())

with mlflow.start_run():
    logged_model_info = mlflow.pyfunc.log_model(
        artifact_path="mcp_agent",
        python_model=agent_script,
        resources=resources,
    )

# TODO Specify your UC model name here
UC_MODEL_NAME = "main.default.databricks_docs_mcp_agent"
registered_model = mlflow.register_model(logged_model_info.model_uri, UC_MODEL_NAME)

agents.deploy(
    model_name=UC_MODEL_NAME,
    model_version=registered_model.version,
)

다음 단계

커서 및 Claude Desktop과 같은 MCP 서버에 클라이언트를 연결합니다.
Databricks Vector Search MCP 서버에 대한 메타 매개 변수

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-01-09

다음을 통해 공유