4.1 MCP服务器（Server）开发

4.1 MCP服务器（Server）开发

第四章：MCP协议开发与实现

4.1 MCP服务器（Server）开发：构建AI的“触角”与“能力引擎”

在模型上下文协议（Model Context Protocol，简称MCP）的宏大愿景中，AI不再是一个孤立的智能体，而是能够安全、灵活地与外部世界互动、获取实时信息并执行复杂任务的强大协作伙伴。实现这一愿景的核心支柱之一，便是MCP服务器（Server）。本章节将聚焦于MCP服务器的开发，深入探讨其设计理念、核心功能、关键技术以及实践指南，揭示如何构建AI与真实世界交互的桥梁。

4.1.1 MCP服务器的角色与重要性

想象一下，大型语言模型（LLM）拥有卓越的语言理解和生成能力，但它本身并不具备直接访问本地文件、查询数据库、调用外部API或控制特定应用的能力。这就好比一位博学的学者，虽然满腹经纶，却无法直接翻阅图书馆的最新藏书或操作复杂的实验设备。MCP服务器正是为了弥补这一差距而设计的。

根据MCP的客户端-服务器（Client-Server）架构：

• 主机（Host）: LLM应用本身（例如，一个集成了LLM的IDE、聊天客户端或自动化平台）。

• 客户端（Client）: 运行在主机内部，负责与MCP服务器建立连接、路由消息、管理服务器能力（如发现可用的工具、资源、提示词）以及处理用户授权。

• 服务器（Server）: 本章节的重点，是提供外部数据和工具的实体。它可以是一个独立的进程、服务或库，通过实现MCP协议与客户端通信。服务器负责将特定领域的知识（资源）、可执行的操作（工具）和预设的工作流程（提示词）暴露给LLM。

简而言之，MCP服务器扮演着AI的“触角”和“能力引擎”的角色。它将AI的能力从纯粹的文本生成扩展到实际操作层面，让AI能够：

1. 获取实时或特定领域的上下文信息（Resources）: 例如，读取项目代码文件、查询数据库记录、获取API返回的最新天气数据、访问用户文档等。

2. 执行外部操作（Tools）: 例如，创建GitHub Issue、运行Shell命令、调用计算器进行精确计算、发送邮件、操作浏览器自动化脚本等。

3. 利用预设的工作流程（Prompts）: 提供结构化的指令模板，引导AI或用户完成特定任务，如代码分析、错误调试流程等。

在MCP出现之前，AI与外部能力的集成通常通过Function Calling等机制实现，但这往往缺乏统一标准，导致不同AI平台和不同工具之间需要重复开发适配器，形成“集成地狱”。MCP通过提供一个开放、标准化的协议，极大地简化了这一过程。开发者只需按照MCP规范开发一次服务器，即可被任何兼容MCP协议的客户端（如Claude Desktop、Cursor等）调用，极大地提升了生态系统的互操作性和开发效率。

4.1.2 MCP服务器的核心原语实现

MCP服务器的核心功能在于暴露和管理三种基本原语：提示词（Prompts）、资源（Resources）和工具（Tools）。理解并正确实现这些原语是开发MCP服务器的关键。

4.1.2.1 提示词（Prompts）的实现

提示词允许服务器定义可复用、参数化的文本模板或多轮对话流程，客户端（进而LLM或用户）可以调用这些模板来获取结构化的指令或指导。这对于定义常见的任务流程或提供领域特定的指导非常有用。

核心概念与结构：

一个提示词通常包含：

• name: 提示词的唯一标识符（字符串）。

• description: 对提示词功能的简短描述（可选，字符串），供客户端向用户展示。

• arguments: 提示词接受的参数列表（可选，数组）。每个参数有 name（参数标识符）、description（参数描述）和 required（是否必填）等属性。

结构示例（Mermaid语法）：

服务器实现要点：

1. 定义提示词元数据： 服务器需要定义每个提示词的名称、描述和所需的参数。这部分信息通过 prompts/list 请求暴露给客户端。

2. 处理 prompts/list 请求： 当客户端发送 prompts/list 请求时，服务器需要返回所有可用提示词的元数据列表。

