为copula构建自定义MCP服务器

一本全面的指南，指导您如何使用FastAPI和Databricks SDK创建与ViewModel集成的生产就绪MCP（模型上下文协议）服务器。

介绍

本指南将引导您创建自定义MCP服务器，使Claude等AI代理能够与您的copula工作区进行交互。您将学习如何：

设置FastAPI+FastMCP应用程序
创建调用copula API的工具
将您的服务器部署到copula Apps
将其连接到Claude CLI，以实现AI驱动的工作空间管理

在本教程结束时，您将拥有一个可工作的MCP服务器，该服务器可以列出集群、执行SQL查询和运行作业——所有这些都可以通过自然语言命令发送到AI代理。

______________________________________________________________________

什么是MCP？

模型上下文协议（MCP） 是一个将AI代理与工具、数据源和上下文信息连接起来的开放标准。将其视为一个通用适配器，让像Claude这样的人工智能助手以标准化的方式与您的服务进行交互。

关键概念

工具：AI代理可以调用的函数

@mcp_server.tool
def list_clusters() -> dict:
    """List all Databricks clusters"""
    # Returns cluster information

提示：AI代理的可重复使用说明

# Analyze Cluster Performance
You are an expert at analyzing Databricks clusters...

运输：代理如何与您的服务器通信

标准：标准输入/输出（适用于Claude Desktop等CLI工具）
HTTP/SSE：基于Web的通信（用于部署的应用程序）

MCP的工作原理

┌──────────────┐                ┌──────────────┐                ┌──────────────┐
│  AI Agent    │                │  MCP Server  │                │  Databricks  │
│  (Claude)    │                │  (Your Code) │                │  Workspace   │
└──────┬───────┘                └──────┬───────┘                └──────┬───────┘
       │                               │                               │
       │  1. "List my clusters"        │                               │
       ├──────────────────────────────>│                               │
       │                               │                               │
       │                               │  2. Call Databricks API       │
       │                               ├──────────────────────────────>│
       │                               │                               │
       │                               │  3. Return cluster data       │
       │                               │ │  MCP Server  │
└──────────────┘                 └──────┬───────┘
                                        │
                                        │ API calls
                                        │
                                 ┌──────▼───────┐
                                 │  Databricks  │
                                 └──────────────┘

特性:

部署简单
与桌面AI应用程序直接集成
PAT或CLI配置文件身份验证
非常适合个人生产力

比较

功能	FastAPI+FastMCP	stdio MCP
Web用户界面	✅ 是	❌ 没有
生产就绪	✅ 是	⚠️ 个人使用
OAuth支持	✅ 是	❌ 没有
复杂性	中等	低
最适合	团队、生产	个人、CLI

本指南重点介绍 FastAPI+FastMCP 因为这是最通用的选择。

______________________________________________________________________

入门指南

步骤1：环境设置

# Create project directory
mkdir my-databricks-mcp
cd my-databricks-mcp

# Initialize uv project
uv init

# Install core dependencies
uv add fastapi uvicorn databricks-sdk fastmcp mcp pyyaml python-dotenv

# Install development dependencies
uv add --dev pytest pytest-asyncio ruff

第二步：使用copula进行身份验证

# Configure Databricks CLI
databricks configure --token

# Enter your workspace URL and token when prompted
# Workspace URL: https://your-workspace.cloud.databricks.com
# Token: dapi...

# Test authentication
databricks current-user me

步骤3：创建环境文件

# Create .env.local
cat > .env.local  WorkspaceClient:
    """Get authenticated Databricks workspace client.

    Uses environment variables:
    - DATABRICKS_HOST: Workspace URL
    - DATABRICKS_TOKEN: Personal access token or auto-injected by Databricks Apps
    """
    return WorkspaceClient(
        host=os.environ.get('DATABRICKS_HOST'),
        token=os.environ.get('DATABRICKS_TOKEN')
    )

