跳至主要内容

模型上下文协议(MCP):含演示项目的指南

了解如何使用 Anthropic 的模型上下文协议(MCP)构建 MCP 服务器,将 Claude 与 GitHub 和 Notion 连接起来。
· 12分钟 阅读

使用 AI 探索

模型上下文协议(MCP) 是一项开放标准,可通过标准化接口,使大型语言模型能够以动态方式与外部工具、数据库和 API 交互。

在这篇博客中,我将带您搭建一个由 MCP 驱动的 PR 审查服务器,并将其集成到 Claude Desktop。该服务器将:

  • 从 GitHub 获取 PR 详情和变更文件
  • 使用 Claude Desktop 应用分析代码变更
  • 生成 PR 审查摘要与建议
  • 将审查结果保存到 Notion

我们将使用模型上下文协议(MCP)来规范服务器与 Claude Desktop 之间的通信,使其模块化、可扩展。

我们通过发送免费周报 The Median,为读者持续更新 AI 最新动态,以清晰梳理本周重点事件。订阅即可每周几分钟快速掌握要点:

什么是模型上下文协议(MCP)?

模型上下文协议(MCP)是由 Anthropic 开发的一项开放标准,旨在实现 AI 模型与外部工具的便捷、标准化集成。它相当于一个通用连接器,使大型语言模型(LLM)能够以动态方式与 API、数据库和业务应用交互。

MCP 最初是为提升 Claude 与外部系统的交互能力而构建,Anthropic 于 2024 年初决定将 MCP 开源,以推动行业广泛采用。通过公开 MCP,他们希望为 AI 与工具的通信建立标准化框架,减少对私有集成的依赖,并提升 AI 应用的模块化与互操作性。

MCP 采用客户端-服务器架构,其中:

  • MCP 客户端 (如 Claude Desktop)请求信息并执行任务。
  • MCP 服务器提供对外部工具和数据源的访问。
  • 宿主应用使用 MCP 在模型与工具之间进行通信。

来源:Model Context Protocol

以下是您在项目中使用 MCP 的理由:

  • 标准化的 AI 集成:MCP 提供了连接 AI 模型与工具的结构化方法。
  • 灵活性:可在不同 AI 模型和厂商之间轻松切换。
  • 安全性:与 AI 交互时可将数据保留在您的基础设施内。
  • 可扩展性:MCP 支持多种传输方式,如 stdio、WebSockets、HTTP SSE 和 UNIX 套接字。

MCP 演示项目概览:PR 审查服务器

该 PR 审查系统通过 Claude Desktop 和 Notion 自动化代码分析与文档记录。

管道的简要流程如下:

  1. 环境配置:加载 GitHub 和 Notion 的凭据。
  2. 服务器初始化:启动 MCP 服务器与 Claude Desktop 通信。
  3. 获取 PR 数据:从 GitHub 检索 PR 变更与元数据。
  4. 代码分析:由 Claude Desktop 直接分析代码变更(无需单独工具)。
  5. Notion 记录:将分析结果保存到 Notion 以便追踪。

步骤 1:设置环境

开始之前,请确保已安装 Python 3.10+。然后,我们配置环境,并先安装 uv 包管理器。Mac 或 Linux:

curl -LsSf https://astral.sh/uv/install.sh | sh  # Mac/Linux

Windows(PowerShell):

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

接着,我们创建新项目目录并使用 uv 初始化:

uv init pr_reviewer
cd pr_reviewer

现在创建并激活虚拟环境。Mac 或 Linux:

uv venv
source .venv/bin/activate

Windows:

.venv\Scripts\activate

现在安装所需依赖:

uv add "mcp[cli]" requests python-dotenv notion-client

本项目我们选择 uv 而非 conda,因为它更快、更轻量,且更专注于 Python 包管理。回顾刚才的操作:

  • 使用 uv 包管理器为 GitHub PR Review MCP 服务器搭建了开发环境
  • 初始化了名为 pr_reviewer 的新项目目录并进入该目录。
  • 创建并激活了虚拟环境,确保包安装彼此隔离。

步骤 2:安装依赖

当环境准备就绪后,我们结合 API 密钥等要求配置依赖。创建 requirements.txt 文件,并加入以下 Python 包:

