下面这段代码是整篇文章Zui「血脉喷张」的地方——它把用户请求、LLM 推理以及 MCP 工具链完美串联起来。为避免一次性塞进太多细节，我把关键步骤标注出来阅读时Ke以先跳过细枝末节，只关注流程图：

MCP 客户端初始化：通过 StdioTransport 启动 composition server 子进程，实现本地 IPC 通信。

L​LM 实例化： 使用异步版 OpenAI SDK 指向 DashScope 的兼容端点。

构造消息历史： 将系统提示和用户输入逐层压栈，以便后续多轮对话使用。

获取可调用工具列表： .list_tools 返回所有Yi注册函数元信息，用来填充 LLM 的 tool 参数字段。

L​LM 首轮生成： 若返回的是直接文本，则结束；若出现 tool 调用，则进入循环处理。

MCP 工具调用 & 回写结果： 把 tool 调用信息发送给服务器执行，将返回值重新写入对话历史，再交给 LLM Zuo下一轮推理。

循环直至没有 tool 调用： Zui终得到完整答案后返回给上层视图函数。

资源回收： 关闭子进程和 HTTP 会话，防止泄漏。

import json, asyncio
from pathlib import Path
from typing import List, Dict, Any
from fastmcp.client import Client, StdioTransport
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageFunctionToolCall
from config import cfg
from log import logger
class MCPClient:
    def __init__:
        # 启动本地 composition server 子进程作为通信桥梁
        self.client = Client(
            StdioTransport(
                command=str.parent.parent / ".venv" / "bin" / "python"),
                args=,
                cwd=str.parent / "mcp_servers"),
            )
        )
        self.llm = AsyncOpenAI(base_url=cfg.llm_base_url,
                               api_key=cfg.llm_api_key)
        self.memory: List] =    # 暂存会话历史
    async def close:
        if self.client:
            await self.client.close
    async def _prepare_tools:
        tools_raw = await self.client.list_tools
        tools = 
        for t in tools_raw:
            tools.append({
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema,
                }
            })
        logger.debug} 个可用工具")
        return tools
    async def _call_llm -> Any:
        resp = await self.llm.chat.completions.create(
            model=cfg.llm_model,
            messages=messages,
            tools=tools,
            temperature=0.7,
        )
        return resp
    async def process -> str:
        if system_prompt:
            self.memory.append
        self.memory.append
        async with self.client:
            tools = await self._prepare_tools
            response_texts: List = 
            while True:
                raw_resp = await self._call_llm
                msg = raw_resp.choices.message
                # 若有直接内容则收集起来
                if getattr:
                    response_texts.append
                # 检查是否有 tool 调用需求，没有则跳出循环结束对话。
                if not getattr:
                    break
                for call in msg.tool_calls:
                    fn_call: ChatCompletionMessageFunctionToolCall = call  # 类型转换帮助 IDE 提示
                    name = fn_call.function.name
                    args = json.loads
                    logger.info
                    tool_result = await self.client.call_tool
                    # 把调用记录写回历史，以便 LLM Neng基于结果继续推理。
                    self.memory.append({
                        "role":"assistant",
                        "tool_calls":
                    })
                    self.memory.append({
                        "role":"tool",
                        "tool_call_id":fn_call.id,
                        "content":str)
                    })
            return "
".join

from flask import Response,jsonify
class APIResponse:
   default_mimetype='application/json'
   def __init__:
       payload=dict
       super.__init__,status=status,mimetype=self.default_mimetype)

import json
from http import HTTPStatus
from flask import Flask, request
from aiagent.mcp_client import MCPClient
app = Flask
@app.get
def health:
   """健康检查，用于容器编排平台"""
   return APIResponse
@app.post
async def chat:
   """
   异步视图：接收用户 Prompt → 调用 MCPClient → 返回Zui终答案。
   注意：Flask 会在内部为每个请求创建独立事件循环。
   """
   try:
       body: dict = request.get_json
       prompt = body.get
       if not isinstance:
           raise ValueError
   except Exception as e:
       return APIResponse)
   client=MCPClient
   try:
       answer=await client.process
       return APIResponse
   except Exception as exc:
       logger.exception
       return APIResponse)
   finally:
       await client.close
if __name__=="__main__":
   app.run

💡 小贴士：Ru果你习惯使用 Gunicorn 部署，只需要把上面的入口改成 ``gunicorn -k gevent -w 4 main:app`` 即可。gevent 为每个 worker 提供协程调度，让异步视图跑得稍微顺畅一点儿，不过仍然受限于每个 worker 同时只Neng处理一条请求这件事儿 😅 。

启动服务：

python main.py      # 开发模式下直接跑，会kan到类似 “Running on http://127.0.0.1:8000” 的日志。
# 或者使用生产模式：
gunicorn main:app -w 4 -k gevent --bind 0.0.0.0:8000 --worker-connections 1000 

Curl 发起一次对话请求：

curl -X POST http://127.0.0.1:8000/chat \
     -H 'Content-Type: application/json' \
     -d '{"prompt":"今天北京几点了？"}'

{"code":200,"msg":"success","data":{"answer":"现在是2026-05-05T14:31:00+0800。"}}

观察服务器日志：

✅ Nengkan到 “Available mcp tools” 与 “Calling tool …” 的信息，说明 Tool 被成功触发； ❗ Ru果没有出现任何 Tool 调用，请检查MCP Server 是否正常启动且被正确导入到 composition 中。

P.S. 若想验证高并发下表现，可借助 ApacheBench 或 wrk 对 ``/chat`` 接口进行压测。记住每个 worker 在异步模式下仍旧只Neng处理一条请求，所以提升并发需要增加 worker 数量或改用原生 ASGI 框架，如 FastAPI。

从代码到部署，我们Yi经完成了以下几件事儿：

MCP 工具被封装成独立 Python 函数，并通过 fastMCP 自动生成 RPC 描述；

AIO‑LLM 客户端利用 OpenAI 标准接口完成“先思考—再调工具—再回答”的闭环；

AIO‑Flask 路由让外部 HTTP 请求Neng够无缝触达上述闭环；

KISS 原则驱动下我们只用了 gunicorn + gevent 就完成了生产级部署脚本。

⚠️ 切记：Flask 虽然Yi经支持异步，但底层仍然是 WSGI，一旦面对高并发 I/O 密集型场景，它会出现“线程+事件循环”的双重开销。Ru果你的项目预计每天要处理上万次聊天请求，强烈建议直接迁移到 FastAPI 或者 Quart，它们天生就是 AsyncIO 原生跑分机器 🚀 。但Ru果你只是想在Yi有 Flask 项目里加一点点 AI 小功Neng，这套方案完全够用了而且改动Zui小、学习成本低！

my_flask_mcp/
├─ aiagent/
│  ├─ __init__.py               # 空文件，使其成为包  
│  ├─ mcp_client.py             # 核心客户端实现  
│  └─ mcp_servers/
│     ├─ __init__.py           
│     ├─ common.py              # 示例工具函数  
│     └─ composition.py         # 合并子 Server  
├─ config.py                     # 配置中心  
├─ log.py                        # 日志封装  
├─ main.py                       # Flask 程序入口  
└─ requirements.txt              # 可选：列出依赖版本