Databricks Custom MCP Server Example

MCP服务器-Hello World

一个简单的、生产就绪的模板，用于使用FastMCP和FastAPI构建模型上下文协议（MCP）服务器。该项目演示了如何创建AI助手可以发现和调用的自定义工具。

关键概念

• 工具：AI助手可以调用的可调用函数（例如，搜索数据库、处理数据、调用API）
• 服务器：通过HTTP上的MCP协议公开工具
• 客户端：发现和调用工具的应用程序（如Claude、AI助手）

特性

• ✅ 基于FastMCP的服务器，支持HTTP流媒体
• ✅ FastAPI集成用于其他REST端点
• ✅ 示例工具：健康检查、用户信息、算术运算和copula作业触发
• ✅ 生产就绪项目结构
• ✅ 已准备好部署copula应用程序
• ✅ 综合测试与集成测试

项目结构

mcp-server-hello-world/
├── server/
│   ├── app.py                    # FastAPI application and MCP server setup
│   ├── main.py                   # Entry point for running the server
│   ├── tools.py                  # MCP tool definitions
│   └── utils.py                  # Databricks authentication helpers
├── scripts/
│   └── dev/
│       ├── start_server.sh           # Start the MCP server locally
│       ├── query_remote.sh           # Interactive script for testing deployed app with OAuth
│       ├── query_remote.py           # Query MCP client (deployed app) with health and user auth
│       └── generate_oauth_token.py   # Generate OAuth tokens for Databricks
├── tests/
│   └── test_integration_server.py   # Integration tests for MCP server
├── pyproject.toml                # Project metadata and dependencies
├── requirements.txt              # Python dependencies (for pip)
├── app.yaml                      # Databricks Apps configuration
├── Claude.md                     # AI assistant context and documentation
└── README.md          

先决条件

• Python 3.11或更高版本
• 紫外线 （推荐）或pip

安装

选项1：使用紫外线（推荐）

# Install uv if you haven't already
# Install dependencies
uv sync

选项2：使用pip

# Create a virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

运行服务器

发展模式

# Quick start with script (syncs dependencies and starts server)
./scripts/dev/start_server.sh

# Or manually using uv (default port 8000)
uv run custom-mcp-server

# Or specify a custom port
uv run custom-mcp-server --port 8080

# Or using the installed command (after pip install -e .)
custom-mcp-server --port 3000

服务器将于启动 http://localhost:8000 默认情况下（或您指定的端口）。

访问服务器

• MCP端点: http://localhost:8000/mcp
• 可用工具:

- health：检查服务器状态 - get_current_user：获取经过身份验证的用户信息 - add_numbers：将两个数字相加（参数化工具示例） - trigger_job_run：触发按作业ID运行的copula作业

测试MCP服务器

此项目包括测试脚本，用于验证您的MCP服务器在本地和部署环境中是否正常工作。

集成测试

该项目包括验证MCP服务器功能的自动化集成测试：

# Run integration tests
uv run pytest tests/ -v -s

测试的作用：