def verify_authentication() -> dict:
    """Verify Databricks authentication is working."""
    try:
        client = get_workspace_client()
        user = client.current_user.me()
        return {
            'success': True,
            'user_name': user.user_name,
            'user_id': user.id,
            'workspace_url': client.config.host
        }
    except DatabricksError as e:
        return {
            'success': False,
            'error': f'Authentication failed: {str(e)}'
        }

第三步：创建你的第一个工具

# server/tools.py
import os
from databricks.sdk.errors import DatabricksError, NotFound, PermissionDenied
from server.services.databricks_client import get_workspace_client

def load_tools(mcp_server):
    """Register all MCP tools with the server."""

    @mcp_server.tool
    def list_clusters() -> dict:
        """List all Databricks clusters in the workspace.

        Returns comprehensive cluster information including:
        - Cluster ID and name
        - Current state (RUNNING, TERMINATED, etc.)
        - Size (number of workers)
        - Spark version and node types
        - Creator and creation time

        Returns:
            Dictionary with success status and list of clusters
        """
        try:
            client = get_workspace_client()
            clusters = []

            for cluster in client.clusters.list():
                clusters.append({
                    'cluster_id': cluster.cluster_id,
                    'cluster_name': cluster.cluster_name,
                    'state': cluster.state.value if cluster.state else 'UNKNOWN',
                    'num_workers': cluster.num_workers,
                    'spark_version': cluster.spark_version,
                    'node_type_id': cluster.node_type_id,
                    'creator_user_name': cluster.creator_user_name,
                    'start_time': cluster.start_time,
                })

            return {
                'success': True,
                'clusters': clusters,
                'count': len(clusters)
            }
        except PermissionDenied:
            return {
                'success': False,
                'error': 'Permission denied to list clusters',
                'error_code': 'PERMISSION_DENIED',
                'suggestion': 'Ensure you have cluster read permissions in your workspace'
            }
        except DatabricksError as e:
            return {
                'success': False,
                'error': str(e),
                'error_code': 'DATABRICKS_ERROR'
            }
        except Exception as e:
            return {
                'success': False,
                'error': f'Unexpected error: {str(e)}',
                'error_code': 'INTERNAL_ERROR'
            }

    @mcp_server.tool
    def execute_sql(
        query: str,
        warehouse_id: str = None,
        catalog: str = None,
        schema: str = None
    ) -> dict:
        """Execute a SQL query on Databricks SQL warehouse.

        Runs the provided SQL statement and returns results in a structured format.
        Automatically uses default warehouse from environment if not specified.

        Args:
            query: SQL statement to execute (SELECT, CREATE, INSERT, etc.)
            warehouse_id: SQL warehouse ID (optional, uses DATABRICKS_WAREHOUSE_ID env var if not provided)
            catalog: Unity Catalog catalog name (optional, e.g., 'main')
            schema: Schema/database name (optional, e.g., 'default')

        Returns:
            Dictionary with query results including:
            - columns: List of column names
            - rows: List of row dictionaries
            - row_count: Number of rows returned

        Example:
            execute_sql("SELECT * FROM my_table LIMIT 10")
        """
        try:
            client = get_workspace_client()

            # Use environment variable as fallback
            if not warehouse_id:
                warehouse_id = os.environ.get('DATABRICKS_WAREHOUSE_ID')
                if not warehouse_id:
                    return {
                        'success': False,
                        'error': 'No warehouse_id provided and DATABRICKS_WAREHOUSE_ID not set',
                        'error_code': 'MISSING_WAREHOUSE_ID',
                        'suggestion': 'Provide warehouse_id or set DATABRICKS_WAREHOUSE_ID environment variable'
                    }

            # Execute statement
            response = client.statement_execution.execute_statement(
                warehouse_id=warehouse_id,
                statement=query,
                catalog=catalog,
                schema=schema,
                wait_timeout='30s'
            )

            # Parse results
            if response.result and response.result.data_array:
                columns = [col.name for col in response.manifest.schema.columns]
                rows = []
                for row_data in response.result.data_array:
                    row_dict = {col: row_data[i] for i, col in enumerate(columns)}
                    rows.append(row_dict)

                return {
                    'success': True,
                    'columns': columns,
                    'rows': rows,
                    'row_count': len(rows),
                    'statement_id': response.statement_id
                }
            else:
                return {
                    'success': True,
                    'message': 'Query executed successfully (no results returned)',
                    'statement_id': response.statement_id
                }

        except DatabricksError as e:
            return {
                'success': False,
                'error': str(e),
                'error_code': 'DATABRICKS_ERROR',
                'query': query
            }
        except Exception as e:
            return {
                'success': False,
                'error': f'Unexpected error: {str(e)}',
                'error_code': 'INTERNAL_ERROR',
                'query': query
            }