3. 处理 prompts/get 请求： 当客户端（通常是在LLM决定使用某个提示词，并提供参数后）发送 prompts/get 请求时，服务器需要根据请求中指定的提示词名称和提供的参数，生成具体的提示消息（messages 数组，遵循LLM消息格式，包含角色、内容等）并返回。提示词的内容可以是静态文本，也可以根据参数动态生成，甚至可以嵌入资源引用。

动态提示词：

服务器可以实现动态提示词，根据客户端提供的参数生成不同的消息内容。例如，一个代码分析提示词可以接受 language 和 fileUri 作为参数，服务器根据这些参数构建包含特定语言代码文件内容的分析指令。实现时，服务器需要根据 prompts/get 请求中的 params 字段来动态生成返回的 messages 数组。

代码实现示例（概念性，使用Python SDK类比）：

from mcp.server.fastmcp import FastMCP
from mcp.mcp import GetPromptRequest, GetPromptResult, PromptMessage, RoleAssistant, NewTextContent, NewResourceContent
mcp = FastMCP("MyPromptServer")
# 定义一个简单的问候提示词
@mcp.prompt(
    name="greeting",
    description="一个友好的问候提示",
    arguments=[{"name": "name", "description": "要问候的人的名字", "required": False}]
)
def handle_greeting_prompt(request: GetPromptRequest) -> GetPromptResult:
    name = request.params.arguments.get("name", "朋友")
    message_content = NewTextContent(f"你好，{name}！今天有什么可以帮您的吗？")
    message = PromptMessage(role=RoleAssistant, content=message_content)
    return NewGetPromptResult(description="友好的问候", messages=[message])
# 定义一个包含资源的动态提示词
@mcp.prompt(
    name="analyze-log-and-code",
    description="分析日志和代码文件",
    arguments=[
        {"name": "log_uri", "description": "日志文件URI", "required": True},
        {"name": "code_uri", "description": "代码文件URI", "required": True}
    ]
)
def handle_analysis_prompt(request: GetPromptRequest) -> GetPromptResult:
    log_uri = request.params.arguments["log_uri"]
    code_uri = request.params.arguments["code_uri"]
    messages = [
        PromptMessage(role=RoleAssistant, content=NewTextContent("请分析以下日志和代码，找出潜在问题：")),
        # 这里需要服务器实际读取资源内容并填充，或者仅提供URI让客户端/LLM处理（取决于协议版本和实现）
        # 更典型的MCP实现中，prompt/get返回的是消息结构，包含resource type和URI
        PromptMessage(role=RoleAssistant, content=NewResourceContent(uri=log_uri, mime_type="text/plain")),
        PromptMessage(role=RoleAssistant, content=NewResourceContent(uri=code_uri, mime_type="text/x-python")),
        PromptMessage(role=RoleAssistant, content=NewTextContent("请重点关注日志中的错误信息与代码中的异常处理。"))
    ]
    return NewGetPromptResult(description="日志和代码分析提示", messages=messages)
# ... 服务器启动逻辑 ...
​

4.1.2.2 资源（Resources）的实现

资源代表服务器能够提供的可读数据或内容，通常用于为LLM提供上下文信息。资源可以是文件、数据库记录、API响应、屏幕内容等。

核心概念与结构：

每个资源由一个唯一的URI标识，格式通常为 [协议]://[主机]/[路径]。服务器可以定义自定义协议和路径结构。资源内容可以是文本（UTF-8编码）或二进制（Base64编码）。

资源元数据（通过 resources/list 请求暴露）包含：

• uri: 资源的唯一标识符（字符串）。

• name: 资源的友好名称（字符串）。

• description: 资源描述（可选，字符串）。

• mimeType: 资源的MIME类型（可选，字符串），用于指示内容格式。

对于动态资源（如根据参数生成的日志报告），服务器可以暴露URI模板（uriTemplate）而非具体URI，客户端根据模板构建实际的资源URI。

结构示例（Mermaid语法）：

资源内容结构（通过 resources/read 请求返回）：

服务器实现要点：

1. 定义资源元数据： 服务器需要确定提供哪些资源，并为每个资源定义元数据（URI、名称、描述、MIME类型）。这部分信息用于响应 resources/list 请求。对于动态资源，暴露URI模板。

