MCP（Model Context Protocol）模型上下文协议开发完全教程
适用读者：有基础编程经验的开发者，希望深入理解并实践 MCP 协议的 AI 应用开发者、工具链构建者。
预计阅读时间：30-40 分钟
目录
- MCP 协议概述与发展背景
- MCP 核心架构：Host / Client / Server
- MCP 协议通信机制：JSON-RPC 2.0
- 三大核心能力：Resources、Tools、Prompts
- MCP Server 开发实战
- MCP Client 集成开发
- 常见 MCP Server 案例
- MCP 与 Function Calling 的区别和联系
- 安全考虑与最佳实践
- MCP 生态系统与未来展望
1. MCP 协议概述与发展背景
1.1 什么是 MCP
MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 于 2024 年底发布的一项开放协议标准，旨在为大语言模型（LLM）与外部数据源、工具之间建立统一的连接方式。
你可以将 MCP 理解为 AI 应用的 USB-C 接口——正如 USB-C 让各种外设可以用同一种线缆连接到不同设备，MCP 让各种 AI 模型可以用同一种协议接入不同的工具和数据源。
1.2 为什么需要 MCP
在 MCP 出现之前，AI 应用开发面临一个核心痛点：集成碎片化。
假设你正在构建一个 AI 助手，需要让它能够：
- 读取本地文件系统
- 查询数据库
- 调用第三方 API（如 GitHub、Slack）
- 访问内部知识库
每一种集成都需要编写定制化的代码，处理不同的认证方式、数据格式和错误逻辑。如果你要支持多个 LLM 提供商（OpenAI、Anthropic、Google），每家的 Function Calling 接口还不完全兼容，工作量会成倍增长。
MCP 的核心价值在于：
| 痛点 | MCP 解决方案 |
|---|---|
| 每个数据源都需要定制集成 | 统一协议，一次开发，多处复用 |
| 不同 LLM 的工具调用接口不兼容 | 标准化的工具描述和调用规范 |
| 上下文传递缺乏规范 | 定义了 Resources 和 Prompts 的标准格式 |
| 安全模型不一致 | 内置权限控制和安全边界 |
1.3 MCP 的发展脉络
- 2024 年 11 月：Anthropic 开源发布 MCP 规范和 Python SDK
- 2024 年 12 月：TypeScript SDK 发布，社区开始构建 MCP Server 生态
- 2025 年初：OpenAI、Google 等厂商开始支持或兼容 MCP
- 2025 年中：MCP 注册中心（Registry）和标准化工具链逐步完善
2. MCP 核心架构：Host / Client / Server
MCP 采用经典的三层架构，角色清晰，职责分明：
┌─────────────────────────────────────────────┐
│ Host │
│ (Claude Desktop, IDE, 自定义应用) │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ MCP Client A │ │ MCP Client B │ │
│ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │
└─────────┼────────────────────┼────────────────┘
│ │
┌─────▼─────┐ ┌────▼──────┐
│ MCP Server │ │ MCP Server │
│ (文件系统) │ │ (数据库) │
└───────────┘ └───────────┘
2.1 Host（宿主层）
Host 是用户的交互入口，负责：
- 管理一个或多个 MCP Client 实例
- 处理用户界面和交互逻辑
- 控制权限和安全策略
- 协调 LLM 与工具之间的对话流
典型的 Host 包括：Claude Desktop App、Cursor IDE、自建的 AI 应用等。
2.2 Client（客户端层）
Client 是 Host 与 Server 之间的桥梁，每个 Client 与一个 Server 保持一对一的连接。Client 的职责包括：
- 维护与 Server 的会话状态
- 处理协议握手和能力协商
- 将 Server 暴露的能力（Resources、Tools、Prompts）注册到 Host
- 转发请求和响应
2.3 Server（服务端层）
Server 是能力的实际提供者，专注于：
- 暴露具体的工具（Tools）、资源（Resources）和提示模板（Prompts）
- 处理来自 Client 的请求
- 管理与底层数据源或服务的连接
Server 的设计原则是轻量、专注、可组合——每个 Server 通常只负责一个领域（如文件操作、数据库查询、API 调用）。
3. MCP 协议通信机制：JSON-RPC 2.0
MCP 基于 JSON-RPC 2.0 协议进行通信，这是一个成熟、轻量的远程过程调用规范。
3.1 三种消息类型
① Request（请求）
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/home/user/document.txt"
}
}
}
② Response（响应）
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "文件内容..."
}
]
}
}
③ Notification（通知，无 id，无需响应）
{
"jsonrpc": "2.0",
"method": "notifications/resources/updated",
"params": {
"uri": "file:///home/user/document.txt"
}
}
3.2 传输层
MCP 支持两种标准传输方式：
① stdio（标准输入/输出）
适用于本地进程间通信，Client 启动 Server 作为子进程，通过 stdin/stdout 交换 JSON-RPC 消息。
Client ──stdin──► Server (子进程)
Client ◄──stdout── Server
这是最常用的传输方式，适合本地工具集成。
② HTTP + SSE（Server-Sent Events）
适用于远程通信，Client 通过 HTTP POST 发送请求，Server 通过 SSE 推送响应和通知。
Client ──HTTP POST──► Server (远程)
Client ◄──SSE stream── Server
适合需要网络访问的场景，如团队共享的 MCP Server。
3.3 生命周期
MCP 连接的完整生命周期如下：
Client Server
│ │
│──── initialize ──────────────►│ 能力协商
│◄─── initialize result ────────│
│──── initialized (通知) ───────►│ 确认连接
│ │
│ ═══ 正常通信阶段 ═══ │
│ │
│──── tools/list ──────────────►│ 发现能力
│──── tools/call ──────────────►│ 调用工具
│◄─── resources/list ───────────│ 获取资源
│ │
│──── shutdown ────────────────►│ 优雅关闭
│◄─── exit ─────────────────────│
4. 三大核心能力：Resources、Tools、Prompts
4.1 Resources（资源）
Resources 是 Server 向 Client 暴露的只读数据，类似于 REST API 中的 GET 端点。
用途：为 LLM 提供上下文信息，如文件内容、数据库记录、API 响应等。
资源标识：使用 URI 格式，如：
file:///path/to/document.txtpostgres://database/customersgithub://repo/owner/repo/issues
资源定义示例：
{
"uri": "file:///home/user/config.json",
"name": "应用配置文件",
"description": "当前应用的 JSON 配置",
"mimeType": "application/json"
}
资源模板（支持动态参数）：
{
"uriTemplate": "file:///home/user/{path}",
"name": "文件资源",
"description": "读取指定路径的文件内容"
}
4.2 Tools（工具）
Tools 是 Server 暴露的可执行操作，LLM 可以决定何时调用它们。这是 MCP 最常用的能力。
与 Resources 的区别：
| 特性 | Resources | Tools |
|---|---|---|
| 语义 | 只读数据 | 可执行动作 |
| 触发方 | Client 主动获取 | LLM 决定调用 |
| 副作用 | 无 | 可能有（写入、发送等） |
| 类比 | REST GET | REST POST/PUT/DELETE |
工具定义示例：
{
"name": "search_files",
"description": "在指定目录中搜索匹配关键词的文件",
"inputSchema": {
"type": "object",
"properties": {
"directory": {
"type": "string",
"description": "要搜索的目录路径"
},
"keyword": {
"type": "string",
"description": "搜索关键词"
},
"max_results": {
"type": "integer",
"default": 10,
"description": "最大返回结果数"
}
},
"required": ["directory", "keyword"]
}
}
4.3 Prompts（提示模板）
Prompts 是 Server 提供的可复用提示模板，帮助用户或 LLM 构建更有效的交互。
用途：封装复杂的提示工程，让用户一键调用专业的工作流。
提示模板定义示例：
{
"name": "code_review",
"description": "对代码片段进行全面审查",
"arguments": [
{
"name": "language",
"description": "编程语言",
"required": true
},
{
"name": "code",
"description": "待审查的代码",
"required": true
}
]
}
当用户选择此 Prompt 时，Server 会返回一组结构化的消息：
{
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": "请对以下 Python 代码进行全面审查，关注性能、安全性和可维护性：\n\n```python\n...\n```"
}
}
]
}
5. MCP Server 开发实战
5.1 使用 Python 开发 MCP Server
环境准备
Python MCP SDK 要求 Python 3.10+，推荐使用 uv 作为包管理器。
# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server
# 初始化项目
uv init --python 3.12
uv add "mcp[cli]"
最小化 Server 示例
创建文件 server.py：
"""
一个简单的 MCP Server 示例：
提供数学计算工具和系统信息资源。
"""
from mcp.server.fastmcp import FastMCP
# 创建 MCP Server 实例
mcp = FastMCP(
name="math-and-sysinfo",
version="1.0.0"
)
# ==================== Tools ====================
@mcp.tool()
def calculate(expression: str) -> str:
"""
计算数学表达式。
支持基本运算：+, -, *, /, **, %
示例: "2 + 3 * 4", "2 ** 10"
Args:
expression: 数学表达式字符串
Returns:
计算结果
"""
# 安全检查：只允许数学运算
allowed_chars = set("0123456789+-*/().% ")
if not all(c in allowed_chars for c in expression):
return f"错误：表达式包含不允许的字符"
try:
result = eval(expression) # 仅用于教学，生产环境应使用安全的表达式解析器
return f"结果：{expression} = {result}"
except Exception as e:
return f"计算错误：{str(e)}"
@mcp.tool()
def unit_convert(value: float, from_unit: str, to_unit: str) -> str:
"""
常用单位转换。
支持：
- 温度：celsius, fahrenheit, kelvin
- 长度：km, miles, meters, feet
- 重量：kg, pounds, grams
Args:
value: 数值
from_unit: 源单位
to_unit: 目标单位
Returns:
转换结果
"""
conversions = {
# 温度需要特殊处理
("celsius", "fahrenheit"): lambda v: v * 9/5 + 32,
("fahrenheit", "celsius"): lambda v: (v - 32) * 5/9,
("celsius", "kelvin"): lambda v: v + 273.15,
("kelvin", "celsius"): lambda v: v - 273.15,
# 长度
("km", "miles"): lambda v: v * 0.621371,
("miles", "km"): lambda v: v * 1.60934,
("meters", "feet"): lambda v: v * 3.28084,
("feet", "meters"): lambda v: v * 0.3048,
# 重量
("kg", "pounds"): lambda v: v * 2.20462,
("pounds", "kg"): lambda v: v * 0.453592,
("kg", "grams"): lambda v: v * 1000,
("grams", "kg"): lambda v: v / 1000,
}
key = (from_unit.lower(), to_unit.lower())
if key not in conversions:
return f"不支持的转换：{from_unit} -> {to_unit}"
result = conversions[key](value)
return f"{value} {from_unit} = {result:.4f} {to_unit}"
# ==================== Resources ====================
@mcp.resource("info://server/status")
def server_status() -> str:
"""返回当前服务器的运行状态信息。"""
import platform
import datetime
return f"""服务器状态：
- 名称：math-and-sysinfo
- 版本：1.0.0
- Python：{platform.python_version()}
- 系统：{platform.system()} {platform.release()}
- 当前时间：{datetime.datetime.now().isoformat()}
"""
# ==================== Prompts ====================
@mcp.prompt()
def math_tutor(topic: str) -> str:
"""
数学辅导提示模板。
Args:
topic: 数学主题（如"微积分"、"线性代数"）
"""
return f"""你是一位耐心的数学老师。请用通俗易懂的方式讲解"{topic}"这个主题。
要求：
1. 先给出直观的解释
2. 然后展示 1-2 个简单的例子
3. 最后提供一道练习题让用户思考
请开始讲解："""
# 启动 Server
if __name__ == "__main__":
mcp.run(transport="stdio")
运行与测试
# 直接运行
uv run server.py
# 使用 MCP Inspector 进行可视化测试
uv run mcp dev server.py
MCP Inspector 会在浏览器中打开一个交互界面，你可以：
- 查看 Server 暴露的所有 Tools、Resources、Prompts
- 手动调用工具并查看返回结果
- 测试资源读取
- 调试协议消息
5.2 使用 TypeScript 开发 MCP Server
环境准备
mkdir my-mcp-server-ts && cd my-mcp-server-ts
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init
TypeScript Server 示例
创建 src/index.ts：
/**
* MCP Server 示例（TypeScript）：
* 提供笔记管理工具和配置资源。
*/
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
// 创建 Server 实例
const server = new McpServer({
name: "note-manager",
version: "1.0.0",
});
// 内存中的笔记存储
const notes: Map<string, { title: string; content: string; created: string }> =
new Map();
// ==================== Tools ====================
server.tool(
"create_note",
"创建一条新笔记",
{
title: z.string().describe("笔记标题"),
content: z.string().describe("笔记内容"),
},
async ({ title, content }) => {
const id = `note-${Date.now()}`;
const created = new Date().toISOString();
notes.set(id, { title, content, created });
return {
content: [
{
type: "text" as const,
text: `笔记创建成功！\nID: ${id}\n标题: ${title}\n创建时间: ${created}`,
},
],
};
}
);
server.tool(
"search_notes",
"根据关键词搜索笔记",
{
keyword: z.string().describe("搜索关键词"),
},
async ({ keyword }) => {
const results: string[] = [];
notes.forEach((note, id) => {
if (
note.title.includes(keyword) ||
note.content.includes(keyword)
) {
results.push(
`[${id}] ${note.title}\n ${note.content.substring(0, 100)}...`
);
}
});
if (results.length === 0) {
return { content: [{ type: "text" as const, text: "未找到匹配的笔记。" }] };
}
return {
content: [
{
type: "text" as const,
text: `找到 ${results.length} 条匹配笔记：\n\n${results.join("\n\n")}`,
},
],
};
}
);
// ==================== Resources ====================
server.resource(
"notes-stats",
"notes://stats",
async (uri) => ({
contents: [
{
uri: uri.href,
text: JSON.stringify(
{
totalNotes: notes.size,
lastUpdated: new Date().toISOString(),
},
null,
2
),
},
],
})
);
// ==================== 启动 ====================
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("Note Manager MCP Server 已启动");
}
main().catch(console.error);
编译并运行：
npx tsc
node dist/index.js
6. MCP Client 集成开发
6.1 Python Client 示例
"""
MCP Client 示例：连接 Server，发现并调用工具。
"""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 配置 Server 连接参数
server_params = StdioServerParameters(
command="uv", # 启动命令
args=["run", "server.py"], # 启动参数
env=None # 环境变量
)
# 建立连接
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# ========== 发现能力 ==========
# 列出所有工具
tools = await session.list_tools()
print("可用工具：")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description}")
# 列出所有资源
resources = await session.list_resources()
print("\n可用资源：")
for resource in resources.resources:
print(f" - {resource.uri}: {resource.name}")
# ========== 调用工具 ==========
# 调用数学计算工具
result = await session.call_tool(
"calculate",
arguments={"expression": "2 ** 10 + 1"}
)
print(f"\n计算结果：{result.content[0].text}")
# 调用单位转换工具
result = await session.call_tool(
"unit_convert",
arguments={
"value": 100,
"from_unit": "celsius",
"to_unit": "fahrenheit"
}
)
print(f"转换结果：{result.content[0].text}")
# ========== 读取资源 ==========
resource_content = await session.read_resource("info://server/status")
print(f"\n服务器状态：\n{resource_content.contents[0].text}")
if __name__ == "__main__":
asyncio.run(main())
6.2 在 LLM 应用中集成 MCP Client
将 MCP Client 集成到实际的 LLM 应用中，核心流程是：
"""
将 MCP 工具集成到 LLM 对话中的完整示例。
使用 OpenAI API 格式演示（兼容大多数 LLM 提供商）。
"""
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def run_agent():
"""运行一个简单的 MCP Agent。"""
server_params = StdioServerParameters(
command="uv",
args=["run", "server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 1. 获取所有可用工具，转换为 LLM 的 function calling 格式
tools_response = await session.list_tools()
available_tools = []
for tool in tools_response.tools:
available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
print(f"已注册 {len(available_tools)} 个工具")
# 2. 模拟 LLM 对话循环
user_query = "帮我算一下 15 的阶乘是多少"
print(f"\n用户：{user_query}")
# 3. 调用 LLM（此处为伪代码，需替换为实际 API 调用）
# llm_response = call_llm(
# messages=[{"role": "user", "content": user_query}],
# tools=available_tools
# )
# 4. 如果 LLM 决定调用工具
# 假设 LLM 返回了工具调用请求
tool_call = {
"name": "calculate",
"arguments": {"expression": "1 * 2 * 3 * 4 * 5 * 6 * 7 * 8 * 9 * 10 * 11 * 12 * 13 * 14 * 15"}
}
# 5. 通过 MCP 执行工具调用
result = await session.call_tool(
tool_call["name"],
arguments=tool_call["arguments"]
)
print(f"工具结果：{result.content[0].text}")
# 6. 将结果返回给 LLM 生成最终回答
# final_response = call_llm(
# messages=[
# {"role": "user", "content": user_query},
# {"role": "assistant", "tool_calls": [...]},
# {"role": "tool", "content": result.content[0].text}
# ]
# )
if __name__ == "__main__":
asyncio.run(run_agent())
7. 常见 MCP Server 案例
7.1 文件系统 Server
文件系统 Server 是最基础也是最常用的 MCP Server，提供文件读写、目录浏览、文件搜索等功能。
"""文件系统 MCP Server 核心实现。"""
import os
import glob
from pathlib import Path
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("filesystem")
# 安全根目录：限制可访问的范围
ALLOWED_ROOT = Path("/home/user/workspace")
def safe_path(requested: str) -> Path:
"""确保路径在允许的范围内。"""
resolved = (ALLOWED_ROOT / requested).resolve()
if not str(resolved).startswith(str(ALLOWED_ROOT.resolve())):
raise ValueError(f"访问被拒绝：路径超出允许范围")
return resolved
@mcp.tool()
def read_file(path: str) -> str:
"""读取文件内容。"""
target = safe_path(path)
if not target.exists():
return f"错误：文件不存在 - {path}"
if not target.is_file():
return f"错误：不是文件 - {path}"
return target.read_text(encoding="utf-8")
@mcp.tool()
def write_file(path: str, content: str) -> str:
"""创建或覆盖文件。"""
target = safe_path(path)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content, encoding="utf-8")
return f"文件已写入：{path}（{len(content)} 字符）"
@mcp.tool()
def list_directory(path: str = ".") -> str:
"""列出目录内容。"""
target = safe_path(path)
if not target.is_dir():
return f"错误：不是目录 - {path}"
entries = []
for item in sorted(target.iterdir()):
rel = item.relative_to(ALLOWED_ROOT)
if item.is_dir():
entries.append(f"📁 {rel}/")
else:
size = item.stat().st_size
entries.append(f"📄 {rel} ({size} bytes)")
return "\n".join(entries) if entries else "（空目录）"
@mcp.tool()
def search_files(directory: str, pattern: str) -> str:
"""使用通配符模式搜索文件。"""
target = safe_path(directory)
matches = glob.glob(str(target / "**" / pattern), recursive=True)
if not matches:
return f"未找到匹配 '{pattern}' 的文件"
results = [str(Path(m).relative_to(ALLOWED_ROOT)) for m in matches[:20]]
return f"找到 {len(matches)} 个文件：\n" + "\n".join(results)
7.2 数据库 Server
"""SQLite 数据库 MCP Server 核心实现。"""
import sqlite3
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("database")
DB_PATH = "data/app.db"
def get_connection():
"""获取数据库连接。"""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
@mcp.tool()
def query_database(sql: str) -> str:
"""
执行 SQL 查询（仅 SELECT）。
Args:
sql: SQL 查询语句
"""
# 安全检查：只允许 SELECT 查询
sql_stripped = sql.strip().upper()
if not sql_stripped.startswith("SELECT"):
return "错误：只允许 SELECT 查询"
try:
conn = get_connection()
cursor = conn.execute(sql)
rows = cursor.fetchall()
conn.close()
if not rows:
return "查询结果为空"
# 格式化输出
columns = rows[0].keys()
header = " | ".join(columns)
separator = "-" * len(header)
lines = [header, separator]
for row in rows[:50]: # 限制返回行数
lines.append(" | ".join(str(row[col]) for col in columns))
if len(rows) > 50:
lines.append(f"\n... 共 {len(rows)} 条记录，仅显示前 50 条")
return "\n".join(lines)
except Exception as e:
return f"查询错误：{str(e)}"
@mcp.tool()
def list_tables() -> str:
"""列出数据库中的所有表。"""
conn = get_connection()
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
tables = [row["name"] for row in cursor.fetchall()]
conn.close()
if not tables:
return "数据库中没有表"
return "数据库表：\n" + "\n".join(f" - {t}" for t in tables)
@mcp.tool()
def describe_table(table_name: str) -> str:
"""查看表结构。"""
conn = get_connection()
cursor = conn.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
conn.close()
if not columns:
return f"表 '{table_name}' 不存在或没有列"
lines = [f"表 '{table_name}' 结构："]
for col in columns:
nullable = "NOT NULL" if col["notnull"] else "NULL"
pk = " [PK]" if col["pk"] else ""
lines.append(f" {col['name']}: {col['type']} {nullable}{pk}")
return "\n".join(lines)
7.3 外部 API Server
"""天气 API MCP Server 核心实现。"""
import json
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("weather-api")
@mcp.tool()
async def get_weather(city: str) -> str:
"""
获取指定城市的当前天气。
Args:
city: 城市名称（中文或英文）
"""
# 此处使用 wttr.in 免费 API 作为示例
# 实际生产中应使用正规天气 API
import httpx
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://wttr.in/{city}?format=j1",
timeout=10.0
)
data = response.json()
current = data["current_condition"][0]
result = {
"城市": city,
"温度": f"{current['temp_C']}°C",
"体感温度": f"{current['FeelsLikeC']}°C",
"天气": current["weatherDesc"][0]["value"],
"湿度": f"{current['humidity']}%",
"风速": f"{current['windspeedKmph']} km/h",
"风向": current["winddir16Point"]
}
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as e:
return f"获取天气信息失败：{str(e)}"
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
"""
获取未来几天的天气预报。
Args:
city: 城市名称
days: 预报天数（1-3）
"""
import httpx
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://wttr.in/{city}?format=j1",
timeout=10.0
)
data = response.json()
forecasts = []
for day in data["weather"][:days]:
forecasts.append({
"日期": day["date"],
"最高温度": f"{day['maxtempC']}°C",
"最低温度": f"{day['mintempC']}°C",
"平均温度": f"{day['avgtempC']}°C",
"日出": day["astronomy"][0]["sunrise"],
"日落": day["astronomy"][0]["sunset"]
})
return json.dumps(forecasts, ensure_ascii=False, indent=2)
except Exception as e:
return f"获取天气预报失败：{str(e)}"
8. MCP 与 Function Calling 的区别和联系
8.1 理解 Function Calling
Function Calling 是 LLM 提供商（如 OpenAI、Anthropic）提供的能力，让模型能够在对话中决定调用预定义的函数。
// OpenAI Function Calling 示例
{
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取天气信息",
"parameters": {
"type": "object",
"properties": {
"city": { "type": "string" }
}
}
}
}
]
}
8.2 核心区别
| 维度 | Function Calling | MCP |
|---|---|---|
| 定位 | LLM 的能力特性 | 应用层协议标准 |
| 范围 | 仅定义工具调用 | 定义完整的上下文交互（工具+资源+提示） |
| 绑定 | 与特定 LLM 提供商绑定 | 与 LLM 无关，跨模型通用 |
| 连接 | 无持久连接概念 | 支持持久会话和实时通知 |
| 发现 | 工具列表在请求中硬编码 | 支持动态发现（tools/list） |
| 传输 | HTTP 请求/响应 | stdio / HTTP+SSE |
| 生态 | 各家私有格式 | 统一的开放标准 |
8.3 它们如何协作
MCP 不是 Function Calling 的替代品，而是补充和扩展。在实际应用中，它们通常协同工作：
用户请求
│
▼
┌─────────┐ ┌──────────────┐ ┌──────────────┐
│ LLM │────►│ MCP Client │────►│ MCP Server │
│ │ │ │ │ │
│ Function│◄────│ 将 MCP 工具 │◄────│ 返回执行结果 │
│ Calling │ │ 转换为 FC │ │ │
└─────────┘ └──────────────┘ └──────────────┘
工作流程：
- MCP Client 向 Server 发现可用工具（
tools/list） - Client 将 MCP 工具定义转换为 LLM 的 Function Calling 格式
- LLM 根据用户请求决定调用哪个函数
- Client 将函数调用转发给 MCP Server 执行
- Server 执行后返回结果
- Client 将结果反馈给 LLM 生成最终回答
9. 安全考虑与最佳实践
9.1 权限控制
原则：最小权限
# ❌ 不好的做法：暴露危险操作
@mcp.tool()
def run_command(cmd: str) -> str:
"""执行任意系统命令。"""
import subprocess
return subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
# ✅ 好的做法：限制操作范围
@mcp.tool()
def list_processes() -> str:
"""列出当前用户的进程。"""
import subprocess
return subprocess.run(
["ps", "-u", os.getenv("USER")],
capture_output=True, text=True
).stdout
9.2 输入验证
@mcp.tool()
def query_database(sql: str) -> str:
"""安全的数据库查询。"""
# 1. 类型检查
if not isinstance(sql, str):
return "错误：SQL 必须是字符串"
# 2. 长度限制
if len(sql) > 10000:
return "错误：SQL 查询过长"
# 3. 语句类型限制
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return "错误：仅允许 SELECT 查询"
# 4. 危险关键词过滤
dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "EXEC"]
for keyword in dangerous:
if keyword in sql_upper:
return f"错误：包含不允许的关键字 {keyword}"
# 5. 执行查询...
9.3 路径遍历防护
def safe_resolve(base_dir: str, user_path: str) -> str:
"""
安全地解析用户提供的路径，防止路径遍历攻击。
"""
import os
# 移除多余的分隔符和 ..
normalized = os.path.normpath(user_path)
# 拼接基础目录
full_path = os.path.join(base_dir, normalized)
# 解析为绝对路径
resolved = os.path.realpath(full_path)
# 验证是否在允许范围内
if not resolved.startswith(os.path.realpath(base_dir)):
raise ValueError(f"路径越界：{user_path}")
return resolved
9.4 最佳实践清单
| 类别 | 实践 | 说明 |
|---|---|---|
| 认证 | 实施 API Key 或 OAuth | 防止未授权访问 |
| 授权 | 基于角色的权限控制 | 不同用户不同权限 |
| 输入 | 验证所有输入参数 | 类型、长度、格式检查 |
| 输出 | 过滤敏感信息 | 不暴露内部路径、密钥等 |
| 网络 | 限制可访问的网络范围 | 白名单优于黑名单 |
| 日志 | 记录所有工具调用 | 便于审计和问题排查 |
| 超时 | 设置合理的超时时间 | 防止资源耗尽 |
| 速率 | 实施速率限制 | 防止滥用 |
| 错误 | 不暴露内部错误细节 | 返回用户友好的错误信息 |
10. MCP 生态系统与未来展望
10.1 当前生态
MCP 生态正在快速发展，主要组成部分包括：
官方资源：
- MCP 规范：https://spec.modelcontextprotocol.io
- Python SDK：
mcp包 - TypeScript SDK：
@modelcontextprotocol/sdk - MCP Inspector：调试和测试工具
社区 MCP Server（部分列举）：
- 文件系统（filesystem）
- GitHub 集成
- PostgreSQL / SQLite 数据库
- Slack 消息
- Google Drive
- Brave Search
- Memory（知识图谱记忆）
- Puppeteer（浏览器自动化）
10.2 MCP 注册中心
MCP 正在建立标准化的注册中心（Registry），让开发者可以：
- 发现和搜索可用的 MCP Server
- 查看 Server 的能力描述和版本信息
- 一键安装和配置 Server
10.3 未来发展方向
- 标准化推进：更多 LLM 原生支持 MCP，减少适配层
- 能力扩展：支持双向通信、流式响应、长时任务
- 安全增强：标准化的认证框架（OAuth 2.0 集成）
- 性能优化：批处理请求、压缩传输、连接池
- 开发工具：更好的调试工具、测试框架、文档生成器
- 企业级特性：审计日志、合规控制、多租户支持
10.4 学习资源
- 官方文档：https://modelcontextprotocol.io
- GitHub 仓库：https://github.com/modelcontextprotocol
- MCP Server 仓库：https://github.com/modelcontextprotocol/servers
- 社区 Discord：MCP 开发者社区
总结
MCP 协议为 AI 应用开发提供了一个标准化的工具集成框架。通过本文的学习，你应该能够：
- ✅ 理解 MCP 的架构设计和核心概念
- ✅ 使用 Python 或 TypeScript 开发自定义 MCP Server
- ✅ 集成 MCP Client 到你的 AI 应用中
- ✅ 构建文件系统、数据库、API 等常见 Server
- ✅ 理解 MCP 与 Function Calling 的关系
- ✅ 遵循安全最佳实践
MCP 仍处于快速发展阶段，建议持续关注官方规范更新和社区动态。掌握 MCP 开发能力，将帮助你构建更强大、更灵活的 AI 应用。
下一步行动：
- 安装 MCP SDK，运行本文的示例代码
- 使用 MCP Inspector 调试你的第一个 Server
- 尝试为你常用的工具（如笔记、任务管理）编写 MCP Server
- 将 MCP 集成到你的 AI 应用或 Agent 框架中