步骤4：创建提示加载器

# server/prompts.py
import glob
import os

def load_prompts(mcp_server):
    """Dynamically load prompts from the prompts directory.

    Each .md file in prompts/ becomes an MCP prompt with:
    - Name: filename without extension
    - Description: First line of file (without # prefix)
    """
    prompt_dir = 'prompts'

    if not os.path.exists(prompt_dir):
        print(f"Warning: {prompt_dir} directory not found")
        return

    prompt_files = glob.glob(f'{prompt_dir}/*.md')

    for prompt_file in prompt_files:
        # Extract prompt name from filename
        prompt_name = os.path.splitext(os.path.basename(prompt_file))[0]

        # Read prompt content
        with open(prompt_file, 'r', encoding='utf-8') as f:
            content = f.read()

        # Extract title from first line
        lines = content.strip().split('\n')
        title = lines[0].strip().lstrip('#').strip() if lines else prompt_name

        # Create closure to capture values properly
        def make_prompt_handler(prompt_content, name, desc):
            @mcp_server.prompt(name=name, description=desc)
            async def handle_prompt():
                return prompt_content
            return handle_prompt

        # Register prompt
        make_prompt_handler(content, prompt_name, title)
        print(f"Loaded prompt: {prompt_name}")

步骤5：创建示例提示