# Core dependencies for PR Analyzer
requests>=2.31.0          # For GitHub API calls
python-dotenv>=1.0.0      # For environment variables
mcp[cli]>=1.4.0          # For MCP server functionality
notion-client>=2.3.0      # For Notion integration

# Optional: Pin versions for stability
# requests==2.31.0
# python-dotenv==1.0.0
# mcp[cli]==1.4.0
# notion-client==2.3.0

requirements.txt 列出了项目所需的核心依赖。要安装依赖,运行以下任一命令(如先前已安装 uv,优先使用它)。

uv pip install -r requirements.txt
pip install -r requirements.txt

至此,您的环境已安装全部所需依赖。

步骤 3:配置环境变量

接下来,创建一个 .env 文件,包含本项目所需的全部密钥和令牌。

生成 GitHub 令牌

生成 GitHub 令牌的步骤:

  • 登录您的 GitHub 账户并进入设置。
  • 前往 developer settings → Personal Access Tokens。
  • 点击 Generate New Token 并选择 classic 版本。
  • 为令牌命名并启用以下权限:
    • read:org
    • read:repo_hook
    • repo
  • 点击 Generate 生成令牌。
  • 复制并妥善保存令牌,它将用于 API 请求的身份验证。

Notion 集成

  • 登录 Notion 的集成页面。
  • 创建一个新的集成并填写表单。将“type”保持为 Internal,并将其添加到您的工作区。

  • 保存集成并复制集成链接 (URL 末尾的 UUID 为通用唯一标识符,需记录下来,作为 Notion Page ID 使用)。
https://www.notion.so/profile/integrations/internal/UUID
  • 创建完成后,点击该集成,在配置中复制 Internal Integration Secret,这将作为您的 Notion API key。

创建 .env 文件

现在,创建 .env 文件,并将上述生成的 API 密钥和令牌按以下内容填入。

GITHUB_TOKEN=your_github_token
NOTION_API_KEY=your_notion_api_key
NOTION_PAGE_ID=your_notion_page_id

步骤 4:GitHub 集成

现在来配置 GitHub 集成模块,以管理并检索某个仓库中的 PR 变更。

创建 github_integration.py 文件,并写入以下代码(稍后我们会进行讲解)。

import os
import requests
import traceback
from dotenv import load_dotenv

# Load environment variables
load_dotenv()
GITHUB_TOKEN = os.getenv('GITHUB_TOKEN')

def fetch_pr_changes(repo_owner: str, repo_name: str, pr_number: int) -> list:
    """Fetch changes from a GitHub pull request.
    
    Args:
        repo_owner: The owner of the GitHub repository
        repo_name: The name of the GitHub repository
        pr_number: The number of the pull request to analyze
        
    Returns:
        A list of file changes with detailed information about each change
    """
    print(f" Fetching PR changes for {repo_owner}/{repo_name}#{pr_number}")
    
    # Fetch PR details
    pr_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/pulls/{pr_number}"
    files_url = f"{pr_url}/files"
    headers = {'Authorization': f'token {GITHUB_TOKEN}'}
    
    try:
        # Get PR metadata
        pr_response = requests.get(pr_url, headers=headers)
        pr_response.raise_for_status()
        pr_data = pr_response.json()
        
        # Get file changes
        files_response = requests.get(files_url, headers=headers)
        files_response.raise_for_status()
        files_data = files_response.json()
        
        # Combine PR metadata with file changes
        changes = []
        for file in files_data:
            change = {
                'filename': file['filename'],
                'status': file['status'],  # added, modified, removed
                'additions': file['additions'],
                'deletions': file['deletions'],
                'changes': file['changes'],
                'patch': file.get('patch', ''),  # The actual diff
                'raw_url': file.get('raw_url', ''),
                'contents_url': file.get('contents_url', '')
            }
            changes.append(change)
        
        # Add PR metadata
        pr_info = {
            'title': pr_data['title'],
            'description': pr_data['body'],
            'author': pr_data['user']['login'],
            'created_at': pr_data['created_at'],
            'updated_at': pr_data['updated_at'],
            'state': pr_data['state'],
            'total_changes': len(changes),
            'changes': changes
        }
        
        print(f"Successfully fetched {len(changes)} changes")
        return pr_info
        
    except Exception as e:
        print(f"Error fetching PR changes: {str(e)}")
        traceback.print_exc()
        return None

