MCP 第三波升级！Function Call 多步调用 + 流式输出详解
目录
前言
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("客户端已关闭。")
结语
这就是今天讲的内容。主要就是对提示词和function call方式进行mcp调用做了一个比较、然后通过function call实现了mcp调用，再实现为流式输出效果，为对接前端做准备。
下期内容预告
整体功能整合，实现一个可用的mcp chat api，可以对接前端进行聊天，可以显示思考过程、最终结果等内容。 过程中遇到的问题与解决，也会做一分享。
不知面前的读者是否有类似的问题，欢迎关注与交流。
我们下期见~_~