官方 MCP Server 生态：filesystem / postgres / redis / puppeteer 等核心 Server

第三十七章：MCP 生态与官方 Server：文件系统、数据库、浏览器控制

37.1 MCP 生态概览

自 2024年11月 MCP 正式发布以来，围绕这一协议形成了一个快速增长的 Server 生态。Anthropic 官方维护了一批高质量的参考 Server，涵盖最常见的集成场景；同时，开源社区贡献了数百个第三方 Server，支持从代码托管平台到企业 SaaS 的各类服务。

官方 Server 仓库：https://github.com/modelcontextprotocol/servers

MCP Server 生态可以分为以下几个大类：

37.2 文件系统 Server

@modelcontextprotocol/server-filesystem 是最基础也最常用的 MCP Server，为 AI 提供对本地文件系统的受控访问。

37.2.1 安装与配置

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/alice/Documents",
        "/Users/alice/code"
      ]
    }
  }
}

Server 启动时接受一个或多个目录路径作为参数，这些目录是 AI 可以访问的"根目录"。所有文件操作都被限制在这些目录内，这是核心安全保障。

37.2.2 提供的工具

37.2.3 实际使用场景

场景：自动化代码重构

用户：「把 src/components 目录下所有 .jsx 文件改成 .tsx，并添加 TypeScript 类型注解」

Claude 通过文件系统 Server 会自动：

1. 调用 list_directory 列出所有 .jsx 文件
2. 对每个文件调用 read_file 读取内容
3. 分析代码，添加类型注解
4. 调用 write_file 写入 .tsx 文件
5. 确认后删除旧的 .jsx 文件

场景：代码库分析

"分析这个 Python 项目的依赖关系，找出循环导入"

# Claude 会：
# 1. list_directory 获取项目结构
# 2. read_multiple_files 读取所有 .py 文件
# 3. 分析 import 语句，构建依赖图
# 4. 检测循环依赖

37.2.4 使用 Python 自定义文件系统 Server

如果官方 Server 不满足需求（如需要自定义过滤规则、需要与其他系统集成），可以扩展官方实现：

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import os
import fnmatch
from pathlib import Path

app = Server("enhanced-filesystem")
ALLOWED_ROOTS = [Path(p) for p in os.environ.get("ROOTS", "/tmp").split(":")]
IGNORE_PATTERNS = [".git", "__pycache__", "node_modules", "*.pyc", ".DS_Store"]

def should_ignore(path: Path) -> bool:
    return any(
        fnmatch.fnmatch(path.name, pattern) or path.name == pattern
        for pattern in IGNORE_PATTERNS
    )

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="smart_search",
            description="智能搜索：支持内容搜索和文件名搜索，自动忽略二进制文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "search_type": {
                        "type": "string",
                        "enum": ["content", "filename", "both"],
                        "default": "both"
                    },
                    "extensions": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "限制文件扩展名，如 ['.py', '.js']"
                    }
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="diff_files",
            description="对比两个文件的差异",
            inputSchema={
                "type": "object",
                "properties": {
                    "file_a": {"type": "string"},
                    "file_b": {"type": "string"}
                },
                "required": ["file_a", "file_b"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "smart_search":
        return await _smart_search(arguments)
    elif name == "diff_files":
        return await _diff_files(arguments)
    raise ValueError(f"未知工具：{name}")

async def _smart_search(args: dict):
    query = args["query"]
    search_type = args.get("search_type", "both")
    extensions = args.get("extensions", [])
    
    results = []
    for root in ALLOWED_ROOTS:
        for path in root.rglob("*"):
            if not path.is_file() or should_ignore(path):
                continue
            if extensions and path.suffix not in extensions:
                continue
            
            # 文件名匹配
            if search_type in ("filename", "both"):
                if query.lower() in path.name.lower():
                    results.append(f"[文件名] {path.relative_to(root)}")
            
            # 内容匹配（跳过大文件和二进制文件）
            if search_type in ("content", "both") and path.stat().st_size < 1_000_000:
                try:
                    content = path.read_text(encoding="utf-8", errors="ignore")
                    if query.lower() in content.lower():
                        # 找到匹配行
                        lines = [
                            f"  L{i+1}: {line.strip()}"
                            for i, line in enumerate(content.splitlines())
                            if query.lower() in line.lower()
                        ][:3]
                        results.append(f"[内容] {path.relative_to(root)}:\n" + "\n".join(lines))
                except Exception:
                    pass
    
    if not results:
        return [types.TextContent(type="text", text=f"未找到 '{query}'")]
    return [types.TextContent(
        type="text",
        text=f"找到 {len(results)} 个匹配：\n\n" + "\n\n".join(results[:20])
    )]

async def _diff_files(args: dict):
    import difflib
    try:
        a = Path(args["file_a"]).read_text(encoding="utf-8")
        b = Path(args["file_b"]).read_text(encoding="utf-8")
        diff = list(difflib.unified_diff(
            a.splitlines(keepends=True),
            b.splitlines(keepends=True),
            fromfile=args["file_a"],
            tofile=args["file_b"]
        ))
        if not diff:
            return [types.TextContent(type="text", text="两个文件内容相同")]
        return [types.TextContent(type="text", text="".join(diff))]
    except FileNotFoundError as e:
        return [types.TextContent(type="text", text=f"文件不存在：{e}")]

37.3 数据库 Server

37.3.1 PostgreSQL Server

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

提供的工具：

• query：执行 SQL 查询（只读，SELECT 语句）
• describe_table：查看表结构
• list_tables：列出所有表

提供的资源：

• postgres://{host}/{database}/schema：数据库的完整 Schema

官方 PostgreSQL Server 默认只允许 SELECT 查询，这是重要的安全设计——AI 可以分析数据，但不能修改。

实际使用场景：

用户：「分析 orders 表，找出过去30天内每个客户的平均订单金额，并按金额降序排列」

Claude 会：
1. 调用 describe_table("orders") 了解表结构
2. 构造 SQL 查询：
   SELECT customer_id, AVG(amount) as avg_amount
   FROM orders  
   WHERE created_at >= NOW() - INTERVAL '30 days'
   GROUP BY customer_id
   ORDER BY avg_amount DESC
3. 调用 query() 执行并返回结果

37.3.2 SQLite Server

SQLite Server 支持读写操作，适合本地开发和轻量级数据库场景：

{
  "mcpServers": {
    "sqlite": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-sqlite", "--db-path", "/path/to/database.db"]
    }
  }
}

提供的工具（相比 PostgreSQL 更多写操作权限）：

• read_query：执行 SELECT 查询
• write_query：执行 INSERT/UPDATE/DELETE
• create_table：创建新表
• list_tables：列出所有表
• describe_table：查看表结构
• append_insight：向备忘录资源添加内容（用于 AI 记录分析发现）

37.3.3 自定义数据库 MCP Server

如果需要连接 MySQL 或其他数据库，可以自己实现：

# mysql_mcp_server.py
import mysql.connector
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import os
import json

app = Server("mysql-server")

def get_connection():
    return mysql.connector.connect(
        host=os.environ.get("MYSQL_HOST", "localhost"),
        port=int(os.environ.get("MYSQL_PORT", 3306)),
        user=os.environ.get("MYSQL_USER"),
        password=os.environ.get("MYSQL_PASSWORD"),
        database=os.environ.get("MYSQL_DATABASE"),
        autocommit=False  # 显式事务控制
    )

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="query",
            description="执行 MySQL SELECT 查询",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SELECT 语句"},
                    "limit": {"type": "integer", "default": 100, "description": "最大返回行数"}
                },
                "required": ["sql"]
            }
        ),
        types.Tool(
            name="list_tables",
            description="列出数据库中的所有表",
            inputSchema={"type": "object", "properties": {}}
        ),
        types.Tool(
            name="describe_table",
            description="查看表的列结构",
            inputSchema={
                "type": "object",
                "properties": {"table": {"type": "string"}},
                "required": ["table"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = get_connection()
    try:
        cursor = conn.cursor(dictionary=True)
        
        if name == "query":
            sql = arguments["sql"].strip()
            # 安全检查：只允许 SELECT
            if not sql.upper().startswith("SELECT"):
                return [types.TextContent(
                    type="text",
                    text="安全限制：只允许 SELECT 查询"
                )]
            limit = min(arguments.get("limit", 100), 1000)
            if "LIMIT" not in sql.upper():
                sql += f" LIMIT {limit}"
            cursor.execute(sql)
            rows = cursor.fetchall()
            return [types.TextContent(
                type="text",
                text=f"查询返回 {len(rows)} 行：\n{json.dumps(rows, ensure_ascii=False, default=str, indent=2)}"
            )]
        
        elif name == "list_tables":
            cursor.execute("SHOW TABLES")
            tables = [list(row.values())[0] for row in cursor.fetchall()]
            return [types.TextContent(
                type="text",
                text=f"数据库中共有 {len(tables)} 张表：\n" + "\n".join(tables)
            )]
        
        elif name == "describe_table":
            cursor.execute(f"DESCRIBE `{arguments['table']}`")
            columns = cursor.fetchall()
            text = f"表 {arguments['table']} 的结构：\n"
            for col in columns:
                nullable = "NULL" if col["Null"] == "YES" else "NOT NULL"
                key = f" [{col['Key']}]" if col["Key"] else ""
                text += f"  {col['Field']} {col['Type']} {nullable}{key}\n"
            return [types.TextContent(type="text", text=text)]
    
    finally:
        conn.close()

import asyncio
async def main():
    async with stdio_server() as (r, w):
        await app.run(r, w, app.create_initialization_options())

asyncio.run(main())

37.4 GitHub Server

37.4.1 配置

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

37.4.2 提供的核心工具

仓库操作：

• create_or_update_file：创建或更新文件
• get_file_contents：获取文件内容
• push_files：批量推送文件变更
• create_repository：创建新仓库
• fork_repository：Fork 仓库

Issue 管理：

• create_issue：创建 Issue
• list_issues：列出 Issues
• update_issue：更新 Issue
• add_issue_comment：添加评论

Pull Request 操作：

• create_pull_request：创建 PR
• get_pull_request：获取 PR 详情
• list_pull_requests：列出 PR
• merge_pull_request：合并 PR
• create_pull_request_review：提交代码审查

搜索：

• search_repositories：搜索仓库
• search_code：搜索代码
• search_issues：搜索 Issue

37.4.3 实际使用场景

用户：「检查 anthropics/sdk 最近7天内新增的 Issues，总结主要问题，并创建一个跟踪 Issue」

Claude 通过 GitHub Server 会：
1. search_issues(repo="anthropics/sdk", since="7天前", state="open")
2. 分析所有新 Issue 的标题和内容
3. 归类问题（如：安全漏洞、功能请求、文档问题）
4. create_issue(title="Issue 周报摘要", body="本周新增 XX 个 Issue，主要集中在...")

37.5 浏览器自动化 Server

37.5.1 Puppeteer Server

{
  "mcpServers": {
    "puppeteer": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-puppeteer"]
    }
  }
}

提供的工具：

37.5.2 网页信息提取实例

用户：「访问 https://news.ycombinator.com，截图并提取今日前10条新闻标题和链接」

Claude 会：
1. puppeteer_navigate("https://news.ycombinator.com")
2. puppeteer_screenshot() → 获得页面截图
3. puppeteer_evaluate("""
     Array.from(document.querySelectorAll('.titleline > a'))
       .slice(0, 10)
       .map(a => ({title: a.textContent, url: a.href}))
   """)
4. 返回结构化的新闻列表

37.5.3 表单自动化实例

用户：「帮我登录 example.com，然后截图首页的仪表板」

Claude 会：
1. puppeteer_navigate("https://example.com/login")
2. puppeteer_fill("#username", "[email protected]")
3. puppeteer_fill("#password", "password")
4. puppeteer_click("#submit-button")
5. puppeteer_screenshot() → 返回仪表板截图

37.5.4 Playwright Server（更强大的替代选项）

社区维护的 @playwright/mcp（2025年 Playwright 官方发布）提供了更稳定的浏览器自动化能力，并支持 Chromium、Firefox 和 WebKit：

{
  "mcpServers": {
    "playwright": {
      "command": "npx",
      "args": ["@playwright/mcp@latest"],
      "env": {
        "PLAYWRIGHT_BROWSER": "chromium"
      }
    }
  }
}

Playwright MCP 的亮点功能：

• 无障碍树模式：优先使用 ARIA 树而非截图，更准确地理解页面结构
• 视觉模式：支持截图并直接分析图像内容
• 多标签页管理：支持同时操作多个标签页
• 网络拦截：可以监控和修改网络请求

37.6 Memory Server（持久化记忆）

37.6.1 配置与使用

{
  "mcpServers": {
    "memory": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-memory"]
    }
  }
}

Memory Server 为 AI 提供了一个跨会话的持久化知识图谱存储。

提供的工具：

• create_entities：创建实体（人物、地点、概念等）
• create_relations：创建实体间的关系
• add_observations：为实体添加观察记录
• delete_entities：删除实体
• delete_observations：删除观察记录
• delete_relations：删除关系
• search_nodes：搜索知识图谱
• open_nodes：获取特定节点详情
• read_graph：读取完整知识图谱

37.6.2 典型使用模式

Memory Server 的典型用途是让 AI 记住用户的偏好、项目背景和重要事实：

[会话1]
用户：「我正在开发一个叫 TaskFlow 的 React 项目，使用 TypeScript 和 Tailwind CSS，
      目前最大的挑战是状态管理」

Claude 使用 Memory Server 记录：
- create_entity({name: "TaskFlow", type: "Project", observations: ["React+TS+Tailwind", "状态管理是当前挑战"]})
- create_entity({name: "用户", type: "Developer"})
- create_relation({from: "用户", to: "TaskFlow", relation: "正在开发"})

[会话2，数天后]
用户：「我上次提到的项目进展怎么样了？」

Claude 使用 Memory Server 搜索：
- search_nodes("TaskFlow") → 找回项目信息
- 返回：「根据我的记录，你在开发 TaskFlow 项目（React+TS+Tailwind），
         上次你提到状态管理是主要挑战，我们当时讨论了...」

37.7 搜索 Server：Brave Search

{
  "mcpServers": {
    "brave-search": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-brave-search"],
      "env": {
        "BRAVE_API_KEY": "your_api_key"
      }
    }
  }
}

工具：

• brave_web_search：网络搜索，返回结果摘要和链接
• brave_local_search：本地地点搜索（餐厅、商店等）

Brave Search 相比 Google 搜索的优势：不跟踪用户数据，API 价格较低，且提供专用的 AI 查询接口。

37.8 构建企业级 MCP Server 集合

37.8.1 典型企业配置

{
  "mcpServers": {
    "codebase": {
      "command": "python",
      "args": ["/opt/mcp-servers/codebase_server.py"],
      "env": {
        "REPOS_ROOT": "/data/repositories",
        "ALLOWED_ORGS": "mycompany"
      }
    },
    "jira": {
      "command": "node",
      "args": ["/opt/mcp-servers/jira_server.js"],
      "env": {
        "JIRA_BASE_URL": "https://mycompany.atlassian.net",
        "JIRA_API_TOKEN": "${JIRA_TOKEN}",
        "JIRA_EMAIL": "[email protected]"
      }
    },
    "confluence": {
      "command": "node",
      "args": ["/opt/mcp-servers/confluence_server.js"],
      "env": {
        "CONFLUENCE_BASE_URL": "https://mycompany.atlassian.net/wiki",
        "CONFLUENCE_API_TOKEN": "${CONFLUENCE_TOKEN}"
      }
    },
    "slack-notify": {
      "command": "python",
      "args": ["/opt/mcp-servers/slack_server.py"],
      "env": {
        "SLACK_BOT_TOKEN": "${SLACK_BOT_TOKEN}",
        "ALLOWED_CHANNELS": "dev-team,alerts"
      },
      "alwaysAllow": []
    }
  }
}

37.8.2 MCP Server 注册中心

对于需要管理大量 MCP Server 的团队，可以建立一个内部 Server 注册中心：

# server_registry.py
import json
from pathlib import Path

SERVER_REGISTRY = {
    "filesystem": {
        "description": "本地文件系统访问",
        "package": "@modelcontextprotocol/server-filesystem",
        "args_template": ["${WORKSPACE_PATH}"],
        "required_env": [],
        "optional_env": [],
        "risk_level": "medium"
    },
    "postgres": {
        "description": "PostgreSQL 只读访问",
        "package": "@modelcontextprotocol/server-postgres",
        "args_template": [],
        "required_env": ["POSTGRES_CONNECTION_STRING"],
        "optional_env": [],
        "risk_level": "low"
    },
    "puppeteer": {
        "description": "浏览器自动化",
        "package": "@modelcontextprotocol/server-puppeteer",
        "args_template": [],
        "required_env": [],
        "optional_env": ["PUPPETEER_HEADLESS"],
        "risk_level": "high"
    }
}

def generate_config(selected_servers: list, env_values: dict) -> dict:
    """根据选择的 Server 和环境变量生成配置"""
    config = {"mcpServers": {}}
    
    for server_name in selected_servers:
        if server_name not in SERVER_REGISTRY:
            continue
        
        info = SERVER_REGISTRY[server_name]
        env = {k: env_values.get(k, f"${{{k}}}") for k in info["required_env"]}
        
        config["mcpServers"][server_name] = {
            "command": "npx",
            "args": ["-y", info["package"]] + [
                arg.replace("${WORKSPACE_PATH}", env_values.get("WORKSPACE_PATH", "."))
                for arg in info["args_template"]
            ],
            "env": env
        }
    
    return config

小结

MCP 生态提供了覆盖文件系统、数据库、代码托管、浏览器自动化、搜索、持久化记忆等全方位场景的官方 Server。借助这些 Server，AI 可以真正地操作本地环境、查询数据库、自动化 Web 操作。

关键要点：

1. 文件系统 Server 通过"根目录限制"确保 AI 操作的安全边界
2. PostgreSQL Server 默认只读，SQLite Server 支持写操作
3. GitHub Server 支持完整的 Issue、PR 和代码搜索工作流
4. Puppeteer/Playwright Server 让 AI 具备浏览器自动化能力
5. Memory Server 通过知识图谱实现跨会话的持久记忆
6. 企业部署时，建立内部 Server 注册中心便于统一管理

下一章将深入讨论 MCP 的安全模型：OAuth 认证、沙箱隔离与最小权限原则。