前言
近年来，大量新兴的 AI 相关第三方库都提供了异步接口，有些甚至出于性能考虑仅支持异步调用，例如 MCP SDK。伴随着 Python 异步编程的发展，FastAPI 等框架迅速流行，许多新项目倾向于采用 FastAPI。但实际上，Flask 自 2.0 版本起也开始支持异步方法，因此我们也能借助 Flask 参与到 MCP 的 AI Agent 浪潮中。
PS：需要注意的是，Flask 的异步性能表现不佳。正如 Flask 官方文档所指出的那样，每当收到请求时，Flask 都会在一个线程内启动一个新的事件循环来执行异步视图函数，然后返回结果。这种方式增加了额外的性能开销。经过个人实测，Flask 的异步视图性能确实不如传统的同步视图。因此，如果有异步需求，建议还是选用像 FastAPI 这样原生支持异步的框架。
Async functions require an event loop to run. Flask, as a WSGI application, uses one worker to handle one request/response cycle. When a request comes in to an async view, Flask will start an event loop in a thread, run the view function there, then return the result.
Each request still ties up one worker, even for async views. The upside is that you can run async code within a view, for example to make multiple concurrent database queries, HTTP requests to an external API, etc. However, the number of requests your application can handle at one time will remain the same.
Async is not inherently faster than sync code. Async is beneficial when performing concurrent IO-bound tasks, but will probably not improve CPU-bound tasks. Traditional Flask views will still be appropriate for most use cases, but Flask's async support enables writing and using code that wasn't possible natively before.
本文使用的 LLM 是阿里的通义千问，需要自行申请 API Key。其他兼容 OpenAI SDK 的模型理论上也可使用。
本文介绍的是如何在 Flask 中集成基于 MCP Client 和 MCP Server 的 AI Agent，并不仅仅是用 Flask 开发一个 MCP Server，所以只关注 Flask 实现 MCP Server 的看众可以关闭本文了。
安装 SDK
虽然 Flask 从 2.0 版本开始支持异步功能，但这部分功能需要额外安装相关依赖。这里我们将 MCP 和 LLM 相关的依赖一起安装。
uv add 'flask[async]' fastmcp openai
此外还需要安装 gunicorn 和 gevent，这两个是在生产环境中常用的部署工具。虽然是开发演示项目，但我们也会安装它们来验证异步功能的支持情况。
uv add gunicorn gevent
代码示例
代码结构
由于这只是个演示项目，所以代码结构相对简单。
├── aiagent # aiagent 模块
│ ├── mcp_client.py # 封装 mcp 的 client
│ └── mcp_servers # mcp 的 servers
│ ├── common.py # 会被导入到 composition 的 mcp server 中
│ └── composition.py # composition 的 mcp server, client 只会连这个 mcp server
├── config.py # 配置模块
├── log.py # 日志模块
├── main.py # 入口文件
├── pyproject.toml
├── README.md
└── uv.lock
配置模块
简单写写，能用就行，注意替换 API Key。config.py
class Config:@propertydef llm_base_url(self) -> str:return "https://dashscope.aliyuncs.com/compatible-mode/v1"@propertydef llm_model(self) -> str:return "qwen-plus"@propertydef llm_api_key(self) -> str:return "<your api key>"cfg = Config()
日志模块
简单写写，能用就行。log.py
import logging
import sysdef setup_logger() -> logging.Logger:level = logging.DEBUGlogger = logging.getLogger("flask-mcp")logger.setLevel(level)handler = logging.StreamHandler(sys.stdout)handler.setLevel(level)fmt = "%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"datefmt = "%Y-%m-%d %H:%M:%S"formatter = logging.Formatter(fmt, datefmt)handler.setFormatter(formatter)logger.addHandler(handler)return loggerlogger = setup_logger()
MCP Servers
common
我们可以将这个 common mcp server 视为子服务器，它将被导入到父级 composition server 中统一管理。
在 common mcp server 中，以下仅实现了一个用于获取当前时间的工具函数。
from datetime import datetimefrom fastmcp import FastMCPmcp = FastMCP(name="common mcp server", instructions="Common MCP server for general tasks.")@mcp.tool
async def get_current_datetime() -> str:"""Get the current date and time as a string. Format: YYYY-MM-DDTHH:MM:SS±hhmm"""return datetime.now().strftime("%Y-%m-%dT%H:%M:%S%z")if __name__ == "__main__":mcp.run(transport="stdio", show_banner=False)
composition
其他的 mcp server 都会组合到这个 composition mcp server 中。一方面 mcp client 只需要连这一个 composition mcp server 即可，另一方面可以按功能组织 mcp server 的代码。
import asyncio
import sys
from pathlib import Pathsys.path.append(str(Path(__file__).parents[2]))
from fastmcp import FastMCPfrom aiagent.mcp_servers.common import mcp as common_mcpcomposition_mcp = FastMCP(name="composition mcp server")async def compose():await composition_mcp.import_server(common_mcp)if __name__ == "__main__":asyncio.run(compose())composition_mcp.run(transport="stdio", show_banner=False)
MCP Client
aiagent/mcp_client.py 负责整合 LLM 与 MCP Server 的交互。示例中的对话记忆仅存储在实例变量中，在实际应用中应考虑持久化存储方案。
import json
from pathlib import Path
from typing import castfrom fastmcp.client import Client, StdioTransport
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageFunctionToolCallfrom config import cfg
from log import loggerclass MCPClient:def __init__(self):self.client = Client(StdioTransport(command=str(Path(__file__).parent.parent / ".venv" / "bin" / "python"),args=[str(Path(__file__).parent / "mcp_servers" / "composition.py")],cwd=str(Path(__file__).parent / "mcp_servers"),))self._llm = AsyncOpenAI(base_url=cfg.llm_base_url,api_key=cfg.llm_api_key,)self._temp_memories = []async def close(self):if self.client:await self.client.close()async def process(self, prompt: str, system_prompt: str = "") -> str:if system_prompt:self._temp_memories.append({"role": "system", "content": system_prompt})self._temp_memories.append({"role": "user", "content": prompt})async with self.client:tools = await self.client.list_tools()available_tools = []for tool in tools:available_tools.append({"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.inputSchema,}})logger.info(f"Available mcp tools: {[tool.name for tool in tools]}")resp = await self._llm.chat.completions.create(model=cfg.llm_model,messages=self._temp_memories,tools=available_tools,temperature=0.3,)# 存储最终响应文本final_text = []# 获取 LLM 的首个响应消息message = resp.choices[0].message# 如果响应包含直接内容，则添加到结果中if hasattr(message, "content") and message.content:final_text.append(message.content)# 循环处理工具调用，直到没有更多工具调用为止while message.tool_calls:# 遍历所有工具调用for tool_call in message.tool_calls:# 确保工具调用有函数信息if not hasattr(tool_call, "function"):continue# 类型转换以获取函数调用详情function_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)function = function_call.functiontool_name = function.name# 解析函数参数tool_args = json.loads(function.arguments)# 检查 MCP 客户端是否已连接if not self.client.is_connected():raise RuntimeError("Session not initialized. Cannot call tool.")# 调用 MCP 服务器上的指定工具logger.info(f"Calling tool: {tool_name} with args: {tool_args}")result = await self.client.call_tool(tool_name, tool_args)# 将助手的工具调用添加到消息历史中self._temp_memories.append({"role": "assistant","tool_calls": [{"id": tool_call.id,"type": "function","function": {"name": function.name,"arguments": function.arguments}}]})# 将工具调用结果添加到消息历史中self._temp_memories.append({"role": "tool","tool_call_id":tool_call.id,"content": str(result.content) if result.content else ""})# 基于工具调用结果再次调用 LLMfinal_resp = await self._llm.chat.completions.create(model=cfg.llm_model,messages=self._temp_memories,tools=available_tools,temperature=0.3,)# 更新消息为最新的 LLM 响应message = final_resp.choices[0].message# 如果响应包含内容，则添加到最终结果中if message.content:final_text.append(message.content)# 返回连接后的完整响应return "\n".join(final_text)
main
main.py 是应用程序的入口文件，其中实现了健康检查的同步视图和处理聊天请求的异步视图。此外还自定义了响应类，确保 Flask 能正确处理非英文字符。
import json
from http import HTTPStatus
from typing import Optionalfrom flask import Flask, Response, requestfrom aiagent.mcp_client import MCPClientapp = Flask(__name__)class APIResponse(Response):def __init__(self, data: Optional[dict] = None, code: HTTPStatus = HTTPStatus.OK, msg: str = "success"):headers = dict({"Content-Type": "application/json; charset=utf-8"})response = json.dumps({"code": code.value,"msg": msg,"data": data,}, ensure_ascii=False, default=str)super().__init__(response=response, status=code.value, headers=headers)@app.get("/health")
def health_check():return APIResponse(data={"status": "ok"})@app.post("/chat")
async def post_chat():try:req_body: Optional[dict] = request.get_json()if not req_body or "prompt" not in req_body:return APIResponse(code=HTTPStatus.BAD_REQUEST, msg="Missing 'prompt' in request body")prompt = req_body.get("prompt")if not isinstance(prompt, str):return APIResponse(code=HTTPStatus.BAD_REQUEST, msg="'prompt' must be a string")except Exception as e:return APIResponse(code=HTTPStatus.BAD_REQUEST, msg=str(e))mcp_client = MCPClient()try:resp = await mcp_client.process(prompt=prompt)resp_body = {"content": resp}return APIResponse(data=resp_body)except Exception as e:return APIResponse(code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=str(e))finally:await mcp_client.close()if __name__ == "__main__":app.run(host="127.0.0.1", port=8000)运行测试
- 首先启动服务端应用
python main.py
- 使用 curl 发起请求进行测试。可以看到接口成功返回了答案。通过服务端控制台日志可以看出 mcp client 成功调用了
get_current_datetime工具。
$ curl --request POST \
--url http://127.0.0.1:8000/chat \
--header 'content-type: application/json' \
--data '{
"prompt": "今天的日期是什么"
}'
{"code": 200, "msg": "success", "data": {"content": "今天的日期是 2025 年 12 月 8 日。"}}# Server 端控制台日志
2025-12-08 21:55:50 | flask-mcp | INFO | mcp_client.py:51 | process | Available mcp tools: ['get_current_datetime']
2025-12-08 21:55:51 | flask-mcp | INFO | mcp_client.py:88 | process | Calling tool: get_current_datetime with args: {}
- 使用 gunicorn 启动应用，测试 gunicorn 对异步方法的支持情况。
gunicorn main:app -n 127.0.0.1:8000 -w 4 -k gevent --worker-connections 1000
- 再次使用 curl 测试。测试依然正常。
$ curl --request POST \--url http://127.0.0.1:8000/chat \--header 'content-type: application/json' \--data '{"prompt": "现在是什么时候？"
}'
{"code": 200, "msg": "success", "data": {"content": "现在是 2025 年 12 月 8 日 22 时 37 分 50 秒。"}}
小结
通过上述示例可以看出，Flask 仍然能够胜任基于 MCP 的 AI Agent 应用开发任务。而且 Flask 2.0 之后的版本与之前版本保持良好的兼容性，因此可以考虑将旧项目升级到新版。不过需要再次强调的是，Flask 的异步视图性能并不理想，对于新的 AI Agent 项目，建议优先选择原生支持异步的框架。