2. 处理 resources/list 请求： 返回可用资源的元数据列表（包括静态资源的URI和动态资源的URI模板）。

3. 处理 resources/read 请求： 当客户端请求读取某个资源时（通过URI指定），服务器需要执行实际的数据读取操作（如打开文件、查询数据库、调用API），将内容封装到 ResourceContents 结构中，并作为 resources/read 请求的响应返回。文本内容直接放入 text 字段，二进制内容进行Base64编码后放入 blob 字。

4. 实现资源更新通知（可选）： 对于内容会动态变化的资源，服务器可以实现资源更新通知机制。客户端通过 resources/subscribe 订阅特定URI，当资源内容变化时，服务器发送 notifications/resources/updated 通知。客户端收到通知后，可以再次发起 resources/read 请求获取最新内容。列表变更则通过 notifications/resources/list_changed 通知。

代码实现示例（概念性，使用Python SDK类比）：

from mcp.server.fastmcp import FastMCP
from mcp.mcp import ReadResourceRequest, ResourceContents, TextResourceContents, ResourceMetadata
import os
from pathlib import Path
mcp = FastMCP("MyResourceServer")
# 定义一个静态文件资源
@mcp.resource(
    uri="file:///local/readme.md",
    name="项目说明文档",
    description="项目的 README 文件",
    mime_type="text/markdown"
)
def read_readme(request: ReadResourceRequest) -> list[ResourceContents]:
    # 假设 README.md 位于服务器运行目录下
    try:
        content = Path("README.md").read_text(encoding="utf-8")
        return [TextResourceContents(uri=request.params.uri, mime_type="text/markdown", text=content)]
    except FileNotFoundError:
        # 根据协议规范，错误应通过 JSON-RPC 响应的 error 字段返回
        # SDK通常会处理异常到错误响应的转换
        raise FileNotFoundError(f"Resource not found: {request.params.uri}")
# 定义一个动态日志资源（URI模板）
@mcp.resource(
    uri_template="logs://server/app?timeframe={timeframe}",
    name="应用日志",
    description="按时间范围查询应用日志",
    mime_type="text/plain"
)
def read_app_logs(request: ReadResourceRequest) -> list[ResourceContents]:
    # 从请求URI中解析参数
    # 注意：SDK通常提供辅助函数解析URI模板参数
    # 这里的实现简化，实际可能需要手动解析或依赖SDK特性
    timeframe = "1h" # 假设默认值或从URI解析
    # 实际从日志系统查询数据
    log_content = f"Simulated logs for timeframe: {timeframe}\nLine 1\nLine 2..."
    return [TextResourceContents(uri=request.params.uri, mime_type="text/plain", text=log_content)]
# ... 服务器启动逻辑 ...
​

注意：实际的SDK会提供更方便的方式来处理URI模板参数解析和错误返回。

4.1.2.3 工具（Tools）的实现

工具是MCP服务器最强大的能力之一，它允许服务器向客户端（进而LLM）暴露可执行的功能。LLM可以通过调用这些工具来执行实际操作，例如修改文件、调用API、运行脚本等。

核心概念与结构：

每个工具包含：

• name: 工具的唯一标识符（字符串）。

• description: 工具功能的描述（可选，字符串），供LLM理解其用途。

• inputSchema: 定义工具接受的参数结构（JSON Schema），用于描述参数名称、类型、是否必填等。这使得LLM能够理解如何构造工具调用的请求。

结构示例（Mermaid语法）：

服务器实现要点：

1. 定义工具元数据和输入 Schema： 服务器需要为每个工具定义名称、描述，并使用JSON Schema定义其参数结构。这部分信息用于响应 tools/list 请求。

2. 处理 tools/list 请求： 返回所有可用工具的元数据列表，包括名称、描述和输入 Schema。LLM客户端通过解析这些信息来决定何时以及如何调用工具。