# Example usage for debugging
# pr_data = fetch_pr_changes('owner', 'repo', 1)
# print(pr_data) 

fetch_pr_changes() 函数用于检索并返回指定 GitHub 拉取请求的变更。它接收三个参数:repo_owner、repo_name 和 pr_number, 并返回包含文件变更和 PR 元数据的结构化列表。

代码使用 requests 库发送带身份验证的 HTTP GET 请求,同时获取 PR 的通用元数据和详细的文件级变更:

  • 第一次 API 请求获取高层次 PR 信息,包括标题、描述、作者详情、时间戳和当前状态。
  • 第二次 API 请求获取 PR 中每个变更文件的详情,如文件名、修改状态、增删行数,以及访问文件内容的 URL。

数据检索完成后,函数将 PR 元数据与文件变更组合为字典。文件变更以列表保存,每个条目包含单个文件的详细信息。最终结构包括 PR 标题、描述、作者、时间戳、状态、变更文件总数,以及文件修改的详细清单。

步骤 5:实现 MCP 服务器

在完成依赖安装与辅助函数之后,我们来搭建 MCP 服务器。创建 pr_analyzer.py 文件,它将:

  • 初始化一个 MCP 服务器。
  • 定义用于获取 PR 详情、分析代码并将结果存储到 Notion 的工具。
  • 将这些工具暴露给 Claude Desktop,用于摘要、给出可能的解决方案等。

先加入代码,随后进行讲解。

import sys
import os
import traceback
from typing import Any, List, Dict
from mcp.server.fastmcp import FastMCP
from github_integration import fetch_pr_changes
from notion_client import Client
from dotenv import load_dotenv

class PRAnalyzer:
    def __init__(self):
        # Load environment variables
        load_dotenv()
        
        # Initialize MCP Server
        self.mcp = FastMCP("github_pr_analysis")
        print("MCP Server initialized", file=sys.stderr)
        
        # Initialize Notion client
        self._init_notion()
        
        # Register MCP tools
        self._register_tools()
    
    def _init_notion(self):
        """Initialize the Notion client with API key and page ID."""
        try:
            self.notion_api_key = os.getenv("NOTION_API_KEY")
            self.notion_page_id = os.getenv("NOTION_PAGE_ID")
            
            if not self.notion_api_key or not self.notion_page_id:
                raise ValueError("Missing Notion API key or page ID in environment variables")
            
            self.notion = Client(auth=self.notion_api_key)
            print(f"Notion client initialized successfully", file=sys.stderr)
            print(f"Using Notion page ID: {self.notion_page_id}", file=sys.stderr)
        except Exception as e:
            print(f"Error initializing Notion client: {str(e)}", file=sys.stderr)
            traceback.print_exc(file=sys.stderr)
            sys.exit(1)
    
    def _register_tools(self):
        """Register MCP tools for PR analysis."""
        @self.mcp.tool()
        async def fetch_pr(repo_owner: str, repo_name: str, pr_number: int) -> Dict[str, Any]:
            """Fetch changes from a GitHub pull request."""
            print(f"Fetching PR #{pr_number} from {repo_owner}/{repo_name}", file=sys.stderr)
            try:
                pr_info = fetch_pr_changes(repo_owner, repo_name, pr_number)
                if pr_info is None:
                    print("No changes returned from fetch_pr_changes", file=sys.stderr)
                    return {}
                print(f"Successfully fetched PR information", file=sys.stderr)
                return pr_info
            except Exception as e:
                print(f"Error fetching PR: {str(e)}", file=sys.stderr)
                traceback.print_exc(file=sys.stderr)
                return {}
        
        @self.mcp.tool()
        async def create_notion_page(title: str, content: str) -> str:
            """Create a Notion page with PR analysis."""
            print(f"Creating Notion page: {title}", file=sys.stderr)
            try:
                self.notion.pages.create(
                    parent={"type": "page_id", "page_id": self.notion_page_id},
                    properties={"title": {"title": [{"text": {"content": title}}]}},
                    children=[{
                        "object": "block",
                        "type": "paragraph",
                        "paragraph": {
                            "rich_text": [{
                                "type": "text",
                                "text": {"content": content}
                            }]
                        }
                    }]
                )
                print(f"Notion page '{title}' created successfully!", file=sys.stderr)
                return f"Notion page '{title}' created successfully!"
            except Exception as e:
                error_msg = f"Error creating Notion page: {str(e)}"
                print(error_msg, file=sys.stderr)
                traceback.print_exc(file=sys.stderr)
                return error_msg
    
    def run(self):
        """Start the MCP server."""
        try:
            print("Running MCP Server for GitHub PR Analysis...", file=sys.stderr)
            self.mcp.run(transport="stdio")
        except Exception as e:
            print(f"Fatal Error in MCP Server: {str(e)}", file=sys.stderr)
            traceback.print_exc(file=sys.stderr)
            sys.exit(1)

