当前位置： 首页 > news >正文

MCP与Function Calling

news 2025/8/6 16:54:40

MCP与Function Calling

视频链接：

MCP与Function Calling

一、MCP与Function Calling的关系误区及纠正

  • 普遍错误观点：认为MCP统一了Function Calling的协议，会取代Function Calling。
  • 正确关系：两者是互补关系，在大模型领域所处地位不同，会长期共存，甚至可出现在同一链路中。
  • 核心区别：MCP作用于服务器与函数之间，规定工具的发现与调用协议；Function Calling作用于模型（或模型API）与函数列表之间，是模型挑选函数的能力。

二、Function Calling的概念

  • 本质定义：指模型与外部工具交互的能力，外部工具本质是编程语言中的函数，因此Function Calling即模型调用函数的能力。
  • 关键特点：模型自身无法直接调用工具，需通过中间人（如服务器）协助完成调用。
  • 扩展理解：在实际应用中，更多关注模型API的Function Calling能力，因其与模型的Function Calling能力本质一致（均为从函数列表中挑选函数），只是视角不同。
    在这里插入图片描述

三、Function Calling链路分析（以ChatGPT为例）

  1. 角色构成：用户、ChatGPT应用（含网络搜索函数、OpenAI服务器、GPT4O模型）。
  2. 流程步骤：
    • 用户发送问题（如“纽约明天的天气怎么样”）给ChatGPT应用。
    • OpenAI服务器将问题及可用工具（如网络搜索）转发给GPT4O模型。
    • 模型评估后决定调用工具，返回工具使用请求（含工具名称和参数）给服务器。
    • 服务器调用对应函数（如网络搜索），将结果返回给模型。
    • 模型根据结果总结答案，经服务器反馈给用户。
  3. 作用环节：聚焦于模型（或模型API）挑选工具、解析工具执行结果的环节，与函数的实际调用过程无关。
    在这里插入图片描述

四、Function Calling协议内容分析

  • 协议核心：规定工具列表传给模型API的方式，以及模型API返回挑选的工具和参数的方式。
  • 通过Mark chat应用演示：
    • 应用流程：用户提问后，先调用search工具获取天气信息，再由模型总结答案。
    • 核心代码：process_user_query函数包含Function Calling的核心逻辑，涉及调用模型API、提取工具信息、执行工具、再次调用模型API等步骤。
    • 模型API请求字段：
      • model：指定使用的模型（如GPT4O）。
      • messages：包含用户问题等历史消息。
      • tools：可用工具列表，含工具名称及参数的JSON schema描述。
      • stream：是否采用流式返回（示例中设为false）。
    • 模型API返回内容：首次返回需调用的工具及参数，二次返回基于工具结果的最终答案。
      详细代码：GitHub 代码链接

class LLMProcessor:def __init__(self):self.api_key = OPENROUTER_API_KEYself.base_url = "https://openrouter.ai/api/v1/chat/completions"self.headers = {"Authorization": f"Bearer {self.api_key}","Content-Type": "application/json"}self.history = []def process_user_query(self, query):self.history.append({"role": "user", "content": query})first_model_response = self.call_model()first_model_message = first_model_response["choices"][0]["message"]self.history.append(first_model_message)# 检查模型是否需要调用工具if "tool_calls" in first_model_message and first_model_message["tool_calls"]:tool_call = first_model_message["tool_calls"][0]tool_name = tool_call["function"]["name"]tool_args = json.loads(tool_call["function"]["arguments"])result = self.execute_tool(tool_name, tool_args)self.history.append({"role": "tool","tool_call_id": tool_call["id"],"name": tool_name,"content": result})second_response_data = self.call_model_after_tool_execution()final_message = second_response_data["choices"][0]["message"]self.history.append(final_message)return {"tool_name": tool_name,"tool_parameters": tool_args,"tool_executed": True,"tool_result": result,"final_response": final_message["content"],}else:return {"final_response": first_model_message["content"],}def execute_tool(self, function_name, args):if function_name == "search":# 正常情况下，这里应该调用相关 API 做搜索，为了减少代码的复杂度，# 这里我们返回一段假的工具执行结果，用以测试return "纽约市今天的天气是晴天，明天的天气是多云。"else:raise ValueError(f"未知的工具名称：{function_name}")def call_model(self):request_body = {"model": MODEL_NAME,"messages": self.history,"tools": TOOLS,"stream": False,}response = requests.post(self.base_url,headers=self.headers,json=request_body)logger.log(f"第一次模型请求：\n{json.dumps(request_body, indent=2, ensure_ascii=False)}\n")logger.log(f"第一次模型返回：\n{json.dumps(response.json(), indent=2, ensure_ascii=False)}\n")if response.status_code != 200:raise Exception(f"API request failed with status {response.status_code}: {response.text}")return response.json()def call_model_after_tool_execution(self):second_request_body = {"model": MODEL_NAME,"messages": self.history,"tools": TOOLS,}# Make the second POST requestsecond_response = requests.post(self.base_url,headers=self.headers,json=second_request_body)logger.log(f"第二次模型请求：\n{json.dumps(second_request_body, indent=2, ensure_ascii=False)}\n")logger.log(f"第二次模型返回：\n{json.dumps(second_response.json(), indent=2, ensure_ascii=False)}\n")# Check if the request was successfulif second_response.status_code != 200:raise Exception(f"API request failed with status {second_response.status_code}: {second_response.text}")# Parse the second responsereturn second_response.json()def execute_tool_with_mcp(self, function_name, args):loop = asyncio.new_event_loop()return loop.run_until_complete(self.execute_tool_with_mcp_async(function_name, args))async def execute_tool_with_mcp_async(self, function_name, args):# 获取与当前脚本同目录下的 mcp_server.py 的绝对地址mcp_server_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "mcp_server.py"))# 启动 MCP Client 并调用 MCP Toolasync with MCPClient("uv", ["run", mcp_server_path]) as client:return await client.call_tool(function_name, args)

