再谈MCP协议-工具
本次我们再一次聊聊MCP中的工具，这是结合之前我们说了MCP中的协议层之后，再一次查看MCP中的工具实现。
在实际讲述之前，我们需要先说下计算1.0和计算2.0这两个模型发展时代的区别，在之前计算1.0的时代，往往输出结果是或与非，也即0或者1的输出方式，而计算2.0时代，则两者运算逻辑高度相似，结合概率计算思维，实现了机器和人的自然交互。
在2.0时代，交互方式发生了变化，但并非是万能的，大模型可以进行计算，进行思考，但由于自身限制，并不能直接执行任务，而为了让大模型可以执行任务，因此发展出了LLM+工具调用体系。
具体的就是MCP的协议实现，其让服务器可以自己通过统一的协议调用工具。最终让大模型和外部系统进行交互，比如查询数据库，调用API。
更加具体的就是服务器端声明了自己支持tools调用
|
{
“capabilities”: { “tools”: { “listChanged”: true } } } |
而在一个工具内部，包含着name，description，inputSchema，annotations这些元数据字段，表明着工具用于做什么的，下面就是一个示例
|
{
“name”: “get_weather”, “description”: “Get current weather information for a location”, “inputSchema”: { “type”: “object”, “properties”: { “location”: { “type”: “string”, “description”: “City name or zip code” } }, “required”: [“location”] } } |
而对应到更加细粒度的代码层面来说，交互流程如下
首先获取到工具列表
|
{
“jsonrpc”: “2.0”, “id”: 1, “method”: “tools/list”, “params”: { “cursor”: “optional-cursor-value” } } |
服务器端的响应如下
|
{
“jsonrpc”: “2.0”, “id”: 1, “result”: { “tools”: [ { “name”: “get_weather”, “description”: “Get current weather information for a location”, “inputSchema”: { “type”: “object”, “properties”: { “location”: { “type”: “string”, “description”: “City name or zip code” } }, “required”: [“location”] } } ], “nextCursor”: “next-page-cursor” } } |
获取到了数组形式的tool的metadata
之后是客户端进行工具的调用，发起tools/call请求
|
{
“jsonrpc”: “2.0”, “id”: 2, “method”: “tools/call”, “params”: { “name”: “get_weather”, “arguments”: { “location”: “New York” } } } |
将要求的参数传入给服务器端，并期待得到响应。
|
{
“jsonrpc”: “2.0”, “id”: 2, “result”: { “content”: [ { “type”: “text”, “text”: “Current weather in New York:\nTemperature: 72°F\nConditions: Partly cloudy” } ], “isError”: false } } |
在其中，需要注意的是返回的信息有不同的类型
诸如文本内容
|
{
“type”: “text”, “text”: “Tool result text” } |
或者图片内容
|
{
“type”: “image”, “data”: “base64-encoded-data”, “mimeType”: “image/png” } |
音频内容
|
{
“type”: “audio”, “data”: “base64-encoded-audio-data”, “mimeType”: “audio/wav” } |
以及其他的内嵌资源。
而且MCP 还有着工具调用的错误机制。
一个是协议错误，常见的适用于 无效参数和服务器错误这类情况。
|
{
“jsonrpc”: “2.0”, “id”: 3, “error”: { “code”: -32602, “message”: “Unknown tool: invalid_tool_name” } } |
一个是工具执行错误，常见的包含 无效数据输入，业务逻辑错误。
|
{
“jsonrpc”: “2.0”, “id”: 4, “result”: { “content”: [ { “type”: “text”, “text”: “Failed to fetch weather data: API rate limit exceeded” } ], “isError”: true } } |
对于工具的调用，在设计的时候，一般都需要注意一点，客户端需要在设计的时候考虑，如果交互涉及了敏感操作，需要在服务器端项用户展示工具输入，或者提供验证机制，交由客户来进行实际的调用。
在这里，我们就以一个简单的示例来展示工具的实战，代码中我们分为服务器端和客户端两者来展示实现。并且分别展示通过高纬度FastMCP和低纬度原生MCP Server两种方式来展示代码。
首先是高纬度，也就是通过FastMCP框架实现的方式。
其支持通过装饰器和注解的方式来暴露相关Tool供客户端调用。
比如@mcp.tool
|
import asyncio
from mcp.server.fastmcp import FastMCP # 初始化 FastMCP 服务器 mcp = FastMCP(“tools-server”) @mcp.tool() async def calculator(operation: str, a: float, b: float) -> str: “””执行基本的数学运算 Args: operation: 运算类型 (add, subtract, multiply, divide) a: 第一个数字 b: 第二个数字 “”” if operation == “add”: return f”计算结果: {a + b}” elif operation == “subtract”: return f”计算结果: {a – b}” elif operation == “multiply”: return f”计算结果: {a * b}” elif operation == “divide”: if b == 0: return “错误：除数不能为零” return f”计算结果: {a / b}” @mcp.tool() async def text_analyzer(text: str) -> str: “””分析文本，统计字符数和单词数 Args: text: 要分析的文本 “”” char_count = len(text) word_count = len(text.split()) return f”字符数: {char_count}\n单词数: {word_count}” if __name__ == “__main__”: mcp.run(transport=”stdio”) |
如果直接使用MCP的Server实现，则是如下代码
|
import asyncio
import mcp.types as types from mcp.server import Server from mcp.server.stdio import stdio_server app = Server(“tools-server”) @app.list_tools() async def list_tools() -> list[types.Tool]: “”” 返回服务器提供的工具列表 “”” return [ types.Tool( name=”calculator”, description=”执行基本的数学运算（加、减、乘、除）”, inputSchema={ “type”: “object”, “properties”: { “operation”: { “type”: “string”, “enum”: [“add”, “subtract”, “multiply”, “divide”] }, “a”: {“type”: “number”}, “b”: {“type”: “number”} }, “required”: [“operation”, “a”, “b”] } ), types.Tool( name=”text_analyzer”, description=”分析文本，统计字符数和单词数”, inputSchema={ “type”: “object”, “properties”: { “text”: {“type”: “string”} }, “required”: [“text”] } ) ] @app.call_tool() async def call_tool( name: str, arguments: dict ) -> list[types.TextContent]: “”” 处理工具调用请求 “”” if name == “calculator”: operation = arguments[“operation”] a = arguments[“a”] b = arguments[“b”] if operation == “add”: result = a + b elif operation == “subtract”: result = a – b elif operation == “multiply”: result = a * b elif operation == “divide”: if b == 0: return [types.TextContent(type=”text”, text=”错误：除数不能为零”)] result = a / b return [types.TextContent(type=”text”, text=f”计算结果: {result}”)] elif name == “text_analyzer”: text = arguments[“text”] char_count = len(text) word_count = len(text.split()) return [types.TextContent( type=”text”, text=f”字符数: {char_count}\n单词数: {word_count}” )] return [types.TextContent(type=”text”, text=f”未知工具: {name}”)] async def main(): async with stdio_server() as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) if __name__ == “__main__”: asyncio.run(main()) |
也提供了对应的装饰器，但是使用起来的粒度更为细，适合有更多客制化需求的场景。
之后是相关的client端的实现。
|
def call_llm_with_tools(messages, tools):
response = client.chat.completions.create( model=”deepseek-chat”, messages=messages, tools=tools, tool_choice=”auto” ) return response.choices[0].message async with stdio_client(params) as (reader, writer): async with ClientSession(reader, writer) as session: await session.initialize() notification = Notification( method=”notifications/initialized”, params={} ) await session.send_notification(notification) # 获取工具列表 response = await session.list_tools() tools = response.tools print(“欢迎使用工具调用系统！\n可用工具列表已加载。\n请输入您的需求（输入’退出’结束）：”) # 构造 tools 列表 tools_list = [{ “type”: “function”, “function”: { “name”: tool.name, “description”: tool.description, “parameters”: tool.inputSchema } } for tool in tools] # 交互主循环 messages = [ {“role”: “system”, “content”: “你是一个智能助手，请根据用户输入选择合适的工具并构造参数，或直接回复用户。”} ] while True: user_input = input(“> “) if user_input.lower() == “退出”: break messages.append({“role”: “user”, “content”: user_input}) message = call_llm_with_tools(messages, tools_list) messages.append(message) if not message.tool_calls: print(message.content) continue # 工具调用 for tool_call in message.tool_calls: args = json.loads(tool_call.function.arguments) result = await session.call_tool(tool_call.function.name, args) messages.append({ “role”: “tool”, “content”: str(result), “tool_call_id”: tool_call.id }) # 再次让 LLM 总结最终回复（这时 messages 里有 “答案” 的内容了） message = call_llm_with_tools(messages, tools_list) print(message.content) messages.append(message) |
通过session获取到tools，然后tools列表和message列表交给Deepseek，剩下的LLM自动完成。
在交互过程中，如果返回的信息要求进行工具调用，则交给session进行相关的工具调用。
在不断你的交互过程中，直到得到最后的答案，那么这就是在MCP中Tool的具体实现。
这里我们总结一下，工具调用原语的关键流程。
MCP协议把每一次列工具和调工具都转换为了标准的JSON-RPC请求
客户端读取 list_tools 返回的工具定义，以及call_tool的返回结果。
并且利用了统一的封装交互方式，避免了原本的Function Calling需要深入细节了解相关工具具体实现的问题。