if __name__ == "__main__":
    analyzer = PRAnalyzer()
    analyzer.run() 

上述代码搭建了一个 MCP 服务器,用于获取 GitHub PR 变更并将分析结果存储到 Notion。关键组件如下:

  1. 环境配置与初始化
    • dotenv 模块用于加载环境变量,确保安全访问 API 密钥与凭据。
    • 随后,PRAnalyzer 类使用 FastMCP() 初始化 MCP 服务器,命名为 github_pr_analysis,以便与 Claude Desktop 进行交互。
    • 同时通过 notion-client 包配置 Notion 客户端,使用存放在 .env 文件中的 Notion API key 与 page ID。
  2. MCP 工具:获取 PR 变更
    • fetch_pr() 函数通过 github_integration.py 中的 fetch_pr_changes() 方法从 GitHub 检索拉取请求的元数据。成功时返回包含 PR 详情的字典。
    • 您可以借助 GitHub 内置的 Webhooks 库将此过程自动化,使新建的 PR 能被服务器自动处理。
  3. MCP 工具:创建 Notion 页面
    • create_notion_page() 函数会根据 PR 分析结果生成 Notion 页面。
    • 它使用 Notion API 在指定的 Notion 工作区下创建新页面。
  4. 服务器运行
    • run() 方法通过 mcp.run(transport="stdio") 启动 MCP 服务器,从而使 Claude Desktop 能与 PR 审查工具交互。

步骤 6:运行 MCP 服务器

当所有代码就位后,在终端运行以下命令启动服务器:

python pr_analyzer.py

Terminal for running the MCP server

Terminal for running MCP server

服务器启动并运行后,打开 Claude Desktop 应用,您会在输入框中看到一个插头图标(🔌)。该图标表示 Claude 环境中存在 MCP。在同一输入框中,您还会注意到一个类似锤子的图标(🔨),点击可查看所有可用的 MCP,如下图所示。

现在,将您想分析的 PR 链接发给 Claude,后续步骤将由它完成。

Claude generating summary

Claude 会先分析该 PR,然后提供摘要与审查意见。它会询问用户是否要将详情上传到 Notion 页面。虽然您可以将该流程自动化,但当前代码允许您在创建新 Notion 页面前先审阅摘要。

Claude uploading summary to notion

更新后的 Notion 页面如下所示:

Notion page created by Claude MCP

结语

我们的基于 MCP 的 PR 审查服务器改进了代码分析与文档流程,提升了审查的效率与组织性。通过使用 MCP、GitHub API 和 Notion 集成,该系统支持自动化 PR 分析、便捷协作与结构化文档记录。借助这一配置,开发者可以快速获取 PR 详情、使用 Claude 分析代码变更,并将洞见存储到 Notion 以备后续参考。

想要了解更多近期的 AI 工具,推荐阅读以下博客:

Aashi Dutt

作者

Aashi Dutt

我是一名 Google Developers 机器学习(生成式 AI)领域的专家、Kaggle 三项专家,以及 Women Techmakers 大使,拥有 3 年以上的技术从业经验。2020 年我共同创办了一家健康科技初创公司,目前在佐治亚理工学院攻读计算机科学硕士,专攻机器学习。

主题