模型交互日志：
在这里插入图片描述
在这里插入图片描述

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述在这里插入图片描述

FUCTION CALLING具体要点：就是是否具有解析和使用的能力。如果能力差，不一定能正确解析和返回正确的参数
在这里插入图片描述

五、同时使用Function Calling和MCP的实现

  • 修改要点：在Mark chat应用中，将执行工具的函数execute_tool替换为遵循MCP规范的execute_tool_with_mcp。
  • 实现逻辑：
    • execute_tool_with_mcp启动MCP客户端，连接项目中的MCP服务器。
    • MCP服务器返回带标识的结果，证明链路中已集成MCP。
  • 效果验证：修改后应用可正常返回结果，且工具调用信息显示来自MCP服务器，验证了两者可协同工作。

就是是需要改下面代码：

        if "tool_calls" in first_model_message and first_model_message["tool_calls"]:tool_call = first_model_message["tool_calls"][0]tool_name = tool_call["function"]["name"]tool_args = json.loads(tool_call["function"]["arguments"])result = self.execute_tool_with_mcp(tool_name, tool_args)self.history.append({"role": "tool","tool_call_id": tool_call["id"],"name": tool_name,"content": result})

这调用MCP服务端代码如下（上面代码也有这里列出）：

def execute_tool_with_mcp(self, function_name, args):# 创建一个新的异步事件循环loop = asyncio.new_event_loop()# 运行异步函数execute_tool_with_mcp_async直到完成，并返回其结果# 这是同步函数调用异步函数的常见方式，通过事件循环驱动异步执行return loop.run_until_complete(self.execute_tool_with_mcp_async(function_name, args))async def execute_tool_with_mcp_async(self, function_name, args):# 计算mcp_server.py的绝对路径：# 1. os.path.dirname(__file__) 获取当前脚本所在目录# 2. os.path.join(...) 拼接当前目录与"mcp_server.py"得到相对路径# 3. os.path.abspath(...) 将相对路径转换为绝对路径，确保跨环境可正确访问mcp_server_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "mcp_server.py"))# 使用异步上下文管理器启动MCP客户端：# 1. "uv" 是启动器（可能是uvicorn的简写，用于运行ASGI应用）# 2. ["run", mcp_server_path] 是传递给uv的参数，意为运行mcp_server.py# 3. async with ... 确保客户端使用完后自动关闭资源（如连接）async with MCPClient("uv", ["run", mcp_server_path]) as client:# 调用MCP客户端的call_tool方法，传入工具名和参数，等待结果返回# 这是实际执行工具调用的核心操作return await client.call_tool(function_name, args)

MCPSERVER 代码如下：和在进阶篇讲的运行MCPSERVER,并和cline进行交互一致

from mcp.server.fastmcp import FastMCP# Initialize FastMCP server
mcp = FastMCP("search_mcp_server", log_level="ERROR")# Constants
@mcp.tool()
async def search(query: str) -> str:"""搜索网络Args:query: 搜索内容"""# 正常情况下，这里应该调用相关 API 做搜索，为了减少代码的复杂度，# 这里我们返回一段假的工具执行结果，用以测试return "来自 MCP Server 的答案：纽约市今天的天气是晴天，明天的天气是多云。"if __name__ == "__main__":# Initialize and run the servermcp.run(transport='stdio')
http://www.lryc.cn/news/611402.html

相关文章：

  • MySQL间隙锁在查询时锁定的范围
  • MinIO02-Docker安装
  • AI编程新时代：从氛围编程到上下文编程的深度实践和思考
  • GPS信号捕获尝试（上）
  • 快接龙 | 要如何对用户的接龙频次进行系统硬控
  • MongoDB 从3.4.0升级到4.0.0完整指南实战-优雅草蜻蜓I即时通讯水银版成功升级-卓伊凡|bigniu
  • 【文本左右对齐】
  • 【web自动化测试】实战
  • Python基础框架
  • WebRTC音视频编码模块深度解析：从编解码器到自适应码率控制（2025技术实践）
  • 前端包管理器深度对比
  • 普通树状数组
  • 贪心算法学习 1
  • Zabbix 企业级高级应用
  • 风丘助力混合动力汽车工况测试：精准采集整车信号解决方案
  • VNC连接VirtualBox中的Ubuntu24.04 desktop图形化（GUI）界面
  • 2025年渗透测试面试题总结-01（题目+回答）
  • GitHub Models：为开源AI项目解决推理难题，让AI更易用、更普及
  • css初学者第三天
  • MySQL 如何优化慢查询
  • Redis中的sdshdr的len和alloc那块的知识点详解
  • 前端记录项目中用到的js
  • python可视化--Seaborn图形绘制方法和技巧，Bokeh图形绘制方法和技巧
  • 最新基于Python科研数据可视化实践技术
  • 磁悬浮转子振动控制：主动电磁力如何成为高速旋转的“振动克星”
  • css动态样式
  • 【Git学习】入门与基础
  • Cisco 3750X交换机更新到IOS 15.2后无法启动 提示：Boot process failed...
  • Laravel The requested URL /hellowzy was not found on this server. 404 问题的解决
  • 嵌入式 - 数据结构：循环链表和内核链表