从 Tool Calling 到标准化协议 —— 构建 LLM 可调用的工具服务,一次定义、所有平台通用
在 LangChain 中,用 @tool 装饰器就能让 LLM 调用函数:
@tool
def get_weather(city: str) -> str:
"""查询城市天气"""
return f"{city}: 晴, 25°C"
llm_with_tools = llm.bind_tools([get_weather])
# LLM 自动判断:北京天气 → 调 get_weather("北京")
每个 LLM 平台的 Tool Calling 格式都不同。如果你有 10 个工具,想在 3 个平台用——
| 平台 | 格式 | 维护成本 |
|---|---|---|
| OpenAI | tools=[{"type":"function","function":{...}}] |
写 1 遍 |
| Anthropic | tools=[{"name":"...","input_schema":{...}}] |
再写 1 遍 |
| Google Gemini | tools=[{"functionDeclarations":[{...}]}] |
再写 1 遍 |
| 问题 | MCP 之前 | MCP 之后 |
|---|---|---|
| 工具定义 | 每个平台一种格式 | 一次定义,所有平台通用 |
| 工具部署 | 嵌入在应用代码里 | 独立进程,即插即用 |
| 工具发现 | 硬编码在代码里 | Client 自动发现 Server 提供的工具 |
| 工具复用 | 每个项目重新写 | 一个 Server 多个项目共用 |
| 安全隔离 | 和 LLM 同进程 | 独立进程,权限隔离 |
| 角色 | 做什么 | 举例 |
|---|---|---|
| MCP Host | 运行 LLM 的应用 | Claude Desktop、Cursor、VS Code Copilot |
| MCP Client | Host 内部的协议客户端 | 连接 Server、管理 Tool 列表、转发调用 |
| MCP Server | 提供具体功能的服务 | 天气查询、文件操作、数据库访问、API 网关 |
通过 stdin/stdout 管道通信
Server 作为子进程运行
零网络配置,进程级隔离
适合:本地工具(文件系统、数据库、CLI)
通过 HTTP Server-Sent Events 通信
Server 作为独立服务运行
支持跨机器访问
适合:远程工具(API 网关、云服务)
执行操作
让 LLM "做"事情
查天气、写文件、发邮件、调 API
读取数据
让 LLM "看"数据
读配置、数据库 Schema、日志文件
提示词模板
让 LLM 快速获取
周报模板、代码审查模板等
每个 Tool 有三个要素,description 是最关键的——LLM 靠它决定"什么时候用":
Tool(
name="get_weather", # 1. 名称:LLM 用来标识
description="获取指定城市的天气信息", # 2. 描述:LLM 用来理解"什么时候用"
inputSchema={ # 3. 参数 Schema:LLM 用来生成"正确的参数"
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"],
},
)
| Tool | Resource | |
|---|---|---|
| 语义 | 做事情(执行操作) | 读数据(获取信息) |
| 副作用 | 有(写文件、发邮件) | 无(只读) |
| 标识 | 通过 name 标识 | 通过 URI 标识 |
| 参数 | 每次调用时传入 | 通过 URI 标识资源 |
| 示例 | write_file、send_email | file:///readme.md、db:///schema |
Prompt(
name="write_weekly_report",
description="生成周报的提示词模板",
arguments=[
{"name": "project", "description": "项目名称", "required": True},
],
)
# LLM 收到 Prompt 模板后,按照专业结构生成内容
# 比用户手写"帮我写周报"质量更高
一个 Server 提供两组共 8 个工具,LLM 根据用户意图自动选择:
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
# 1. 创建 Server
server = Server("my-server")
# 2. 声明 Tool 列表(LLM 的"工具菜单")
@server.list_tools()
async def list_tools() -> list[Tool]:
return [Tool(name=..., description=..., inputSchema=...)]
# 3. 实现 Tool 调用(执行逻辑)
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "my_tool":
return [TextContent(type="text", text=result)]
# 4. 启动 Server(监听 stdin/stdout)
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, ...)
LLM 通过文件系统工具操作本地文件
→ 探索目录结构 list_directory
→ 读取文件内容 read_file
→ 搜索特定文件 search_files
→ 创建/修改文件 write_file
LLM 通过数据库工具查询结构化数据
→ 了解有哪些表 db_list_tables
→ 理解表结构 db_describe_table
→ 执行分析查询 db_query
→ 导出完整数据 db_export_table
用户可能对 LLM 说"读取 /etc/passwd",LLM 可能传入 ../../../etc/passwd:
def read_file(user_path: str):
# 危险!直接拼接用户路径
full = Path("/workspace") / user_path
return full.read_text()
# user_path = "../../../etc/passwd"
# → /etc/passwd ← 系统文件泄露!
def safe_path(user_path: str) -> Path:
full = Path(allowed_root) / user_path
resolved = full.resolve() # 展开 ../ 和符号链接
if not str(resolved).startswith(str(allowed_root)):
raise PermissionError("路径越界!")
return resolved
# user_path = "../../../etc/passwd"
# → PermissionError: 路径穿越检测!
def _handle_db_query(args: dict):
sql = args["sql"].strip()
# 检查 1:只允许 SELECT
if not sql.upper().startswith("SELECT"):
return [TextContent(text="⛔ 只允许 SELECT 查询")]
# 检查 2:禁止危险关键词
dangerous = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER"]
for kw in dangerous:
if kw in sql.upper():
return [TextContent(text=f"⛔ 禁止 {kw}")]
# 检查 3:只读连接(即使绕过前两步,数据库也不允许写)
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
| 防护层 | 措施 | 防什么 |
|---|---|---|
| 路径校验 | safe_path() resolve + prefix 检查 | 读取系统文件、覆盖关键配置 |
| 操作白名单 | 只允许 SELECT,拒绝 DROP/DELETE/INSERT 等 | 数据被删除、篡改 |
| 只读连接 | sqlite3.connect("file:...?mode=ro") | 即使绕过代码检查也无法写入 |
| 文件大小限制 | 拒绝读取超过 1MB 的文件 | 内存溢出、读取二进制大文件 |
| 结果行数限制 | SQL 查询最多返回 100 行 | 全表扫描导致 Token 爆炸 |
在项目根目录创建 .mcp.json:
{
"mcpServers": {
"local-tools": {
"command": "python3",
"args": ["mcp_server.py"],
"env": {
"WORKSPACE_DIR": "/path/to/your/workspace",
"DB_PATH": "/path/to/your/database.db"
}
}
}
}
编辑 ~/.claude/settings.json:
{
"mcpServers": {
"local-tools": {
"command": "python3",
"args": ["/absolute/path/to/mcp_server.py"]
}
}
}
配置完成后,在 Claude Code 中验证:
# 验证连接成功
/list_tools
# 应该看到 list_directory, read_file, db_query 等工具
然后直接用自然语言操作:
"帮我看看工作目录下有什么文件"
"读取 config.json 的内容"
"搜索所有 .py 文件"
"创建一个 hello.py 文件"
"数据库中有哪些表?"
"查看 products 表的结构"
"查询每个分类的产品数量和平均价格"
"导出 orders 表的数据"
| OpenAI Function Calling | LangChain @tool | MCP | |
|---|---|---|---|
| 范围 | 单次对话内的工具 | 单个应用内的工具 | 跨应用、跨平台标准 |
| 定义方式 | JSON Schema | Python 装饰器 | Python 装饰器 + JSON Schema |
| 部署 | 嵌入代码 | 嵌入代码 | 独立进程 |
| 传输 | HTTP API | 内存调用 | stdio / SSE |
| 平台绑定 | 仅 OpenAI | LangChain 生态 | 所有支持 MCP 的平台 |
| 工具复用 | 每个项目重新定义 | 可复用但不跨平台 | 一次定义到处使用 |
| 安全隔离 | 无 | 无 | 进程级隔离 |
| 资源 | 说明 |
|---|---|
mcp_guide.md | 完整教学指南(11 章:Tool Calling → MCP 概念 → Claude Code 集成) |
mcp_server.py | MCP Server 实现(8 个工具 + 安全防护 + --demo 自测模式) |
006-langchain-learning | 前置知识:LangChain Tool Calling、@tool 装饰器 |