3. 处理 tools/call 请求： 这是工具执行的核心。当客户端发送 tools/call 请求时，服务器需要：

   • 验证请求中的工具名称是否存在。

   • 根据工具的 inputSchema 验证请求参数 (params.arguments) 是否符合要求。这是重要的安全措施，防止恶意或无效参数。

   • 执行工具的实际逻辑（例如，调用外部API、运行脚本、执行计算等）。

   • 将执行结果封装到 CallToolResult 结构中返回。结果通常包含一个 content 字段，可以是文本、JSON或其他结构化数据，描述工具执行的结果。

安全性：

工具调用通常涉及对外部系统进行写操作或执行有副作用的任务。MCP协议强调安全性，客户端在调用工具前通常需要征得用户的明确批准（Human-in-the-loop），并且服务器端必须进行严格的输入验证和权限检查，确保AI模型不会执行未经授权或危险的操作。

代码实现示例（概念性，使用Python SDK类比）：

from mcp.server.fastmcp import FastMCP
from mcp.mcp import CallToolRequest, CallToolResult, TextContent, FormatTextResult
import json # 通常工具结果是结构化数据，JSON很常见
mcp = FastMCP("MyToolServer")
# 定义一个简单的计算器工具
@mcp.tool(
    name="calculate",
    description="执行基本的算术运算",
    input_schema={
        "type": "object",
        "properties": {
            "operation": {"type": "string", "description": "要执行的算术运算类型", "enum": ["add", "subtract", "multiply", "divide"]},
            "x": {"type": "number", "description": "第一个数字"},
            "y": {"type": "number", "description": "第二个数字"}
        },
        "required": ["operation", "x", "y"]
    }
)
def handle_calculate_tool(request: CallToolRequest) -> CallToolResult:
    op = request.params.arguments.get("operation")
    x = request.params.arguments.get("x")
    y = request.params.arguments.get("y")
    # 输入验证（虽然SDK会做部分，但业务逻辑层可能需要更细致的检查）
    if op not in ["add", "subtract", "multiply", "divide"]:
        # 返回错误，SDK会将其转换为JSON-RPC error响应
        raise ValueError(f"Invalid operation: {op}")
    if not isinstance(x, (int, float)) or not isinstance(y, (int, float)):
         raise ValueError("x and y must be numbers")
    result = None
    try:
        if op == "add":
            result = x + y
        elif op == "subtract":
            result = x - y
        elif op == "multiply":
            result = x * y
        elif op == "divide":
            if y == 0:
                raise ValueError("Division by zero is not allowed")
            result = x / y
    except Exception as e:
         raise RuntimeError(f"Calculation failed: {e}") # 抛出异常由SDK处理为错误响应
    # 返回结果
    # 协议允许多种结果格式，FormatTextResult是一个方便的辅助函数
    return FormatTextResult(str(result))
# 定义一个模拟的GitHub创建Issue工具
@mcp.tool(
    name="github_create_issue",
    description="在 GitHub 仓库创建 Issue",
    input_schema={
        "type": "object",
        "properties": {
            "repo": {"type": "string", "description": "仓库名称 (格式: owner/repo)"},
            "title": {"type": "string", "description": "Issue 标题"},
            "body": {"type": "string", "description": "Issue 内容", "required": False},
            "labels": {"type": "array", "items": {"type": "string"}, "description": "Issue 标签列表", "required": False}
        },
        "required": ["repo", "title"]
    }
)
def handle_create_issue_tool(request: CallToolRequest) -> CallToolResult:
    repo = request.params.arguments.get("repo")
    title = request.params.arguments.get("title")
    body = request.params.arguments.get("body", "")
    labels = request.params.arguments.get("labels", [])
    # 实际调用 GitHub API
    # import requests # 假设已安装 requests 库
    # github_token = os.getenv("GITHUB_TOKEN") # 从环境变量获取敏感信息
    # if not github_token:
    #     raise PermissionError("GitHub token not configured")
    #
    # api_url = f"https://api.github.com/repos/{repo}/issues"
    # headers = {"Authorization": f"token {github_token}"}
    # data = {"title": title, "body": body, "labels": labels}
    #
    # response = requests.post(api_url, headers=headers, json=data)
    # response.raise_for_status() # 检查HTTP错误
    # issue_data = response.json()
    # 模拟成功响应
    issue_data = {"html_url": f"https://github.com/{repo}/issues/123", "number": 123}
    # 返回 Issue 创建结果的结构化数据
    # 协议允许返回 JSON 结构作为结果
    return CallToolResult(content=[{"type": "json", "json": issue_data}])
# ... 服务器启动逻辑 ...
​

注意：敏感信息（如API密钥）不应硬编码在服务器代码中，应通过环境变量、安全配置等方式加载。

4.1.3 消息传输与生命周期

MCP服务器通过基础协议与客户端通信，该协议基于JSON-RPC 2.0消息格式，并支持多种传输机制。

消息格式（JSON-RPC 2.0）：

• Request（请求）: 客户端向服务器（或反之）发起操作。包含 jsonrpc ("2.0")、id（请求唯一标识）、method（方法名，如 "prompts/list"、"resources/read"、"tools/call"）、params（方法参数）。

• Response（响应）: 对请求的回复。包含 jsonrpc ("2.0")、id（对应请求的id）、result（成功结果）或 error（错误信息，包含 code 和 message）。

• Notification（通知）: 单向消息，无需回复。包含 jsonrpc ("2.0")、method（通知方法名，如 "notifications/resources/updated"）、params（通知参数）。

消息流示例（Mermaid语法）：

传输机制（Transports）：

MCP内置支持两种标准传输机制：

1. 标准输入/输出 (Stdio): 通过进程的标准输入和标准输出流进行通信。适用于本地集成、命令行工具或简单的进程间通信。实现相对简单，常用于本地开发和调试。

2. Server-Sent Events (SSE): 基于HTTP POST请求实现客户端到服务器通信，同时支持服务器到客户端的流式推送。适用于需要服务器主动向客户端发送更新（如资源内容变更通知）或在受限网络环境中的场景。

选择哪种传输方式取决于服务器的部署环境和功能需求。许多SDK（如Go SDK mcp-go）提供了对这两种传输方式的抽象支持。

生命周期：

MCP服务器的生命周期通常包括初始化、运行和关闭阶段：

1. 初始化： 服务器启动，加载配置，注册其提供的Prompts、Resources和Tools。客户端连接后，会发送 initialize 请求，服务器返回自身信息（名称、版本）和支持的能力。

2. 运行： 服务器进入主循环，监听传入的请求（Requests）和订阅（如资源更新）。接收到请求后，根据方法名路由到相应的处理函数，执行逻辑，并返回响应（Responses）。对于已订阅的资源，内容变更时发送通知（Notifications）。

3. 关闭： 服务器接收到关闭信号（如客户端断开连接或收到 shutdown 请求）后，进行资源清理，优雅地终止运行。

服务器开发者需要确保在初始化阶段正确注册所有能力，在运行阶段高效处理请求和通知，并在关闭阶段妥善清理资源。

4.1.4 开发实践与工具

开发一个MCP服务器通常会依赖于特定语言的SDK或框架，这些工具封装了底层的JSON-RPC消息处理、传输机制和能力注册等细节，让开发者可以专注于业务逻辑。

常见的MCP SDK/框架包括：

• Python: mcp (尤其是 fastmcp 模块), FastMCP

• TypeScript/JavaScript: @modelcontextprotocol/server, mcp-framework, FastMCP

• Go: mcp-go, Foxy Contexts