# Create prompts/analyze_cluster.md
cat > prompts/analyze_cluster.md  config.yaml =3.11"
dependencies = [
    "fastapi>=0.104.1",
    "uvicorn[standard]>=0.24.0",
    "databricks-sdk==0.59.0",
    "fastmcp>=0.2.0",
    "mcp>=1.12.0",
    "pyyaml>=6.0.2",
    "python-dotenv>=1.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "pytest-asyncio>=0.21.0",
    "ruff>=0.1.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 100
target-version = "py311"

app.yaml（用于copula应用程序部署）

command: ["python", "-m", "server.app"]

source_code_path: "."

environment:
  # Automatically injected by Databricks Apps
  - name: DATABRICKS_HOST
    value_from: workspace

  - name: DATABRICKS_TOKEN
    value_from: pat

  - name: DATABRICKS_APP_PORT
    value_from: app_port

  # Custom environment variables
  - name: LOG_LEVEL
    value: INFO

微笑。

cat > .gitignore  scripts/watch.sh  scripts/fix.sh  scripts/test.sh  claude_scripts/inspect_local_mcp.sh  claude_scripts/test_local_mcp_curl.sh  scripts/deploy.sh  claude_scripts/test_remote_mcp_curl.sh  dict:
    """Trigger a Databricks job run."""

@mcp_server.tool
def get_job_run_status(run_id: str) -> dict:
    """Get the status of a job run."""

统一目录:

@mcp_server.tool
def list_catalogs() -> dict:
    """List all Unity Catalog catalogs."""

@mcp_server.tool
def create_catalog(name: str, comment: str = None) -> dict:
    """Create a new Unity Catalog catalog."""

笔记本操作:

@mcp_server.tool
def list_notebooks(path: str) -> dict:
    """List notebooks in a workspace directory."""

@mcp_server.tool
def export_notebook(path: str, format: str = "SOURCE") -> dict:
    """Export a notebook in specified format."""

添加Web UI

创建React或Vue前端：

# Create React app
cd client
npm create vite@latest . -- --template react-ts
npm install

# Update package.json scripts
"scripts": {
  "dev": "vite",
  "build": "vite build",
  "preview": "vite preview"
}

# Start development
npm run dev

添加测试

创建全面的测试套件：

# tests/test_tools.py
import pytest
from unittest.mock import Mock, patch
from server.tools import load_tools
from fastmcp import FastMCP

@pytest.fixture
def mcp_server():
    server = FastMCP(name="test")
    load_tools(server)
    return server

def test_list_clusters_success(mcp_server):
    """Test successful cluster listing."""
    with patch('server.tools.get_workspace_client') as mock_client:
        # Setup mock
        mock_cluster = Mock()
        mock_cluster.cluster_id = 'test-123'
        mock_cluster.cluster_name = 'test-cluster'
        mock_cluster.state.value = 'RUNNING'

        mock_client.return_value.clusters.list.return_value = [mock_cluster]

        # Call tool
        tool = mcp_server._tools['list_clusters']
        result = tool.func()

        # Assertions
        assert result['success'] is True
        assert len(result['clusters']) == 1

优化生产

添加缓存:

from functools import lru_cache
from datetime import datetime, timedelta

_cache_time = None
_cache_data = None

@mcp_server.tool
def list_clusters_cached() -> dict:
    """List clusters with 5-minute cache."""
    global _cache_time, _cache_data

    now = datetime.now()
    if _cache_data and _cache_time and (now - _cache_time)  dict:
    """List clusters with logging."""
    logger.info("Listing clusters...")
    try:
        result = # ... cluster listing logic
        logger.info(f"Found {len(result['clusters'])} clusters")
        return result
    except Exception as e:
        logger.error(f"Failed to list clusters: {e}")
        raise

______________________________________________________________________

资源

官方文件

MCP协议: https://modelcontextprotocol.io/
Databricks SDK: https://databricks-sdk-py.readthedocs.io/
快速API: https://fastapi.tiangolo.com/
FastMCP: https://github.com/jlowin/fastmcp
copula应用程序: https://docs.databricks.com/en/dev-tools/databricks-apps/

示例项目

\_CHMCP示例: https://github.com/databricks/databricks-mcp-examples
自定义MCP模板：基于本指南中的项目

社区

MCP故障: https://discord.gg/mcp
copula社区: https://community.databricks.com/

工具

MCP检查员: npx @modelcontextprotocol/inspector
copula命令行界面: pip install databricks-cli
uv包管理器: https://github.com/astral-sh/uv

______________________________________________________________________

结论

现在，您已经为copula构建了一个生产就绪的MCP服务器！你可以：

✅ 创建调用copula API的工具 ✅ 从markdown文件加载提示 ✅ 部署到copula应用程序 ✅ 使用OAuth身份验证连接到Claude CLI

后续步骤:

为您的特定用例添加更多工具
为您的工作流创建自定义提示
为非人工智能用户构建web UI
与您的团队共享您的MCP服务器

快乐建筑！ 🚀

Databricks MCP Guide

快速接入

详细介绍

为copula构建自定义MCP服务器

目录

介绍

什么是MCP？

关键概念

MCP的工作原理

比较

入门指南

步骤1：环境设置

第二步：使用copula进行身份验证

步骤3：创建环境文件

第三步：创建你的第一个工具

步骤4：创建提示加载器

步骤5：创建示例提示

app.yaml（用于copula应用程序部署）

微笑。

添加Web UI

添加测试

优化生产

资源

官方文件

示例项目

社区

工具

结论

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP