MCP协议深度解析：让AI真正连接世界的技术革命（附LangChain/Claude Desktop实战）
·
目录
- 引言：AI为什么需要MCP
- 一、MCP到底是什么
- 二、MCP核心架构原理
- 三、MCP vs 传统API：本质区别
- 四、MCP Server 开发实战
- 五、Claude Desktop 配置 MCP
- 六、Claude AI 连接本地资源
- 七、Multi-Agent 多Agent协同
- 八、完整项目：用MCP打造私有知识库AI助手
- 九、常见问题与避坑指南
- 总结与展望
引言：AI为什么需要MCP
2023年之后，大模型能力突飞猛进，但一个根本矛盾始终没有解决：模型再强，也只是一个"聪明的哑巴"——它无法主动读取你的数据库、搜索你的文件、操控你的浏览器。
为了解决这个问题，行业走过三个阶段：
| 阶段 | 方式 | 痛点 |
|---|---|---|
| Prompt Engineering | 把所有信息塞进提示词 | 上下文窗口爆炸、信息过时 |
| Function Calling / Tool Use | 让模型调用预定义函数 | 每个工具要单独对接、无统一协议 |
| MCP (Model Context Protocol) | 统一协议层 + 可插拔Server | 真正实现"即插即用"的AI工具生态 |
MCP 的出现，本质上是把 USB 接口的概念引入了 AI 世界。 就像 USB 让各种设备可以即插即用地连接到电脑，MCP 让各种数据源和工具可以即插即用地连接到 AI 模型。
一、MCP到底是什么
1.1 官方定义
MCP（Model Context Protocol） 是由 Anthropic 于2024年底开源的一种标准化协议，用于在 AI 模型和数据源/工具之间建立双向通信通道。
官网：https://modelcontextprotocol.io
开源仓库：https://github.com/modelcontextprotocol
1.2 MCP的核心目标
┌─────────────────────────────────────────────────────────────┐
│ MCP 的三个核心目标 │
├─────────────────────────────────────────────────────────────┤
│ 1. 🔌 即插即用 (Plug & Play) │
│ 任何 MCP Server 实现协议即可被任何 MCP Host 消费 │
│ │
│ 2. 🔒 安全沙箱 (Secure Sandbox) │
│ Host 控制资源访问权限，Server 无法绕过授权 │
│ │
│ 3. 📦 可组合生态 (Composable Ecosystem) │
│ 多个 Server 可以同时运行，工具可被动态发现和组合 │
└─────────────────────────────────────────────────────────────┘
1.3 协议传输层
MCP 使用 JSON-RPC 2.0 作为消息格式，支持两种传输层：
传输层选项：
├── stdio ← 本地进程通信（Claude Desktop、CLI工具）
│ 优点：零网络配置、天然沙箱隔离
│
└── HTTP + SSE ← 远程通信（生产环境）
优点：支持分布式部署、多客户端共享
SSE：Server-Sent Events 实现服务端推送
二、MCP核心架构原理
2.1 三层架构总览
MCP 采用 Host → Client → Server 三层分离架构，这是理解 MCP 最关键的部分：
┌──────────────────────────────────────────────────────────────┐
│ MCP 架构全景图 │
│ │
│ ┌──────────────┐ │
│ │ MCP Host │ ← 用户交互入口（Claude Desktop / Claude AI / Agent）│
│ │ (主机层) │ 负责：解析用户意图 → 分发任务 → 展示结果 │
│ └──────┬───────┘ │
│ │ 1:N │
│ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ MCP Client Layer │ │
│ │ (客户端层) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │Client:Filesystem│Client:Database│Client:Browser│ │ │
│ │ │ (文件系统) │ │ (数据库) │ │ (浏览器) │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ └─────────┼────────────────┼────────────────┼─────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ MCP Server Layer │ │
│ │ (服务端层) │ │
│ │ │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ FileSystem │ │ PostgreSQL │ │ Chrome │ │ │
│ │ │ Server │ │ Server │ │ DevTools │ │ │
│ │ │ (本地文件) │ │ Server │ │ Server │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
│ │ ↑ ↑ ↑ │ │
│ │ │ │ │ │ │
│ │ 本地文件系统 远程数据库 浏览器自动化 │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘
2.2 三层职责详解
🏠 MCP Host（主机层）
定义： AI应用的前端入口，负责用户交互和任务编排。
职责：
- 接收用户自然语言输入
- 调用 MCP Client 发送请求
- 聚合多个 Server 返回的结果
- 管理授权和权限控制
- 渲染最终响应给用户
代表实现：
- Claude Desktop App
- Claude.ai 网页版
- LangChain MCP Adapters
- 自定义 AI Agent
🔌 MCP Client（客户端层）
定义： Host 与 Server 之间的 1:1 通信代理。每个连接的 Server 对应一个 Client 实例。
职责：
- 维护与 Server 的长连接（会话生命周期管理）
- 序列化和反序列化 JSON-RPC 消息
- 处理协议握手和版本协商
- 路由请求/响应到正确的 Server
关键特性： 一个 Host 可以同时运行多个 Client，每个 Client 独立管理一个 Server 连接。
⚙️ MCP Server（服务端层）
定义： 暴露工具（Tools）、资源（Resources）、提示模板（Prompts）的服务进程。
三类核心能力：
┌──────────────────────────────────────────────────────────────┐
│ MCP Server 三大能力 │
├────────────────┬─────────────────────────┬───────────────────┤
│ Tools │ Resources │ Prompts │
│ (工具) │ (资源) │ (提示模板) │
├────────────────┼─────────────────────────┼───────────────────┤
│ AI可调用的函数 │ AI可读取的数据源 │ 预定义的提示词模板 │
│ 有副作用操作 │ 只读数据访问 │ 可带参数生成 │
│ e.g. 搜索网页 │ e.g. 本地文件内容 │ e.g. 代码评审模板 │
│ e.g. 查询数据库 │ e.g. API响应 │ e.g. 周报生成模板 │
│ e.g. 发送邮件 │ e.g. 日志流 │ e.g. 会议纪要模板 │
└────────────────┴─────────────────────────┴───────────────────┘
2.3 通信流程完整解析
用户: "帮我查一下 orders 表中昨天订单量最多的前10个用户"
┌──────────┐ JSON-RPC ┌──────────┐ JSON-RPC ┌──────────────┐
│ Host │ ──────────────▶ │ Client │ ──────────────▶ │ Server │
│ (Claude) │ "查询订单数据" │(PostgreSQL)│ "execute_sql" │ (Database) │
│ │ ◀────────────── │ │ ◀────────────── │ │
│ │ 返回结构化数据 │ │ SQL执行结果 │ │
└──────────┘ └──────────┘ └──────────────┘
具体 JSON-RPC 消息示例：
【请求】tools/call
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "execute_sql",
"arguments": {
"query": "SELECT user_id, COUNT(*) as cnt FROM orders WHERE ..."
}
}
}
【响应】
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "user_007: 156单, user_123: 142单, ..."
}
]
}
}
2.4 协议消息类型一览
MCP 协议定义了以下核心 JSON-RPC 方法：
┌─────────────────────────────────────────────────────────────────┐
│ MCP JSON-RPC 方法全览 │
├──────────────────────┬──────────────────────────────────────────┤
│ 初始化 │ initialize ← 握手 + 版本协商 │
│ │ initialized ← 握手完成确认 │
├──────────────────────┼──────────────────────────────────────────┤
│ 资源管理 (Resources) │ resources/list ← 列出可用资源 │
│ │ resources/read ← 读取资源内容 │
│ │ resources/subscribe ← 订阅资源变更 │
├──────────────────────┼──────────────────────────────────────────┤
│ 工具调用 (Tools) │ tools/list ← 列出可用工具 │
│ │ tools/call ← 调用工具并获取结果 │
├──────────────────────┼──────────────────────────────────────────┤
│ 提示模板 (Prompts) │ prompts/list ← 列出可用模板 │
│ │ prompts/get ← 获取模板内容 │
├──────────────────────┼──────────────────────────────────────────┤
│ 日志级别 │ logging/setLevel ← 设置日志级别 │
└──────────────────────┴──────────────────────────────────────────┘
三、MCP vs 传统API：本质区别
3.1 核心差异对比表
| 维度 | 传统 API 调用 | MCP |
|---|---|---|
| 接口发现 | 开发者手动查阅文档 | Server 启动时自动声明 (tools/list) |
| 类型安全 | 靠约定或手写 SDK | JSON Schema 自动校验参数结构 |
| 认证方式 | 每个 API 独立鉴权 | Host 统一管理，所有 Server 共享安全上下文 |
| 版本协商 | URL 路径版本或 Header | 协议层 initialize 握手协商 |
| 双向通信 | 单向请求-响应 | 支持 SSE 实现服务端推送 |
| 工具组合 | 硬编码组合 | 运行时动态发现，Host 自由编排 |
| 开发成本 | 每对组合需单独适配 | Server 写一次，任何 Host 即插即用 |
| 沙箱安全 | 依赖 API 提供方 | Host 控制资源访问，Server 无特权 |
3.2 动态发现机制（核心创新）
传统 API 的痛点：AI 不知道有哪些工具可用，只能靠提示词告诉它。
【传统方式】—— AI 靠"背答案"知道工具
Prompt: "你有三个工具: search_web(query), get_weather(city), send_email(to, body)"
问题：工具列表写死在 Prompt 里，添加新工具要改 Prompt
【MCP 方式】—— AI 运行时动态发现
1. Server 启动 → 注册 tools/list
2. Host 调用 tools/list → 获取所有可用工具及其 JSON Schema
3. 模型在运行时知道确切有哪些工具、参数是什么
4. 新增 Server 只需重启，AI 自动感知，无需改 Prompt
3.3 类型安全示例
// MCP Server 声明的工具 schema（MCP 自动生成）
{
"name": "query_database",
"description": "执行只读 SQL 查询",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL 查询语句（仅支持 SELECT）"
},
"timeout_ms": {
"type": "integer",
"description": "查询超时（毫秒）",
"default": 30000
}
},
"required": ["query"]
}
}
对比传统 API：这个 JSON Schema 就是自动生成的"接口文档"，AI 和代码都可以直接解析，无需人工维护。
四、MCP Server 开发实战
4.1 环境准备
# Python 版所需依赖
pip install mcp python-dotenv psycopg2-binary
# TypeScript 版所需依赖
npm install @modelcontextprotocol/sdk zod dotenv
4.2 Python 实现：一个 PostgreSQL MCP Server
这个 Server 提供数据库查询能力，是生产环境中非常常见的场景。
# server/postgres_mcp_server.py
"""
MCP Server 示例：PostgreSQL 数据库查询工具
功能：安全地暴露数据库只读查询能力给 AI
"""
import json
import os
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import AnyUrl
import psycopg2
from psycopg2 import sql
import re
# ──────────────────────────────────────────────────────────────
# 配置
# ──────────────────────────────────────────────────────────────
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://localhost:5432/mydb")
READ_ONLY_USER = os.getenv("READONLY_USER", "readonly_user")
READ_ONLY_PASSWORD = os.getenv("READONLY_PASSWORD", "")
# 危险 SQL 关键词黑名单（多重防护）
FORBIDDEN_PATTERNS = [
r"\b(INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|TRUNCATE|GRANT|REVOKE|EXEC|EXECUTE)\b",
r";\s*\w+", # 多语句注入
]
# ──────────────────────────────────────────────────────────────
# 数据库连接池（实际生产应使用连接池）
# ──────────────────────────────────────────────────────────────
def get_db_connection():
"""创建只读数据库连接"""
return psycopg2.connect(
host=DATABASE_URL.split("@")[-1].split("/")[0].split(":")[0],
port=5432,
database=DATABASE_URL.split("/")[-1],
user=READ_ONLY_USER,
password=READ_ONLY_PASSWORD
)
def is_safe_query(query: str) -> bool:
"""安全检查：防止注入和危险操作"""
query_upper = query.upper()
# 检查危险关键词
for pattern in FORBIDDEN_PATTERNS:
if re.search(pattern, query_upper):
return False
# 检查注释注入
if "--" in query or "/*" in query:
return False
return True
# ──────────────────────────────────────────────────────────────
# MCP Server 实例
# ──────────────────────────────────────────────────────────────
server = Server("postgres-mcp-server")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""声明本 Server 提供的所有工具"""
return [
Tool(
name="query_database",
description=(
"执行只读 SQL SELECT 查询。"
"自动限制返回行数（最多1000行），防止大结果集。\n"
"适用于：数据分析、报表生成、数据探索。"
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL SELECT 查询语句"
},
"params": {
"type": "array",
"description": "查询参数（用于参数化查询，防止 SQL 注入）",
"items": {"type": "string"},
"default": []
},
"limit": {
"type": "integer",
"description": "最大返回行数",
"default": 100
}
},
"required": ["query"]
}
),
Tool(
name="list_tables",
description="列出数据库中所有表名及其注释",
inputSchema={
"type": "object",
"properties": {
"schema": {
"type": "string",
"description": "数据库 schema（默认 public）",
"default": "public"
}
}
}
),
Tool(
name="describe_table",
description="查看某个表的结构（列名、类型、约束、注释）",
inputSchema={
"type": "object",
"properties": {
"table_name": {
"type": "string",
"description": "表名"
},
"schema": {
"type": "string",
"description": "schema 名",
"default": "public"
}
},
"required": ["table_name"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""处理工具调用请求"""
if name == "query_database":
return await _handle_query(arguments)
elif name == "list_tables":
return await _handle_list_tables(arguments)
elif name == "describe_table":
return await _handle_describe_table(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
async def _handle_query(args: dict) -> list[TextContent]:
"""执行 SELECT 查询"""
query = args["query"]
limit = args.get("limit", 100)
# 安全检查
if not is_safe_query(query):
return [TextContent(
type="text",
text="❌ 安全检查失败：查询包含禁止的操作或潜在注入风险。"
)]
# 强制追加 LIMIT
if "LIMIT" not in query.upper():
query = f"{query.rstrip(';')} LIMIT {limit}"
try:
conn = get_db_connection()
cur = conn.cursor()
cur.execute(query)
# 获取列名和结果
columns = [desc[0] for desc in cur.description]
rows = cur.fetchall()
# 格式化为 Markdown 表格
header = "| " + " | ".join(columns) + " |"
separator = "|" + "|".join([" --- " for _ in columns]) + "|"
body_rows = [""| " + "|".join(str(cell) for cell in row) + "|" for row in rows]
result_md = (
f"✅ 查询成功，返回 {len(rows)} 行：\n\n"
f"{header}\n{separator}\n" + "\n".join(body_rows)
)
cur.close()
conn.close()
return [TextContent(type="text", text=result_md)]
except Exception as e:\n return [TextContent(type="text", text=f"❌ 查询失败: {str(e)}")]
async def _handle_list_tables(args: dict) -> list[TextContent]:
"""列出所有表"""
schema = args.get("schema", "public")
query = """
SELECT
t.table_name,
COALESCE(obj_description(t.tableoid), '') as description,
(SELECT COUNT(*) FROM information_schema.table_statistics ts
WHERE ts.table_schema = t.table_schema AND ts.table_name = t.table_name) as estimated_rows
FROM information_schema.tables t
WHERE t.table_schema = %s AND t.table_type = 'BASE TABLE'
ORDER BY t.table_name;
"""
try:
conn = get_db_connection()
cur = conn.cursor()
cur.execute(query, (schema,))
rows = cur.fetchall()
if not rows:
return [TextContent(type="text", text=f"Schema '{schema}' 中没有找到表。")]
lines = [f"## 📊 Schema: {schema} 中的表（共 {len(rows)} 张）\n"]
for table, desc, rows_cnt in rows:
lines.append(f"- **{table}** ({rows_cnt} 行) — {desc or '无注释'}")
cur.close()
conn.close()
return [TextContent(type="text", text="\n".join(lines))]
except Exception as e:\n return [TextContent(type="text", text=f"❌ 失败: {str(e)}")]
async def _handle_describe_table(args: dict) -> list[TextContent]:
"""查看表结构"""
table_name = args["table_name"]
schema = args.get("schema", "public")
# 使用参数化查询防止注入
query = """
SELECT
c.column_name,
c.data_type,
c.is_nullable,
c.column_default,
col_description((c.table_schema || '.' || c.table_name)::regclass, c.ordinal_position) as column_comment
FROM information_schema.columns c
WHERE c.table_schema = %s AND c.table_name = %s
ORDER BY c.ordinal_position;
"""
try:
conn = get_db_connection()
cur = conn.cursor()
cur.execute(query, (schema, table_name))
rows = cur.fetchall()
if not rows:
return [TextContent(type="text", text=f"❌ 表 '{schema}.{table_name}' 不存在。")]
lines = [f"## 📋 表结构: {schema}.{table_name}\n"]
lines.append("| 列名 | 类型 | 可空 | 默认值 | 注释 |")
lines.append("|---|---|---|---|---|")
for col_name, dtype, nullable, default, comment in rows:
lines.append(f"| {col_name} | {dtype} | {nullable} | {default or ''} | {comment or ''} |")
cur.close()
conn.close()
return [TextContent(type="text", text="\n".join(lines))]
except Exception as e:\n return [TextContent(type="text", text=f"❌ 失败: {str(e)}")]
# ──────────────────────────────────────────────────────────────
# 启动 Server（stdio 模式，供 Claude Desktop 使用）
# ──────────────────────────────────────────────────────────────
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
4.3 TypeScript 实现：一个文件系统 MCP Server
// server/filesystem-mcp-server.ts
/**
* MCP Server 示例：文件系统工具
* 功能：安全暴露本地文件读取、搜索、写入能力
*/
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
CallToolRequestSchema,
ListToolsRequestSchema,
ListResourcesRequestSchema,
ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs/promises";
import * as fsSync from "fs";
import * as path from "path";
import * as readline from "readline";
import { z } from "zod";
// ──────────────────────────────────────────────────────────────
// 安全配置
// ──────────────────────────────────────────────────────────────
const ALLOWED_DIRECTORIES = (process.env.ALLOWED_DIRS || process.cwd()).split(",");
const MAX_FILE_SIZE = parseInt(process.env.MAX_FILE_SIZE || "1048576"); // 1MB
// ──────────────────────────────────────────────────────────────
// 辅助函数
// ──────────────────────────────────────────────────────────────
function isPathAllowed(filePath: string): boolean {
const resolved = path.resolve(filePath);
return ALLOWED_DIRECTORIES.some(dir => resolved.startsWith(path.resolve(dir)));
}
function sanitizePath(filePath: string): string {
// 阻止路径遍历攻击
return path.normalize(filePath).replace(/^(\.\.[\/\\])+/, "");
}
// ──────────────────────────────────────────────────────────────
// MCP Server 初始化
// ──────────────────────────────────────────────────────────────
const server = new Server(
{
name: "filesystem-mcp-server",
version: "1.0.0",
},
{
capabilities: {
tools: {},
resources: {},
},
}
);
// ──────────────────────────────────────────────────────────────
// 工具列表声明
// ──────────────────────────────────────────────────────────────
server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: "read_file",
description: "读取单个文件的完整内容（推荐用于 < 1MB 的文本文件）",
inputSchema: {
type: "object",
properties: {
path: {
type: "string",
description: "文件路径（相对于工作目录）",
},
maxLines: {
type: "integer",
description: "最多读取行数（防止大文件）",
default: 1000,
},
},
required: ["path"],
},
},
{
name: "search_files",
description: "递归搜索文件内容（支持正则表达式）",
inputSchema: {
type: "object",
properties: {
directory: {
type: "string",
description: "搜索根目录",
},
pattern: {
type: "string",
description: "文件名或正则模式",
},
contentSearch: {
type: "string",
description: "在文件内容中搜索（可选）",
},
caseSensitive: {
type: "boolean",
default: false,
},
},
required: ["directory", "pattern"],
},
},
{
name: "write_file",
description: "创建或覆盖文件（谨慎使用，建议先读取确认）",
inputSchema: {
type: "object",
properties: {
path: {
type: "string",
description: "目标文件路径",
},
content: {
type: "string",
description: "文件内容",
},
},
required: ["path", "content"],
},
},
{
name: "list_directory",
description: "列出目录内容（包含文件类型和大小信息）",
inputSchema: {
type: "object",
properties: {
path: {
type: "string",
description: "目录路径",
},
recursive: {
type: "boolean",
description: "递归列出子目录",
default: false,
},
},
required: ["path"],
},
},
],
};
});
// ──────────────────────────────────────────────────────────────
// 资源列表（可被 AI 主动读取的只读数据）
// ──────────────────────────────────────────────────────────────
server.setRequestHandler(ListResourcesRequestSchema, async () => {
return {
resources: [
{
uri: "file://cwd",
name: "当前工作目录",
mimeType: "text/plain",
description: "当前工作目录的绝对路径",
},
],
};
});
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
const uri = request.params.uri;
if (uri === "file://cwd") {
return {
contents: [
{
uri,
mimeType: "text/plain",
text: process.cwd(),
},
],
};
}
throw new Error(`Unknown resource: ${uri}`);
});
// ──────────────────────────────────────────────────────────────
// 工具调用处理
// ──────────────────────────────────────────────────────────────
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
try {
switch (name) {
case "read_file": {
const { path: filePath, maxLines = 1000 } = args as {
path: string;
maxLines?: number;
};
if (!isPathAllowed(filePath)) {
return { content: [{ type: "text", text: `❌ 访问被拒绝：${filePath} 不在允许目录内。` }], isError: true };
}
const stats = await fs.stat(filePath);
if (stats.size > MAX_FILE_SIZE) {
return {
content: [{ type: "text", text: `❌ 文件过大（${stats.size} bytes），超过限制 ${MAX_FILE_SIZE} bytes。` }],
isError: true
};
}
const content = await fs.readFile(filePath, "utf-8");
const lines = content.split("\n").slice(0, maxLines);
const truncated = lines.length < content.split("\n").length;
return {
content: [{
type: "text",
text: `📄 ${filePath} (${stats.size} bytes)\n\`\`\`\n${lines.join("\n")}\n${truncated ? "```\n⚠️ 已截断（超过最大行数限制）" : "```"}`,
}],
};
}
case "search_files": {
const { directory, pattern, contentSearch, caseSensitive = false } = args as {
directory: string;
pattern: string;
contentSearch?: string;
caseSensitive?: boolean;
};
if (!isPathAllowed(directory)) {
return { content: [{ type: "text", text: "❌ 访问被拒绝。" }], isError: true };
}
const results: string[] = [];
async function search(dir: string, depth = 0): Promise<void> {
if (depth > 5) return; // 防止无限递归
try {
const entries = await fs.readdir(dir, { withFileTypes: true });
for (const entry of entries) {
const fullPath = path.join(dir, entry.name);
// 文件名匹配
const nameMatches = caseSensitive
? entry.name.includes(pattern)
: entry.name.toLowerCase().includes(pattern.toLowerCase());
if (entry.isFile() && nameMatches) {
if (contentSearch) {
// 内容搜索
const fileContent = await fs.readFile(fullPath, "utf-8").catch(() => "");
const contentMatch = caseSensitive
? fileContent.includes(contentSearch)
: fileContent.toLowerCase().includes(contentSearch.toLowerCase());
if (contentMatch) results.push(`✓ ${fullPath}`);
} else {
results.push(`📄 ${fullPath}`);
}
} else if (entry.isDirectory() && !entry.name.startsWith(".")) {
await search(fullPath, depth + 1);
}
}
} catch {
// 跳过无权限目录
}
}
await search(directory);
if (results.length === 0) {
return { content: [{ type: "text", text: `🔍 在 ${directory} 中未找到匹配 "${pattern}" 的文件。` }] };
}
return {
content: [{ type: "text", text: `🔍 找到 ${results.length} 个结果：\n${results.join("\n")}` }],
};
}
case "write_file": {
const { path: filePath, content } = args as { path: string; content: string };
if (!isPathAllowed(filePath)) {
return { content: [{ type: "text", text: "❌ 访问被拒绝。" }], isError: true };
}
await fs.writeFile(filePath, content, "utf-8");
return {
content: [{ type: "text", text: `✅ 文件已写入：${filePath} (${content.length} bytes)` }],
};
}
case "list_directory": {
const { path: dirPath, recursive = false } = args as { path: string; recursive?: boolean };
if (!isPathAllowed(dirPath)) {
return { content: [{ type: "text", text: "❌ 访问被拒绝。" }], isError: true };
}
const lines: string[] = [`📁 ${dirPath}`];
async function list(dir: string, indent = ""): Promise<void> {
try {
const entries = await fs.readdir(dir, { withFileTypes: true });
for (const entry of entries) {
if (entry.name.startsWith(".")) continue;
const icon = entry.isDirectory() ? "📂" : "📄";
const prefix = `${indent}${icon}`;
if (entry.isDirectory()) {
lines.push(`${prefix} ${entry.name}/`);
if (recursive) await list(path.join(dir, entry.name), indent + " ");
} else {
const stats = await fs.stat(path.join(dir, entry.name));
lines.push(`${prefix} ${entry.name} (${(stats.size / 1024).toFixed(1)} KB)`);
}
}
} catch {
lines.push(`${indent}❌ 无法读取`);
}
}
await list(dirPath);
return { content: [{ type: "text", text: lines.join("\n") }] };
}
default:
return {
content: [{ type: "text", text: `❌ 未知工具: ${name}` }],
isError: true,
};
}
} catch (error) {
return {
content: [{ type: "text", text: `❌ 错误: ${error instanceof Error ? error.message : String(error)}` }],
isError: true,
};
}
});
// ──────────────────────────────────────────────────────────────
// 启动
// ──────────────────────────────────────────────────────────────
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("📁 Filesystem MCP Server 已启动");
}
main().catch(console.error);
五、Claude Desktop 配置 MCP
5.1 安装 Claude Desktop
从 Anthropic 官网 下载 Claude Desktop 应用（支持 macOS 和 Windows）。
5.2 配置文件位置
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
5.3 配置示例
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": [
"C:\\path\\to\\filesystem-mcp-server\\dist\\index.js"
],
"env": {
"ALLOWED_DIRS": "C:\\Users\\zhiyu\\Projects,C:\\Users\\zhiyu\\Documents",
"MAX_FILE_SIZE": "2097152"
}
},
"postgres": {
"command": "python",
"args": [
"C:\\path\\to\\postgres_mcp_server.py"
],
"env": {
"DATABASE_URL": "postgresql://localhost:5432/mydb",
"READONLY_USER": "readonly_user",
"READONLY_PASSWORD": "your_password_here"
}
},
"github": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-github",
"--github-token",
"your_github_token_here"
]
},
"brave-search": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-brave-search",
"--brave-api-key",
"your_brave_api_key_here"
]
}
}
}
5.4 常用官方 MCP Server 一键安装
| Server | 命令 | 功能 |
|---|---|---|
| GitHub | npx -y @modelcontextprotocol/server-github |
仓库操作、Issue、PR |
| Brave Search | npx -y @modelcontextprotocol/server-brave-search |
网页搜索 |
| Filesystem | npx -y @modelcontextprotocol/server-filesystem |
本地文件访问 |
| Slack | npx -y @modelcontextprotocol/server-slack |
消息发送/读取 |
| Google Maps | npx -y @modelcontextprotocol/server-google-maps |
地图、地点、路线 |
5.5 验证配置是否生效
在 Claude Desktop 中发送：
请列出你当前可用的所有工具。
如果 MCP Server 配置正确，Claude 会自动调用 tools/list 并展示所有可用工具的名称和描述。
六、Claude AI 连接本地资源
6.1 通过 MCP 连接本地数据库
场景： 在 Claude.ai 网页版直接用自然语言查询本地 PostgreSQL 数据库。
实现原理：
用户（Claude.ai网页） → MCP Gateway（本地代理） → MCP Server（PostgreSQL）
↑
stdio/WebSocket
方案一：使用 MCP Gateway 代理
# 安装 mcp-gateway
npm install -g @modelcontextprotocol/gateway
# 启动 gateway（将 stdio MCP Server 代理为 HTTP）
mcp-gateway --port 8080 --transport stdio -- python postgres_mcp_server.py
方案二：通过 ngrok 内网穿透（推荐个人使用）
# gateway_with_ngrok.py
"""将本地 MCP Server 通过 ngrok 暴露给远程 Claude"""
import asyncio
import subprocess
import httpx
import json
async def start_mcp_server():
"""启动本地 PostgreSQL MCP Server"""
process = subprocess.Popen(
["python", "postgres_mcp_server.py"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
return process
async def forward_to_ngrok():
"""通过 ngrok 将本地端口暴露到公网"""
# 启动 ngrok
ngrok_proc = subprocess.Popen(
["ngrok", "http", "8080", "--log", "stdout"],
stdout=subprocess.PIPE
)
# 获取 ngrok 分配的公网地址
async with httpx.AsyncClient() as client:
await asyncio.sleep(3) # 等待 ngrok 启动
resp = await client.get("http://localhost:4040/api/tunnels")
tunnels = resp.json()["tunnels"]
https_url = next(t["public_url"] for t in tunnels if t["proto"] == "https")
print(f"🔗 公开地址: {https_url}")
return https_url
# 使用说明：
# 1. 运行此脚本获取 https://xxx.ngrok.io
# 2. 在 Claude Desktop 中配置自定义 MCP Server 指向此地址
6.2 通过 MCP 连接本地文件系统
Claude Desktop 已内置文件系统支持，无需额外配置。在 claude_desktop_config.json 中添加：
{
"mcpServers": {
"local-files": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem",
"C:\\Users\\zhiyu\\Projects",
"C:\\Users\\zhiyu\\workspace"
]
}
}
}
6.3 通过 MCP 连接浏览器
# 需要 Chrome 浏览器 + Chrome DevTools Protocol
# 安装 chrome-mcp 或 playwright-based server
npm install -g @modelcontextprotocol/server-chrome
{
"mcpServers": {
"chrome": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-chrome"]
}
}
}
可用浏览器操作：
- 打开/关闭标签页
- 获取当前页面截图
- 执行 JavaScript
- 填写表单、点击元素
- 获取页面 HTML 内容
七、Multi-Agent 多Agent协同
7.1 传统 Multi-Agent 的痛点
在没有 MCP 的情况下，多 Agent 通信通常是硬编码的：
# 传统方式：硬编码 Agent 间协议
class AgentA:
def call_agent_b(self, task):
# Agent A 必须知道 Agent B 的 API 接口
return requests.post("http://agent-b:8001/analyze", json={"task": task})
class AgentB:
def __init__(self):
# Agent B 必须硬编码 Agent A 的地址
self.upstream = "http://agent-a:8000"
问题：
- 每增加一个 Agent，需要修改所有其他 Agent 的代码
- 无统一的消息格式
- 难以动态添加/移除 Agent
7.2 基于 MCP 的 Multi-Agent 架构
┌─────────────────────────────────────────────────────────────┐
│ MCP-Based Multi-Agent System │
│ │
│ ┌──────────┐ MCP Server (作为消息总线) ┌──────────┐ │
│ │ Agent A │ ◀─────────────────────────────▶ │ Agent B │ │
│ │Researcher│ tools/call: "send_message" │ Analyst │ │
│ └──────────┘ └──────────┘ │
│ ↑ │
│ │ tools/call: "store_knowledge" │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ MCP Server: Shared Knowledge Base │ │
│ │ (Resources + Prompts + 工具) │ │
│ └─────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ tools/call: "query_knowledge" │
│ ▼ │
│ ┌──────────┐ ┌──────────┐ │
│ │ Agent C │ ────▶ MCP Server ◀───────────│ Agent D │ │
│ │ Coder │ (Orchestrator) │ Writer │ │
│ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
7.3 Agent 间通信 MCP Server 实现
# server/agent_communication_server.py
"""
多 Agent 通信总线 MCP Server
功能：作为 Agent 间消息传递和知识共享的中央协调器
"""
import uuid
import asyncio
from datetime import datetime
from typing import Any
from collections import defaultdict
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource
server = Server("multi-agent-communication")
# ──────────────────────────────────────────────────────────────
# 内存存储（生产环境请使用 Redis/数据库）
# ──────────────────────────────────────────────────────────────
message_queues: dict[str, list[dict]] = defaultdict(list)
knowledge_base: dict[str, dict] = {}
agent_registry: dict[str, dict] = {}
# ──────────────────────────────────────────────────────────────
# 工具定义
# ──────────────────────────────────────────────────────────────
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="register_agent",
description="注册一个 Agent 到通信总线",
inputSchema={
"type": "object",
"properties": {
"agent_id": {"type": "string"},
"role": {"type": "string", "enum": ["researcher", "analyst", "coder", "writer", "reviewer"]},
"description": {"type": "string"}
},
"required": ["agent_id", "role"]
}
),
Tool(
name="send_message",
description="向另一个 Agent 发送消息",
inputSchema={
"type": "object",
"properties": {
"to_agent": {"type": "string"},
"content": {"type": "string"},
"priority": {"type": "string", "enum": ["low", "normal", "high"], "default": "normal"}
},
"required": ["to_agent", "content"]
}
),
Tool(
name="broadcast",
description="向所有已注册的 Agent 广播消息",
inputSchema={
"type": "object",
"properties": {
"content": {"type": "string"},
"from_agent": {"type": "string"}
},
"required": ["content", "from_agent"]
}
),
Tool(
name="get_messages",
description="获取发给当前 Agent 的所有未读消息",
inputSchema={
"type": "object",
"properties": {
"agent_id": {"type": "string"},
"limit": {"type": "integer", "default": 50}
},
"required": ["agent_id"]
}
),
Tool(
name="store_knowledge",
description="存储结构化知识到共享知识库",
inputSchema={
"type": "object",
"properties": {
"key": {"type": "string", "description": "知识唯一标识"},
"value": {"type": "string", "description": "知识内容"},
"tags": {"type": "array", "items": {"type": "string"}}
},
"required": ["key", "value"]
}
),
Tool(
name="query_knowledge",
description="从共享知识库检索知识",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string"},
"tags": {"type": "array", "items": {"type": "string"}}
}
}
),
]
@server.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri=f"agent://{agent_id}",
name=f"Agent: {info['role']}",
description=info.get("description", ""),
mimeType="application/json"
)
for agent_id, info in agent_registry.items()
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
handler = {
"register_agent": _handle_register,
"send_message": _handle_send_message,
"broadcast": _handle_broadcast,
"get_messages": _handle_get_messages,
"store_knowledge": _handle_store_knowledge,
"query_knowledge": _handle_query_knowledge,
}.get(name)
if not handler:
raise ValueError(f"Unknown tool: {name}")
return await handler(arguments)
async def _handle_register(args: dict) -> list[TextContent]:
agent_id = args["agent_id"]
agent_registry[agent_id] = {
"role": args["role"],
"description": args.get("description", ""),
"registered_at": datetime.now().isoformat()
}
return [TextContent(
type="text",
text=f"✅ Agent '{agent_id}' (role: {args['role']}) 已注册到通信总线。"
)]
async def _handle_send_message(args: dict) -> list[TextContent]:
msg = {
"id": str(uuid.uuid4()),
"from": "anonymous", # MCP 调用方自己标识
"to": args["to_agent"],
"content": args["content"],
"priority": args.get("priority", "normal"),
"timestamp": datetime.now().isoformat()
}
message_queues[args["to_agent"]].append(msg)
return [TextContent(
type="text",
text=f"📨 消息已发送给 {args['to_agent']}，优先级: {msg['priority']}，消息ID: {msg['id']}"
)]
async def _handle_broadcast(args: dict) -> list[TextContent]:
recipients = [aid for aid in agent_registry.keys() if aid != args["from_agent"]]
for agent_id in recipients:
msg = {
"id": str(uuid.uuid4()),
"from": args["from_agent"],
"to": "broadcast",
"content": args["content"],
"timestamp": datetime.now().isoformat()
}
message_queues[agent_id].append(msg)
return [TextContent(
type="text",
text=f"📢 已广播给 {len(recipients)} 个 Agent"
)]
async def _handle_get_messages(args: dict) -> list[TextContent]:
messages = message_queues.get(args["agent_id"], [])[:args.get("limit", 50)]
message_queues[args["agent_id"]] = [] # 清空已读
if not messages:
return [TextContent(type="text", text="📭 没有未读消息")]
lines = [f"📬 {len(messages)} 条消息：\n"]
for msg in messages:
lines.append(f"- [{msg['timestamp']}] {msg['from']} → {msg['to']}: {msg['content'][:100]}")
return [TextContent(type="text", text="\n".join(lines))]
async def _handle_store_knowledge(args: dict) -> list[TextContent]:
knowledge_base[args["key"]] = {
"value": args["value"],
"tags": args.get("tags", []),
"stored_at": datetime.now().isoformat()
}
return [TextContent(type="text", text=f"💾 知识 '{args['key']}' 已存储")]
async def _handle_query_knowledge(args: dict) -> list[TextContent]:
results = []
for key, data in knowledge_base.items():
if args.get("tags"):
if any(tag in data["tags"] for tag in args["tags"]):
results.append(f"**{key}** [{', '.join(data['tags'])}]: {data['value']}")
elif args.get("query"):
if args["query"].lower() in data["value"].lower() or args["query"].lower() in key.lower():
results.append(f"**{key}**: {data['value']}")
if not results:
return [TextContent(type="text", text="🔍 知识库中未找到匹配结果")]
return [TextContent(type="text", text=f"🔍 找到 {len(results)} 条知识：\n" + "\n".join(results))]
# ──────────────────────────────────────────────────────────────
# 启动
# ──────────────────────────────────────────────────────────────
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
7.4 Multi-Agent 协作工作流示例
工作流：研究 → 分析 → 编码 → 评审
Step 1: Researcher Agent
→ 调用 send_message(to="analyst", content="请分析这份数据报告的结论")
Step 2: Analyst Agent
→ 接收消息，执行分析
→ 调用 store_knowledge(key="finding_2026q2", value="用户留存率下降12%")
→ 调用 send_message(to="coder", content="需要修复用户留存统计模块")
Step 3: Coder Agent
→ 接收消息，编写代码修复
→ 调用 send_message(to="reviewer", content="代码已提交，请审查")
Step 4: Reviewer Agent
→ 调用 query_knowledge(query="留存率") 获取背景
→ 审查代码，给出反馈
八、完整项目：用MCP打造私有知识库AI助手
8.1 项目概述
项目名称：KnowledgeBase AI Assistant
目标：构建一个可以在本地知识库中进行语义检索、问答、内容生成的 AI 助手
核心技术栈：LangChain + ChromaDB + MCP + Claude
8.2 架构设计
┌─────────────────────────────────────────────────────────────┐
│ 私有知识库 AI 助手 架构 │
│ │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ LangChain Agent │ │
│ │ (编排层：意图识别 + 工具调度 + 答案生成) │ │
│ └──────────────┬──────────────────────┬──────────────────┘ │
│ │ │ │
│ MCP Client MCP Client │
│ (RAG工具) (知识管理) │
│ │ │ │
│ ┌──────────────▼──────────────────────▼──────────────────┐ │
│ │ MCP Server Layer │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ RAG Search │ │ Doc Manager │ │ Web Search │ │ │
│ │ │ Server │ │ Server │ │ Server │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ └─────────┼────────────────┼────────────────┼───────────┘ │
│ │ │ │ │
│ ┌─────────▼────────────────▼────────────────▼──────────┐ │
│ │ ChromaDB + 本地文档 + Web │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
8.3 完整代码
第一部分：RAG MCP Server
# server/rag_mcp_server.py
"""
RAG (检索增强生成) MCP Server
功能：基于向量数据库的知识检索工具
"""
import os
import hashlib
import asyncio
from typing import Any, Optional
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource
try:
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader, TextLoader
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
# ──────────────────────────────────────────────────────────────
# 向量数据库初始化
# ──────────────────────────────────────────────────────────────
PERSIST_DIRECTORY = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
_vectorstore: Optional[Chroma] = None
def get_vectorstore() -> Chroma:
global _vectorstore
if _vectorstore is None:
if not LANGCHAIN_AVAILABLE:
raise ImportError("langchain_community is required. Run: pip install langchain-community chromadb sentence-transformers")
embedding = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)
_vectorstore = Chroma(
persist_directory=PERSIST_DIRECTORY,
embedding_function=embedding
)
return _vectorstore
# ──────────────────────────────────────────────────────────────
# MCP Server
# ──────────────────────────────────────────────────────────────
server = Server("rag-knowledgebase-server")
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="index_documents",
description=(
"将本地目录中的文档（.txt, .md, .pdf, .docx）索引到向量数据库。"
"支持增量索引：已索引文件会跳过。"
),
inputSchema={
"type": "object",
"properties": {
"directory": {
"type": "string",
"description": "文档所在目录路径"
},
"collection_name": {
"type": "string",
"description": "向量集合名称（用于隔离不同知识库）",
"default": "default"
}
},
"required": ["directory"]
}
),
Tool(
name="semantic_search",
description=(
"对知识库进行语义检索（不是关键词匹配，是理解意图的搜索）。"
"返回最相关的文档片段及其来源。"
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "自然语言查询"
},
"top_k": {
"type": "integer",
"description": "返回最相关的片段数量",
"default": 5
},
"collection_name": {
"type": "string",
"default": "default"
}
},
"required": ["query"]
}
),
Tool(
name="add_knowledge",
description="手动向知识库添加单条知识（直接指定文本，不从文件读取）",
inputSchema={
"type": "object",
"properties": {
"content": {"type": "string", "description": "知识内容文本"},
"source": {"type": "string", "description": "来源标注"},
"metadata": {"type": "object", "description": "额外元数据"},
"collection_name": {"type": "string", "default": "default"}
},
"required": ["content", "source"]
}
),
Tool(
name="get_stats",
description="查看知识库统计信息（文档数、集合列表）",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="delete_by_source",
description="删除指定来源的所有知识条目（用于更新过期内容）",
inputSchema={
"type": "object",
"properties": {
"source": {"type": "string"},
"collection_name": {"type": "string", "default": "default"}
},
"required": ["source"]
}
),
]
@server.list_resources()
async def list_resources() -> list[Resource]:
vs = get_vectorstore()
return [
Resource(
uri="kb://stats",
name="知识库统计",
mimeType="application/json",
description=f"当前知识库共 {vs._collection.count()} 条记录"
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
if name == "index_documents":
return await _handle_index_documents(arguments)
elif name == "semantic_search":
return await _handle_semantic_search(arguments)
elif name == "add_knowledge":
return await _handle_add_knowledge(arguments)
elif name == "get_stats":
return await _handle_get_stats(arguments)
elif name == "delete_by_source":
return await _handle_delete_by_source(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
async def _handle_index_documents(args: dict) -> list[TextContent]:
directory = args["directory"]
collection_name = args.get("collection_name", "default")
if not os.path.exists(directory):
return [TextContent(type="text", text=f"❌ 目录不存在: {directory}")]
# 加载文档
try:
loader = DirectoryLoader(
directory,
glob="**/*.{txt,md}",
loader_cls=TextLoader,
show_progress=True
)
documents = loader.load()
except Exception as e:\n return [TextContent(type="text", text=f"❌ 加载文档失败: {str(e)}}")]
if not documents:
return [TextContent(type="text", text="❌ 未找到任何文档")]
# 文本分块
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
length_function=len
)
chunks = text_splitter.split_documents(documents)
# 为每个 chunk 添加元数据
for i, chunk in enumerate(chunks):
chunk.metadata.update({
"chunk_id": hashlib.md5(chunk.page_content.encode()).hexdigest()[:8],
"indexed_at": datetime.now().isoformat(),
"chunk_index": i
})
# 存入向量数据库
vs = get_vectorstore()
vs.add_documents(chunks)
return [TextContent(
type="text",
text=f"✅ 成功索引 {len(documents)} 个文档，切分为 {len(chunks)} 个知识块"
)]
async def _handle_semantic_search(args: dict) -> list[TextContent]:
query = args["query"]
top_k = args.get("top_k", 5)
vs = get_vectorstore()
results = vs.similarity_search_with_score(query, k=top_k)
if not results:
return [TextContent(type="text", text="🔍 知识库中未找到相关内容")]
lines = [f"🔍 找到 {len(results)} 条相关知识：\n"]
for i, (doc, score) in enumerate(results, 1):
similarity_pct = max(0, (1 - score) * 100)
source = doc.metadata.get("source", "未知来源")
lines.append(
f"\n--- 结果 {i} [相似度: {similarity_pct:.1f}%] ---\n"
f"📂 来源: {source}\n"
f"📝 内容:\n{doc.page_content[:300]}"
+ ("..." if len(doc.page_content) > 300 else "")
)
return [TextContent(type="text", text="\n".join(lines))]
async def _handle_add_knowledge(args: dict) -> list[TextContent]:
from langchain.schema import Document
content = args["content"]
source = args["source"]
metadata = args.get("metadata", {})
collection_name = args.get("collection_name", "default")
doc = Document(
page_content=content,
metadata={
**metadata,
"source": source,
"added_at": datetime.now().isoformat(),
"chunk_id": hashlib.md5(content.encode()).hexdigest()[:8]
}
)
vs = get_vectorstore()
vs.add_documents([doc])
return [TextContent(type="text", text=f"✅ 知识已添加: [{source}] {content[:50]}...")]
async def _handle_get_stats(args: dict) -> list[TextContent]:
vs = get_vectorstore()
count = vs._collection.count()
return [TextContent(
type="text",
text=f"📊 知识库统计：\n- 总知识块数: {count}\n- 向量维度: 384\n- 嵌入模型: {EMBEDDING_MODEL}\n- 存储路径: {PERSIST_DIRECTORY}"
)]
async def _handle_delete_by_source(args: dict) -> list[TextContent]:
source = args["source"]
vs = get_vectorstore()
# 通过元数据过滤删除
deleted_count = vs._collection.delete(
where={"source": {"$eq": source}}
)
return [TextContent(type="text", text=f"✅ 已删除来源 '{source}' 的所有知识")]
# ──────────────────────────────────────────────────────────────
# 启动
# ──────────────────────────────────────────────────────────────
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
第二部分：LangChain Agent 编排层
# knowledge_base_agent.py
"""
LangChain Agent：整合 MCP 工具，打造私有知识库 AI 助手
"""
import os
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, SystemMessage
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_openai import ChatOpenAI
# ──────────────────────────────────────────────────────────────
# MCP 工具适配器（LangChain → MCP）
# ──────────────────────────────────────────────────────────────
from langchain_mcp_adapters.tools import load_mcp_server_tools
from mcp import ClientSession, StdioServerParameters
# ──────────────────────────────────────────────────────────────
# 系统提示词
# ──────────────────────────────────────────────────────────────
SYSTEM_PROMPT = """你是一个专业的私有知识库助手。
你的能力：
1. 🔍 语义搜索：当用户询问知识库相关内容时，使用 semantic_search 工具检索最相关的知识
2. 📚 知识管理：使用 add_knowledge 添加新知识，使用 get_stats 查看知识库状态
3. 📂 文档索引：使用 index_documents 将本地文档索引到知识库
4. 🌐 网络搜索：当知识库无法回答时，使用 web_search 进行补充
回答原则：
- 优先使用知识库中的知识回答问题，并标注来源
- 如果知识库检索结果不完整，可以补充网络搜索
- 不知道的问题明确告知，不要编造
- 使用中文回答，格式清晰，适当使用 Markdown 格式化
"""
# ──────────────────────────────────────────────────────────────
# MCP Server 配置
# ──────────────────────────────────────────────────────────────
MCP_SERVERS = {
"rag": {
"command": "python",
"args": [os.path.join(os.path.dirname(__file__), "server", "rag_mcp_server.py")],
"env": {
"CHROMA_PERSIST_DIR": "./chroma_db",
"EMBEDDING_MODEL": "sentence-transformers/all-MiniLM-L6-v2"
}
},
"filesystem": {
"command": "node",
"args": [os.path.join(os.path.dirname(__file__), "server", "filesystem_mcp_server.js")],
"env": {
"ALLOWED_DIRS": os.path.join(os.path.dirname(__file__), "knowledge_base")
}
}
}
# ──────────────────────────────────────────────────────────────
# 加载 MCP 工具
# ──────────────────────────────────────────────────────────────
async def load_tools():
"""从所有 MCP Server 加载可用工具"""
all_tools = []
for name, config in MCP_SERVERS.items():
try:
server_params = StdioServerParameters(
command=config["command"],
args=config["args"],
env=config.get("env", {})
)
async with ClientSession(server_params) as session:
await session.initialize()
tools = await load_mcp_server_tools(session)
all_tools.extend(tools)
print(f"✅ 已从 {name} MCP Server 加载 {len(tools)} 个工具")
except Exception as e:\n print(f"⚠️ MCP Server '{name}' 加载失败: {e}")
return all_tools
# ──────────────────────────────────────────────────────────────
# 创建 Agent
# ──────────────────────────────────────────────────────────────
def create_kb_agent(tools, llm: ChatOpenAI):
"""创建知识库 Agent"""
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True
)
return agent_executor
# ──────────────────────────────────────────────────────────────
# 交互式 CLI
# ──────────────────────────────────────────────────────────────
async def run_cli():
"""启动交互式命令行界面"""
import asyncio
print("=" * 60)
print("🧠 私有知识库 AI 助手")
print("=" * 60)
print("提示：输入 'quit' 退出，'stats' 查看知识库状态")
print()
# 加载工具
tools = await load_tools()
if not tools:
print("❌ 未能加载任何 MCP 工具，请检查配置")
return
print(f"\n已加载 {len(tools)} 个工具，准备就绪！\n")
# 初始化 LLM
llm = ChatOpenAI(
model=os.getenv("OPENAI_MODEL", "gpt-4o"),
api_key=os.getenv("OPENAI_API_KEY"),
temperature=0.7
)
# 创建 Agent
agent_executor = create_kb_agent(tools, llm)
chat_history = []
while True:
try:
user_input = input("👤 你: ").strip()
if not user_input:
continue
if user_input.lower() in ["quit", "exit", "退出"]:
print("👋 再见！")
break
if user_input.lower() == "stats":
# 直接调用 stats 工具
from langchain_core.tools import tool
# 这里简化处理，实际应该从 tools 中找到 stats 工具
print("📊 请通过对话询问，我会查询知识库状态")
continue
# 调用 Agent
response = await agent_executor.ainvoke({
"input": user_input,
"chat_history": chat_history
})
print(f"\n🤖 AI: {response['output']}\n")
# 更新历史
chat_history.append(HumanMessage(content=user_input))
chat_history.append(SystemMessage(content=response["output"]))
except KeyboardInterrupt:
print("\n👋 再见！")
break
except Exception as e:\n print(f"\n❌ 错误: {e}\n")
# ──────────────────────────────────────────────────────────────
# 入口
# ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
import asyncio
asyncio.run(run_cli())
第三部分：Claude Desktop 配置
{
"mcpServers": {
"rag-knowledgebase": {
"command": "python",
"args": ["C:\\path\\to\\project\\server\\rag_mcp_server.py"],
"env": {
"CHROMA_PERSIST_DIR": "C:\\path\\to\\project\\chroma_db",
"EMBEDDING_MODEL": "sentence-transformers/all-MiniLM-L6-v2"
}
},
"knowledge-base-files": {
"command": "node",
"args": ["C:\\path\\to\\project\\server\\filesystem_mcp_server.js"],
"env": {
"ALLOWED_DIRS": "C:\\path\\to\\project\\knowledge_base"
}
}
}
}
8.4 项目目录结构
knowledge-base-assistant/
├── server/
│ ├── rag_mcp_server.py # RAG 检索服务
│ ├── filesystem_mcp_server.ts # 文件系统服务
│ └── agent_communication_server.py # 多 Agent 通信
├── knowledge_base/ # 知识库文档目录
│ ├── docs/ # 待索引的文档
│ └── uploads/ # 用户上传的文档
├── chroma_db/ # 向量数据库存储
├── knowledge_base_agent.py # LangChain Agent 主程序
├── requirements.txt
└── README.md
8.5 快速启动
# 1. 安装依赖
pip install -r requirements.txt
# 2. 设置环境变量
export OPENAI_API_KEY="sk-..."
export EMBEDDING_MODEL="sentence-transformers/all-MiniLM-L6-v2"
# 3. 添加知识到知识库
echo "MCP协议是Anthropic开源的AI工具连接标准协议。" > knowledge_base/intro.txt
# 4. 启动交互式助手
python knowledge_base_agent.py
# 5. 在 Claude Desktop 中配置后，直接对话
九、常见问题与避坑指南
9.1 MCP Server 开发常见问题
Q1: Server 启动成功但 Claude Desktop 检测不到工具？
# 调试步骤
# 1. 确认配置文件路径正确（Windows 是 %APPDATA%\Claude\claude_desktop_config.json）
# 2. 检查 command 和 args 路径是否正确
# 3. 用命令行手动测试 Server 是否正常：
python your_mcp_server.py
# 如果报错，错误信息会显示在 stderr
# 4. 查看 Claude Desktop 日志：
# macOS: ~/Library/Logs/Claude/
# Windows: %APPDATA%\Claude\logs\
Q2: JSON-RPC 消息格式错误？
# ✅ 正确：使用 MCP SDK 提供的方法
from mcp.types import TextContent
return [TextContent(type="text", text="result")]
# ❌ 错误：直接返回字典
return {"content": "result"} # MCP Client 无法解析
Q3: async/await 混淆导致死锁？
MCP Server 的所有 handler 必须是 async 函数：
# ✅
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
...
# ❌ 普通函数会导致事件循环阻塞
@server.call_tool()
def call_tool(name: str, arguments: Any):
...
9.2 安全注意事项
⚠️ 安全红线：
1. 【必需】不要在 MCP Server 中暴露管理员权限
→ 使用只读用户、最小权限原则
2. 【必需】严格校验文件路径，防止路径遍历
→ realpath() / isPathAllowed() / 禁止 ../
3. 【必需】SQL 查询必须参数化 + 只支持 SELECT
→ 防止 SQL 注入
4. 【必需】不要在工具描述中泄露敏感信息
→ 工具的 description 字段可能暴露给 AI 和用户
5. 【建议】添加请求频率限制
→ 防止恶意刷接口
9.3 性能优化建议
| 场景 | 优化方案 |
|---|---|
| 大量文件索引 | 使用批处理 + asyncio 并行处理 |
| 向量查询慢 | 使用更小的 embedding 模型 / 添加索引 |
| Server 启动慢 | 使用预热机制（lazy init） |
| 连接数过多 | 使用 HTTP/SSE 模式复用连接 |
总结与展望
MCP 的意义
MCP 不仅仅是一个技术协议，它是 AI 从"单体应用"走向"生态互联"的关键基础设施：
单体 AI 时代 → MCP 互联时代
───────────────── ─────────────────
模型 + 固定工具 → 模型 + 可插拔工具生态
开发者定制集成 → 用户自由组合
封闭的信息孤岛 → 开放的知识网络
技术演进方向
根据 MCP 社区的 Roadmap，未来值得关注的方向：
- MCP Registry：类似 npm 的 MCP Server 包管理仓库
- MCP Gateway：支持远程 MCP Server 的标准代理方案
- 双向工具调用：Server 也能主动调用 Host 的能力
- 标准化身份认证：跨平台的 MCP 认证框架
- 多模型支持：同一套 MCP 工具适配不同模型提供商
学习资源
| 资源 | 链接 |
|---|---|
| MCP 官方文档 | https://modelcontextprotocol.io |
| MCP SDK (Python) | pip install mcp |
| MCP SDK (TypeScript) | @modelcontextprotocol/sdk |
| 官方 Server 示例 | github.com/modelcontextprotocol/servers |
| MCP Python 指南 | modelcontextprotocol.io/docs/python-server |
📌 写在最后
MCP 协议正处于快速迭代期（2026年已有大量新特性），建议读者关注官方仓库的 Release Notes，及时跟进协议更新。本文中所有代码均基于截至 2026年7月 的 MCP SDK 版本，部分 API 可能随版本更新发生变化。
如果本文对你有帮助，欢迎在评论区交流你的 MCP 实践心得！
© 2026 技术分享，版权所有，转载请注明出处
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
0
0- 0
已为社区贡献18条内容
所有评论(0)