• Rust: ModelContextProtocol.NET (via C#)

• Java: quarkus-mcp-server, spring-ai-mcp

这些SDK通常提供：

• 能力注册机制： 通过装饰器（如Python的@mcp.tool）或特定函数方便地将处理逻辑注册为Prompts、Resources或Tools。

• 请求解析与路由： 自动解析传入的JSON-RPC请求，并将其路由到对应的处理函数，将请求参数作为函数参数传递。

• 响应封装： 自动将处理函数的返回值或抛出的异常封装为符合JSON-RPC规范的成功或错误响应。

• 传输层抽象： 提供简单的方式来选择和配置Stdio或SSE等传输方式。

• 类型安全和验证： 利用语言特性或Schema定义，提供输入参数的自动验证。

开发流程建议：

1. 选择语言和SDK： 根据项目需求、团队熟悉度和现有生态选择合适的语言和MCP SDK。

2. 定义服务器能力： 明确你的服务器需要向AI暴露哪些Prompts、Resources和Tools。

3. 实现处理函数： 针对每项能力，编写对应的处理函数，实现具体的业务逻辑（如文件操作、数据库查询、API调用）。

4. 注册能力： 使用SDK提供的机制（如装饰器或注册函数）将处理函数与MCP原语关联起来，并提供元数据（名称、描述、Schema等）。

5. 选择传输方式： 配置服务器使用Stdio（常用于本地开发）或SSE（常用于网络服务）。

6. 测试与调试： 利用MCP开发工具进行测试。

重要的开发工具：

• MCP CLI (@mcpm/cli 或 SDK 内置CLI): 用于安装、连接、管理MCP服务器。

• MCP Inspector: 一个Web界面工具，可以连接到正在运行的MCP服务器，可视化其提供的Prompts、Resources和Tools，并允许手动发送请求进行测试和调试。这是开发过程中非常有价值的工具。通常通过运行 mcp dev your_server_file.py 命令启动，它会同时运行服务器和Inspector代理。

测试流程示例（Mermaid语法）：

通过 mcp dev 命令启动服务器，开发者可以在浏览器中打开MCP Inspector，查看服务器暴露的所有Prompts、Resources和Tools，检查它们的元数据和Schema，并直接向服务器发送测试请求，观察响应和任何错误信息。这极大地提高了开发效率和调试便利性。

4.1.5 安全性考虑

构建安全的MCP服务器至关重要，因为服务器可能处理敏感数据或执行有潜在风险的操作。开发者在设计和实现时应考虑以下安全措施：

1. 权限控制： 严格控制服务器对底层系统和外部服务的访问权限。服务器应只被授予执行其声明功能所需的最小权限。例如，文件系统服务器应限制在特定目录下操作；数据库服务器应使用只读账户（除非明确需要写操作）并限制可执行的查询类型。

2. 输入验证： 严格验证所有传入请求的参数，尤其是工具调用的参数。使用JSON Schema定义并强制执行参数类型、格式和取值范围，防止注入攻击或其他恶意操作。对于字符串参数，进行转义或过滤，防止命令注入或SQL注入。对于数值参数，验证其是否在合理范围内。

3. 安全配置： 敏感信息（如API密钥、数据库密码）不应硬编码在代码中，而应通过环境变量、配置文件或密钥管理服务安全地加载。确保配置文件具有适当的访问权限。

4. 最小化攻击面： 移除不必要的功能和服务，减少潜在的漏洞。定期更新依赖库，修复已知的安全漏洞。

5. 日志记录和监控： 记录所有重要的操作和事件，包括请求、响应、错误和安全事件。监控服务器的性能和安全指标，及时发现异常行为。

6. 传输层安全： 对于通过网络传输敏感数据的服务器（如使用SSE），启用TLS加密，确保数据在传输过程中不被窃听或篡改。

7. 用户授权： 某些操作可能需要用户的明确授权。MCP协议本身不强制规定授权机制，但服务器可以集成OAuth 2.0或其他授权协议，确保只有经过授权的用户才能调用特定工具或访问特定资源。

4.1.6 案例研究：构建一个GitHub Issue创建服务器

为了更具体地说明MCP服务器的开发过程，我们考虑一个案例：构建一个GitHub Issue创建服务器。这个服务器允许LLM通过调用工具，在指定的GitHub仓库中创建新的Issue。

1. 确定服务器能力：

• 工具： github_create_issue

  • name: github_create_issue

  • description: "在指定的GitHub仓库创建Issue"

  • inputSchema:

    {
      "type": "object",
      "properties": {
        "repo": { "type": "string", "description": "仓库名称 (格式: owner/repo)" },
        "title": { "type": "string", "description": "Issue 标题" },
        "body": { "type": "string", "description": "Issue 内容", "required": false },
        "labels": { "type": "array", "items": { "type": "string" }, "description": "Issue 标签列表", "required": false }
      },
      "required": [ "repo", "title" ]
    }
    ​

2. 选择语言和SDK：

这里选择Python和mcp SDK。

3. 实现处理函数：

import os
import requests
from mcp.server.fastmcp import FastMCP
from mcp.mcp import CallToolRequest, CallToolResult, TextContent, FormatTextResult
mcp = FastMCP("GitHubIssueCreator")
@mcp.tool(
    name="github_create_issue",
    description="在指定的GitHub仓库创建Issue",
    input_schema={
        "type": "object",
        "properties": {
            "repo": {"type": "string", "description": "仓库名称 (格式: owner/repo)"},
            "title": {"type": "string", "description": "Issue 标题"},
            "body": {"type": "string", "description": "Issue 内容", "required": False},
            "labels": {"type": "array", "items": {"type": "string"}, "description": "Issue 标签列表", "required": False}
        },
        "required": ["repo", "title"]
    }
)
def handle_create_issue_tool(request: CallToolRequest) -> CallToolResult:
    repo = request.params.arguments.get("repo")
    title = request.params.arguments.get("title")
    body = request.params.arguments.get("body", "")
    labels = request.params.arguments.get("labels", [])
    github_token = os.getenv("GITHUB_TOKEN")
    if not github_token:
        raise PermissionError("GitHub token not configured")
    api_url = f"https://api.github.com/repos/{repo}/issues"
    headers = {"Authorization": f"token {github_token}"}
    data = {"title": title, "body": body, "labels": labels}
    try:
        response = requests.post(api_url, headers=headers, json=data)
        response.raise_for_status()
        issue_data = response.json()
        return CallToolResult(content=[{"type": "json", "json": issue_data}])
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Failed to create issue: {e}")
if __name__ == "__main__":
    mcp.run()
​

4. 配置和运行：

1. 确保安装了requests库: pip install requests

2. 设置环境变量 GITHUB_TOKEN 为你的GitHub Personal Access Token (需要 repo 权限).

3. 运行服务器: python your_server_file.py

5. 测试：

使用MCP Inspector或自定义客户端发送 tools/call 请求，验证Issue是否成功创建。

安全性：

• 从环境变量加载 GITHUB_TOKEN，而不是硬编码。

• 验证 repo 参数的格式，防止注入攻击。

• 使用 requests.raise_for_status() 检查API响应状态码，确保请求成功。

4.1.7 高级主题

1. 多租户支持： 如果服务器需要为多个用户或组织提供服务，需要实现多租户支持，确保不同租户之间的数据隔离和权限隔离。

2. 异步操作： 对于耗时的操作（如调用外部API），使用异步编程技术（如Python的asyncio）避免阻塞服务器主循环，提高响应速度。

3. 缓存： 对于读取频率高但更新频率低的资源，实现缓存机制，减少对底层数据源的访问压力。

4. 监控和告警： 集成监控系统，收集服务器的性能指标（如CPU使用率、内存占用、请求延迟）和错误信息，设置告警规则，及时发现和解决问题。

5. 自定义传输协议： 虽然MCP内置支持Stdio和SSE，但在某些特殊场景下，可能需要实现自定义的传输协议。这需要深入理解MCP协议规范和底层网络编程技术。

6. 流式响应： 对于大型资源或需要实时推送的数据，可以使用流式响应，将数据分块发送给客户端，提高传输效率和用户体验。

4.1.8 未来展望

MCP服务器的未来发展方向包括：

1. 更丰富的生态系统： 涌现更多高质量、领域特定的MCP服务器，覆盖更广泛的应用场景。

2. 更智能的工具选择： LLM能够更准确、更智能地选择和组合工具，完成更复杂的任务。

3. 更强大的安全机制： 完善安全机制，防止AI模型执行未经授权或危险的操作。

4. 与边缘计算的融合： MCP服务器部署到边缘设备，实现低延迟、本地化的AI服务。

5. 多模态支持： 扩展MCP协议，支持图像、视频等多模态数据的处理和传输。

4.1.9 总结

MCP服务器是实现AI与真实世界交互的关键组件。通过理解和正确实现提示词、资源和工具这三种核心原语，并关注安全性、性能和可扩展性，开发者可以构建强大的MCP服务器，为LLM提供丰富的上下文信息和强大的操作能力，开启AI应用的新篇章。随着MCP生态系统的不断发展和完善，我们有理由相信，AI将在更多领域发挥更大的作用。