MCP 生态与常用 Server 实践

写 MCP Server 之前，先看看哪些已经有了——大部分常见场景都已经有人实现了，不用重复造轮子。

1.1 MCP 生态现状（2026 年 3 月）

MCP 生态全景图
MCP 生态全景——官方与社区 Server 覆盖文件系统、代码开发、数据库、搜索引擎等各类场景

截至 2026 年 3 月，MCP 生态已经相当成熟。官方和社区 Server 数量超过数千个，涵盖开发工具、数据库、SaaS 平台、云服务等各类场景。

值得关注的里程碑：OpenAI 在 2025 年宣布支持 MCP。这意味着 MCP 已经不只是 Anthropic 的私有协议，而是成了整个 AI 行业事实上的工具调用标准。当最大的竞争对手也采用你的标准，这个标准基本就赢了。

Anthropic 官方维护的 Server（TypeScript 实现）：

code

@modelcontextprotocol/server-filesystem    文件系统读写
@modelcontextprotocol/server-github        GitHub 仓库操作
@modelcontextprotocol/server-google-drive  Google Drive 文件
@modelcontextprotocol/server-slack         Slack 消息和频道
@modelcontextprotocol/server-postgres      PostgreSQL 数据库
@modelcontextprotocol/server-sqlite        SQLite 数据库（轻量级嵌入式数据库）
@modelcontextprotocol/server-brave-search  Brave 搜索引擎
@modelcontextprotocol/server-puppeteer     Chrome 浏览器自动化
@modelcontextprotocol/server-memory        本地知识图谱/记忆存储

社区热门 Server：

code

mcp-server-fetch          HTTP 请求工具（调任意 API）
mcp-server-redis          Redis 缓存操作
mcp-server-aws            AWS 云服务
mcp-server-notion         Notion 笔记操作
mcp-server-jira           Jira 工单管理
mcp-server-docker         Docker 容器管理
mcp-server-kubernetes     K8s 集群操作

寻找更多 Server：github.com/modelcontextprotocol/servers 是官方维护的列表，也可以在 GitHub 搜索 mcp-server 关键词。

1.2 实战一：配置文件系统 Server

这是最常用的入门配置，让 Claude 能读写你指定目录的文件：

json

// claude_desktop_config.json
// npx 是 Node.js 自带工具，用于直接运行 npm 包，无需手动安装
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/yourname/Documents",
        "/Users/yourname/Desktop"
      ]
    }
  }
}

-y 参数表示自动确认安装（如果包未安装，npx 会自动下载）。路径参数可以写多个，Claude 只能访问这些目录。

配置后 Claude 可以做的事情：

"帮我读取 Desktop 上的 report.txt，总结主要内容"
"在 Documents 目录里创建一个 meeting-notes.md，记录今天的会议要点"
"找出 Documents 里所有包含'季度'关键词的文档"

1.3 实战二：配置 GitHub Server

让 Claude 能操作 GitHub 仓库，非常适合开发工作流：

第一步：生成 GitHub Personal Access Token

进入 GitHub Settings → Developer settings → Personal access tokens → Tokens (classic)，生成一个 Token，勾选需要的权限（repo 用于代码操作，issues 用于 issue 管理）。

第二步：配置 claude_desktop_config.json

json

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

配置后 Claude 可以做的事情：

"查看 myrepo 仓库最近 10 个 open issue"
"帮我创建一个 issue，标题是'登录页面 UI 优化'，描述是……"
"读取 src/main.py 文件的内容，给我解释这段代码"
"搜索仓库里所有包含 TODO 注释的文件"

1.4 实战三：配置数据库 Server

让 Claude 能查询你的数据库，用于数据分析和业务查询：

PostgreSQL 配置：

json

{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-postgres",
        "postgresql://username:password@localhost:5432/mydb"
      ]
    }
  }
}

SQLite 配置（适合本地数据文件）：

json

{
  "mcpServers": {
    "sqlite": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-sqlite",
        "/Users/yourname/data/mydata.db"
      ]
    }
  }
}

配置后 Claude 可以做的事情：

"查询上个月销售额最高的 10 个产品"
"帮我写一个 SQL，统计每个地区的用户数量，并且按用户数降序排列"
"分析这张表的数据结构，告诉我有哪些字段"

注意：数据库 Server 默认支持执行任意 SQL，包括写操作。如果不想让 AI 修改数据，应该：

给数据库用户只读权限（GRANT SELECT ON ALL TABLES TO readonly_user）
或者在 Server 层面限制只允许 SELECT 操作

1.5 多个 Server 同时配置

可以在配置文件中同时配置多个 Server，Claude 会根据任务自动选择合适的工具：

json

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/yourname/workspace"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://user:pass@localhost/mydb"]
    },
    "my-custom-server": {
      "command": "python",
      "args": ["/Users/yourname/mcp_servers/custom_server.py"]
    }
  }
}

配置多个 Server 后，Claude 拥有了文件操作、代码管理、数据查询和自定义业务功能，可以完成复杂的跨系统任务。

1.6 为企业内部系统构建 MCP Server

企业内部有很多系统——工单系统、ERP、CRM、内部 API——这些都可以封装成 MCP Server，让 AI 助手能直接使用。

下面是一个企业工单系统的 MCP Server 示例：

python

# ticket_server.py
# 企业内部工单系统 MCP Server
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import httpx  # 异步 HTTP 客户端，比 requests 更适合 async 环境
import json

app = Server("internal-ticket-system")

# 内部系统配置
TICKET_API_BASE = "https://internal.company.com/api/tickets"
API_TOKEN = "your-internal-api-token"  # 实际使用时从环境变量读取：os.getenv("TICKET_API_TOKEN")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="create_ticket",
            description="创建新工单。适用于用户报告问题、请求支持或提交需求时使用。",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "工单标题，简明描述问题，50字以内"
                    },
                    "description": {
                        "type": "string",
                        "description": "详细问题描述，包括复现步骤、期望结果、实际结果"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high", "urgent"],
                        "description": "优先级：low=低，medium=中，high=高，urgent=紧急"
                    },
                    "assignee": {
                        "type": "string",
                        "description": "负责人邮箱（可选，不填则自动分配）"
                    }
                },
                "required": ["title", "description", "priority"]
            }
        ),
        types.Tool(
            name="search_tickets",
            description="搜索工单。根据关键词和状态查找相关工单。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词，如'登录失败'、'性能问题'"
                    },
                    "status": {
                        "type": "string",
                        "enum": ["open", "in_progress", "closed", "all"],
                        "description": "工单状态筛选，all=全部，默认 all"
                    }
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="get_ticket",
            description="获取工单详情，包括描述、状态、评论历史。需要工单 ID。",
            inputSchema={
                "type": "object",
                "properties": {
                    "ticket_id": {
                        "type": "string",
                        "description": "工单 ID，格式如 'TICKET-1234'"
                    }
                },
                "required": ["ticket_id"]
            }
        ),
        types.Tool(
            name="update_ticket_status",
            description="更新工单状态。用于标记工单进度。",
            inputSchema={
                "type": "object",
                "properties": {
                    "ticket_id": {
                        "type": "string",
                        "description": "工单 ID"
                    },
                    "status": {
                        "type": "string",
                        "enum": ["open", "in_progress", "resolved", "closed"],
                        "description": "新状态"
                    },
                    "comment": {
                        "type": "string",
                        "description": "状态变更说明（可选）"
                    }
                },
                "required": ["ticket_id", "status"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    # Bearer Token：HTTP 身份验证方式，把密钥放在请求头里证明身份
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }

    # httpx.AsyncClient 是异步 HTTP 客户端，async with 确保连接自动关闭
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            if name == "create_ticket":
                response = await client.post(
                    f"{TICKET_API_BASE}/create",
                    json={
                        "title": arguments["title"],
                        "description": arguments["description"],
                        "priority": arguments["priority"],
                        "assignee": arguments.get("assignee")
                    },
                    headers=headers
                )
                response.raise_for_status()  # 如果状态码不是 2xx，抛出异常
                data = response.json()
                return [types.TextContent(
                    type="text",
                    text=f"工单创建成功！\n"
                         f"工单编号：#{data['id']}\n"
                         f"标题：{data['title']}\n"
                         f"状态：{data['status']}\n"
                         f"链接：{data['url']}"
                )]

            elif name == "search_tickets":
                params = {
                    "q": arguments["query"],
                    "status": arguments.get("status", "all")
                }
                response = await client.get(
                    f"{TICKET_API_BASE}/search",
                    params=params,
                    headers=headers
                )
                response.raise_for_status()
                tickets = response.json().get("tickets", [])

                if not tickets:
                    return [types.TextContent(
                        type="text",
                        text=f"没有找到与 '{arguments['query']}' 相关的工单。"
                    )]

                result = f"找到 {len(tickets)} 个工单（显示前10条）：\n\n"
                for t in tickets[:10]:
                    result += f"- **#{t['id']}** [{t['status']}] {t['title']}\n"
                    result += f"  优先级：{t['priority']}，负责人：{t.get('assignee', '未分配')}\n"

                return [types.TextContent(type="text", text=result)]

            elif name == "get_ticket":
                response = await client.get(
                    f"{TICKET_API_BASE}/{arguments['ticket_id']}",
                    headers=headers
                )
                response.raise_for_status()
                ticket = response.json()

                result = f"""工单详情：

**#{ticket['id']} - {ticket['title']}**
- 状态：{ticket['status']}
- 优先级：{ticket['priority']}
- 负责人：{ticket.get('assignee', '未分配')}
- 创建时间：{ticket['created_at']}
- 更新时间：{ticket['updated_at']}

**描述：**
{ticket['description']}
"""
                comments = ticket.get('comments', [])
                if comments:
                    result += f"\n**最近评论（共{len(comments)}条）：**\n"
                    for comment in comments[-3:]:  # 只显示最近3条
                        result += f"- {comment['author']}（{comment['created_at']}）：{comment['content']}\n"

                return [types.TextContent(type="text", text=result)]

            elif name == "update_ticket_status":
                response = await client.patch(
                    f"{TICKET_API_BASE}/{arguments['ticket_id']}/status",
                    json={
                        "status": arguments["status"],
                        "comment": arguments.get("comment", "")
                    },
                    headers=headers
                )
                response.raise_for_status()
                return [types.TextContent(
                    type="text",
                    text=f"工单 #{arguments['ticket_id']} 状态已更新为：{arguments['status']}"
                )]

        except httpx.HTTPStatusError as e:
            return [types.TextContent(
                type="text",
                text=f"API 请求失败（{e.response.status_code}）：{e.response.text[:200]}"
            )]
        except httpx.ConnectError:
            return [types.TextContent(
                type="text",
                text="无法连接到工单系统，请检查网络或联系管理员"
            )]
        except Exception as e:
            return [types.TextContent(type="text", text=f"操作失败：{str(e)}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream,
                     app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

1.7 MCP vs Function Calling：场景选择

虽然两者都能实现工具调用，但在不同场景下有明显的优劣：

场景	推荐方案	原因
在代码中动态定义工具	Function Calling（第 9 章）	灵活，运行时可以动态增减工具
工具需要在多个 AI 应用复用	MCP	一次实现，到处使用
个人开发环境工具（文件、Git）	MCP	Claude 桌面版等客户端直接支持
企业内部系统集成	MCP	标准化接口，便于统一维护
生产环境 AI 服务（需要精细控制）	Function Calling	更好的调用控制和监控
已有 LangChain/LangGraph 项目	Function Calling + LangChain Tools	与现有代码集成更顺畅

简单判断规则：

如果是写代码调用 LLM，用 Function Calling
如果是给 Claude 桌面版/Cursor 等现成 AI 客户端配置工具，用 MCP

1.8 MCP Server 的安全最佳实践

MCP Server 赋予了 AI 操作真实系统的能力，安全必须认真对待：

最小权限原则

不要暴露不必要的工具。数据库 Server 只给只读权限；文件 Server 只允许访问特定目录；工单 Server 只开放需要的操作类型。

python

# 好的做法：严格限制文件访问范围
ALLOWED_ROOT = "/home/user/ai-workspace"  # 只有这一个目录

# 坏的做法：允许访问整个系统
# ALLOWED_ROOT = "/"

输入验证

不要相信 AI 传来的参数，都要做合法性检查：

python

def validate_ticket_id(ticket_id: str) -> bool:
    """工单 ID 只能是 TICKET- 开头加数字"""
    import re
    return bool(re.match(r'^TICKET-\d+$', ticket_id))

def validate_path(filename: str) -> bool:
    """防止路径遍历攻击"""
    normalized = os.path.normpath(filename)
    return not normalized.startswith('..') and '..' not in normalized

敏感操作需要确认

删除数据、发送邮件、修改生产配置这类操作，考虑在工具执行前设置一个确认步骤，或者要求 AI 在执行前明确告知用户将要做什么。

日志记录

记录所有工具调用，便于审计和排查问题：

python

import logging
import datetime

logger = logging.getLogger(__name__)

# 在 call_tool 函数开头添加
logger.info(f"[{datetime.datetime.now()}] 工具调用：{name}，参数：{json.dumps(arguments, ensure_ascii=False)}")

速率限制

防止 AI 无限循环调用工具导致 API 过度消耗：

python

from collections import defaultdict
import time

call_counts = defaultdict(list)

def check_rate_limit(tool_name: str, max_calls: int = 10, window_seconds: int = 60) -> bool:
    """每分钟最多调用 N 次"""
    now = time.time()
    calls = call_counts[tool_name]
    # 清理过期记录
    calls[:] = [t for t in calls if now - t < window_seconds]
    if len(calls) >= max_calls:
        return False
    calls.append(now)
    return True

1.9 小结：MCP 是 AI 应用的基础设施

从第 9 章的 Function Calling（单个应用内调用工具），到第 10 章的 MCP（标准化、可复用的工具服务），这是 AI 工具生态从"各自为战"走向"生态协同"的关键一步。

对企业来说，MCP 的价值在于：把公司内部系统（ERP、CRM、工单系统、数据仓库）接入 MCP 标准，一次接入，公司里所有的 AI 工具都能共用，大幅降低 AI 集成的长期维护成本。

现在能做到的事情：

用现成的官方 Server，让 Claude 桌面版操作文件、GitHub、数据库
为企业内部系统写 MCP Server，接入 AI 工作流
在自己开发的 AI 应用里集成 MCP Client，使用标准化工具生态

会写 MCP Server 正在成为 AI 工程师的基本技能，就像会写 REST API 是后端工程师的基本技能一样。