• 自动启动MCP服务器
• 测试一下 list_tools() 工作正常
• 无测试参数工具(health, get_current_user)
• 测试参数化工具(add_numbers 具有各种输入）
• 测试错误处理(trigger_job_run 作业ID无效）
• 测试完成后自动清理服务器

手动测试

端到端测试本地运行的MCP服务器

./scripts/dev/start_server.sh

from databricks_mcp import DatabricksMCPClient
mcp_client = DatabricksMCPClient(
    server_url="http://localhost:8000"
)
# List available MCP tools
print(mcp_client.list_tools())

该脚本无需身份验证即可连接到本地MCP服务器，并列出可用工具。

端到端测试您部署的MCP服务器

部署到copula Apps后，使用交互式shell脚本进行用户级OAuth身份验证测试：

chmod +x scripts/dev/query_remote.sh
./scripts/dev/query_remote.sh

该脚本将指导您完成：

1. 配置文件选择：选择您的copula CLI配置文件
2. 应用程序名称：输入您部署的应用程序名称
3. 自动配置：自动提取应用程序范围和URL
4. OAuth流程：通过浏览器生成用户OAuth令牌
5. 端到端测试：测试 list_tools()，并调用list_tools中返回的每个工具

它的作用：

• 使用检索应用程序配置 databricks apps get
• 从中提取用户授权范围 effective_user_api_scopes
• 从copula配置文件获取工作区主机
• 可选择接受作业ID进行测试 trigger_job_run 工具
• 生成具有正确作用域的OAuth令牌
• 使用用户级身份验证测试MCP客户端
• 验证所有工具是否正常工作： health, get_current_user, add_numbers，并且可选 trigger_job_run

此测试模拟了最终用户授权您的应用程序并使用其凭据时的真实体验。

或者，使用命令行参数手动测试：

python scripts/dev/query_remote.py \
    --host "https://your-workspace.cloud.databricks.com" \
    --token "eyJr...Dkag" \
    --app-url "https://your-workspace.cloud.databricks.com/serving-endpoints/your-app"

这 scripts/dev/query_remote.py 该脚本使用OAuth身份验证连接到部署的MCP服务器，并测试健康检查和用户授权功能。

添加新工具

要向MCP服务器添加新工具，请执行以下操作：

1. 打开 server/tools.py
2. 在内部添加新函数 load_tools() 随着 @mcp_server.tool 装饰师：

@mcp_server.tool
def add_numbers(a: float, b: float) -> dict:
    """
    Add two numbers together.
    
    This tool performs basic arithmetic addition of two numeric values.
    
    Args:
        a (float): The first number to add
        b (float): The second number to add
    
    Returns:
        dict: A dictionary containing the result and input values
    """
    result = a + b
    return {
        "result": result,
        "a": a,
        "b": b,
    }

1. 重新启动服务器-新工具将自动对客户端可用

工具最佳实践

• 明确命名：使用描述性、面向行动的名称
• 全面的文档字符串：AI使用这些来了解何时调用您的工具
• 键入提示：帮助验证和记录
• 结构化回报：返回一致数据的字典或Pydantic模型
• 错误处理：使用try-except块并返回错误信息

连接到copula

这 utils.py 模块提供了两种辅助方法，用于通过Databricks SDK工作区客户端与copula资源交互：

当部署为copula应用程序时：

• get_workspace_client() -返回一个经过身份验证的客户端，该客户端是与应用程序关联的服务主体。看 应用程序授权 了解更多详情。
• get_user_authenticated_workspace_client() -返回一个经过最终用户身份验证的客户端，其作用域由应用程序创建者指定。看 用户授权 了解更多详情。

在本地运行时：

• 由于本地环境中不存在服务主体标识，因此这两种方法都返回一个经过身份验证的客户端作为当前开发人员。

工具使用示例：

from server import utils

# Example 1: Get current user information (user-authenticated)
@mcp_server.tool
def get_current_user() -> dict:
    """Get current user information."""
    try:
        w = utils.get_user_authenticated_workspace_client()
        user = w.current_user.me()
        return {
            "display_name": user.display_name,
            "user_name": user.user_name,
            "active": user.active,
        }
    except Exception as e:
        return {"error": str(e)}

# Example 2: Trigger a Databricks job (app-authenticated)
@mcp_server.tool
def trigger_job_run(job_id: int) -> dict:
    """Trigger a Databricks job run."""
    try:
        w = utils.get_workspace_client()
        run = w.jobs.run_now(job_id=job_id)
        
        # Construct the run page URL
        workspace_host = w.config.host.rstrip("/")
        run_page_url = f"{workspace_host}/jobs/{job_id}/runs/{run.run_id}"
        
        return {
            "success": True,
            "run_id": run.run_id,
            "job_id": job_id,
            "run_page_url": run_page_url,
        }
    except Exception as e:
        return {"success": False, "job_id": job_id, "error": str(e)}

请参阅 get_current_user 和 trigger_job_run 工具在 server/tools.py 对于完整的实现。

生成OAuth令牌

对于高级用例，您可以使用提供的脚本手动生成OAuth令牌，以访问copula工作区。这实现了 OAuth U2M（用户到机器）流.

生成工作区级OAuth令牌

python scripts/dev/generate_oauth_token.py \
    --host https://your-workspace.cloud.databricks.com \
    --scopes "all-apis offline_access"

参数：

• --host：copula工作区URL（必需）
• --scopes：空格分隔的OAuth作用域（默认值： all-apis offline_access)
• --redirect-uri：回调URI（默认值： http://localhost:8020)

注： 脚本使用 databricks-cli 默认情况下为OAuth客户端ID。

脚本将：

1. 生成PKCE代码验证器和质询
2. 打开浏览器进行授权
3. 通过本地HTTP服务器捕获授权码
4. 将代码交换为访问令牌
5. 将令牌响应显示为JSON（令牌有效期为1小时）

自定义范围示例：

python scripts/dev/generate_oauth_token.py \
    --host https://your-workspace.cloud.databricks.com \
    --scopes "clusters:read jobs:write sql:read"

配置

服务器设置

可以使用命令行参数配置服务器：

# Change port
uv run custom-mcp-server --port 8080

# Get help
uv run custom-mcp-server --help

默认配置：

• 主机: 0.0.0.0 （监听所有网络接口）
• 端口: 8000 （可通过以下方式配置 --port 论点）

部署

copula应用程序

此项目已配置为用于copula Apps部署：

1. 使用copula CLI或UI进行部署
2. 服务器将可通过您的copula应用程序URL访问

有关更多信息，请参阅文档 这里

在AI游乐场中尝试您的MCP服务器

在将MCP服务器部署到copula Apps后，您可以在copula AI Playground中交互式地对其进行测试：

1. 导航到 AI游乐场 在您的copula工作区中
2. 选择一个型号 工具已启用 标签
3. 点击 工具>+添加工具 并选择已部署的MCP服务器
4. 开始与AI代理聊天-它会根据需要自动调用您的MCP服务器的工具

AI Playground提供了一个可视化界面，可以在将MCP服务器集成到生产应用程序之前，使用不同的模型和配置对其进行原型制作和测试。

有关更多信息，请参见 AI Playground中的原型工具调用代理.

发展

代码格式化

# Format code with ruff
uv run ruff format .

# Check for lint errors
uv run ruff check .

定制

重命名项目

1. 更新 name 在 pyproject.toml
2. 更新 name 参数在 server/app.py: FastMCP(name="your-name")
3. 更新中的命令脚本 pyproject.toml 在...之下 [project.scripts]

添加自定义API端点

将路线添加到 app 中的FastAPI实例 server/app.py:

@app.get("/custom-endpoint")
def custom_endpoint():
    return {"message": "Hello from custom endpoint"}

故障排除

端口已在使用中

更改端口 server/main.py 或设置 PORT 环境变量。

导入错误

确保安装了所有依赖项：

uv sync  # or pip install -r requirements.txt

资源

• \_CHMCP文档
• copula应用程序
• FastMCP文档
• 模型上下文协议规范
• FastAPI文档
• Uvicorn文档

AI助手上下文

看 Agents.md 了解专门为使用此代码库的AI助手设计的详细项目上下文。