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

MCP 第三波升级！Function Call 多步调用 + 流式输出详解

web 2025/7/15 20:57:44

前言

mcp调用

调用方式比较

function call方式调用

调用过程的流式输出

结语

下期内容预告

前言

前面两篇介绍了MCP基础调用与现有Webapi转MCP。接下来我们会继续改进：

1）通过function call方式实现mcp调用

这里我们也会介绍下，通过提示词与function call调用实现的比较

2）实现调用的流式输出，方便前端对接

mcp调用

调用方式比较

一开始我们是通过提示词的方式实现的mcp，后来改为了function call方式，这里我们做个比较。

比较	提示词	function call
简易度	实现简单，直接提示词拼接tools即可	相对简单，按照大模型调用格式，传递tools即可
可控性	较差，可控性依赖大模型能力与提示词	由大模型自己控制，不需要提示词干预，更可控
可扩展性	较弱，随着tool越来越多，维护会越发困难	较好，不需要做程序的改动，mcp tool自动转为大模型tools调用
token消耗	较多，tools都在提示词里，会消耗大量token	较多，大模型tools也会消耗token，不过就是节省了提示词的token消耗
大模型支持度	相对更通用，但是依赖提示词的控制	有些大模型不支持function call或支持不够好

function call方式调用

将mcp tool转为大模型的tools格式

def convert_mcp_tool_to_openai_tool(mcp_tool):"""将 MCP 工具转换为 OpenAI 的 function call 格式"""return {"type": "function","function": {"name": mcp_tool.name,"description": mcp_tool.description,"parameters": mcp_tool.inputSchema,},}

通过function call方式调用大模型

  async def chat(self, prompt, role="user"):"""与LLM进行交互，返回包含 tool_call 的完整响应"""if prompt != "":self.messages.append({"role": role, "content": prompt})response = await self.client.chat.completions.create(model=self.model,messages=self.messages,tools=self.f_tools,  # 使用工具定义stream=True,tool_choice="auto",  # 或者指定具体工具名)return response

判断与调用toll call

async def chat_loop(self, input) -> AsyncGenerator[Dict[str, Any], None]:"""运行交互式聊天循环"""response = await self.chat(input)while True:tool_call_response = Falsefunction_name, args, text = "", "", ""tool_call = {}async for chunk in response:delta = chunk.choices[0].deltaif delta.tool_calls:tool_call = delta.tool_calls[0]tool_call_response = Trueif not function_name:function_name = delta.tool_calls[0].function.nameargs_delta = delta.tool_calls[0].function.arguments# print(args_delta)  # 打印每次得到的数据if args_delta:  # 追加args = args + args_deltaelif delta.content:tool_call_response = Falsetext_delta = delta.contentyield {"type": "result", "content": text_delta}# text = text + text_deltaif tool_call_response:# tool_call = tool_calltool_name = function_nametool_args = json.loads(args)tool_call.function.name = function_nametool_call.function.arguments = args# 执行工具并处理流式输出async for event in self.exec_tool(tool_name, tool_args):# 如果是进度更新，则直接发送if event["type"] == "progress":yield eventelif event["type"] == "error":# 错误情况下也返回错误信息yield eventelif (event["type"] == "tool_start" or event["type"] == "tool_finish"):# 开始、完成调用yield eventelse:# 添加 tool_call 和 tool_response 到 messagesself.messages.append({"role": "assistant","content": None,"tool_calls": [tool_call.model_dump()],})# 添加 tool_response 到 messagesself.messages.append({"role": "tool","name": tool_call.function.name,"content": str(event["content"]),})yield eventresponse = await self.chat("")else:break

上面代码做下解释说明：

response = await self.chat(input) 这里是第一次调用大模型，然后在循环里判断，大模型返回内容（注意这里是流式输出模式，所以用到async for迭代器），如果是tool_calls，则从流式输出中获取调用function name和调用参数args。

然后就可以通过mcp调用tool了，调用结果再去调用大模型，如此反复。

网上关于mcp调用的案例，多数只能单步调用，在这里我们实现了多步调用。

代码中while True，就是要一直判断，直到大模型返回不是tool_calls（就是输出了最终结果，不需要再执行工具调用了）。这个时候就中断了循环。

这里还有一点要注意，在完成tool调用后，需要把工具调用信息、工具调用结果都拼到大模型messages里，这样大模型才能根据这些消息，进行下一步的推理。

调用过程的流式输出

从上面代码也可以看出来，调用工具的代码也是yield流式输出，而且设置了type，这样方便前端区分是思考调用过程、还是最终输出结果。

如下是我测试的的代码，已经是可以流式输出的效果。后续再和对话api对接到前端就行了。

async def main():host = Nonetry:host = Host()await host.connect_mcp_servers()# input = "查下北京的天气，再告诉我怎么从天安门去故宫"# await host.chat_loop("查下北京的天气，再告诉我怎么从天安门去故宫")# 改为可以多次对话while True:user_input = input("请输入：")if user_input == "/x":breakasync for event in host.chat_loop(user_input):print(event)except Exception as e:print(f"主程序发生错误: {type(e).__name__}: {e}")# 打印完整的调用堆栈traceback.print_exc()finally:# 无论如何，最后都要尝试断开连接并清理资源print("\n正在关闭客户端...")await host.disconnect_mcp_servers()print("客户端已关闭。")