MCP终极篇！MCP Web Chat项目实战分享
目录
前言
MCP Web Chat
功能概要说明
MCP Web Chat代码调用结构说明
api动态生成MCP Server
方法一（之前的方法）
方法二（现在的方法）
做个比较
相关代码
相关问题解决说明
稳定性
由此引申而来的异步任务问题
MCP周期问题（待讨论）
结语
前言
前面三篇文章，循序渐进的从MCP概念、初步使用到多tool调用到现有api动态生成mcp server，逐步像实际项目迈进。
本篇是MCP终极篇。将介绍如何实现MCP的Web Chat，以及过程中遇到的一些问题与解决。相关代码，我已经在Github开源。【github地址：loli0123456789/MCPWebChat: 基于MCP的WebChat】
本篇主要内容：
-
功能概要说明
-
MCP Web Chat主体架构设计说明
通过该部分内容，你可以更好的理解为何如此规划设计
-
api动态生成MCP Server 功能说明
这块会说下动态生成MCP Server的两种思路与比较
-
相关问题解决说明
MCP系列文章，可通过如下链接快速回看：
全网少有-通过Python调用MCP_python mcp-CSDN博客
MCP第二弹，支持Webapi调用与动态MCP【附完整代码】-CSDN博客
MCP 第三波升级！Function Call 多步调用 + 流式输出详解-CSDN博客
MCP Web Chat
功能概要说明
1、通过数据库或配置文件管理MCP Server
2、内置通用MCP Server，如获取当前时间、访问网页内容等，可自行根据需要添加
3、内置根据api信息自动生成tool的MCP Server
4、可以根据获取的MCP tools进行Web界面对话，对话可流式显示工具调用过程、最终结果
MCP Web Chat代码调用结构说明
在网上的多数MCP代码案例中，多数MCP的类是Server、Client，其中Client负责连接Server，同时负责通过大模型关联tools进行chat对话。这里Server只有在需要自己创建MCP Server的情况才需要，因此主要就是Client。
但是这里有两个问题：
1）把大模型放到Client里，有多个Server需要连接怎么办？
2）有多个用户要进行chat对话，消息如何管理？
基于以上问题，我对此的规划是：MCP_Client、MCP_Host、MCP_Chat。
MCP_Client：用于定义MCP Client
注意： 后续为了解决MCP Client长时间会挂掉的问题，同时为了降低自己写代码的复杂性，Client直接使用了百度appbuilder实现的MCP_Client功能。
MCP_Host：负责连接多个MCP_Client，全局单例
MCP_Chat：负责关联MCP_Host，并对接用户Chat，每次对话会实例化一个MCP_Chat对象，实现每个对话话的消息记录单独管理
一些MCP相关的通用方法，会放到mcp_utils文件。
在chat api层面，就是直接和MCP_Chat实例化对象交互，同时会把最新消息记录传递到对象，从而实现多轮对话。
对于流式输出，就是在MCP_Chat调用过程中，把相关信息都通过yeild的方式给流式输出，api层面再传出去就可以了。前端流式输出的效果，对content消息增加type区分，就可以知道是思考过程，还是最终结果。
api动态生成MCP Server
之前文章里这块内容也写过，但是当时实现方式不够理想，后来又换了一个更底层，更可控的方法。正好两个方法可以对比说明下。
方法一（之前的方法）
1）根据api信息，动态生成调用方法
2）通过mcp.add_tool添加1)中方法
方法二（现在的方法）
使用mcp官方的底层server方法，可以自定义list_tool和call_tool方法，这样灵活性、可控性会更高。也就是说只要把列出tool的方法和调用tool的方法实现，就可以了。
做个比较
1）传递给大模型的tool信息完善度
通过自定义list_tool，可以自定义传递给大模型的tools信息，这样你可以设置更完善的function信息，包括参数信息、必填信息等。而通过mcp.add_tool()添加的工具，tool信息是底层自己设置，可能信息不够全。
对于大模型来说，只有给的信息足够全，那么最终的效果才会更好。
2）调用tool的可控性
通过自定义call_tool，可以更好的控制如何调用工具，对于需要传递鉴权信息，或者有些api调用特殊的情况，都可以很好的自我控制。同样的，通过mcp.add_tool()添加的工具，调用的底层自己去调用，灵活度就比较低了。
相关代码
async def init_tools():tools = []for api in api_configs:# 编码，可作为tool名称code = api.get("name", "")# 方法描述description = api.get("discription", "")# 实际调用接口地址api_url = api.get("api_url", "")# 入参input_params = api.get("input_params", "")tool = types.Tool(name=code,description=description,inputSchema=process_input_param(input_params),)tools.append(tool)return tools@mcp.list_tools()
async def list_tools() -> list[types.Tool]:tools = await init_tools()return tools
如上代码，你可以自己控制tool的详细参数信息，给尽可能完整的信息，让大模型可以更好的理解上下文。
async def request_api(name: str, arguments: dict):"""请求api"""api_url = ""for api in api_configs:if api["name"] == name:api_url = api["api_url"]breakelse:raise ValueError(f"Unknown tool: {name}")result = await post_form_data(api_url, arguments, token=None)return result@mcp.call_tool()
async def fetch_tool(name: str, arguments: dict) -> list[types.Content]:print(f"name: {name}")print(f"arguments: {arguments}")result = await request_api(name, arguments)return [types.TextContent(type="text", text=result)]
如上代码，你可以更好的控制如何去调用api，传递token等参数。
相关问题解决说明
过程中遇到的一个主要问题就是MCP连接的稳定性与周期问题。
稳定性
高德MCP Server（amap）在连接一段时间后会不可用：
1）几分钟不调用，再调用的时候会报aclose错误（它自动关闭连接了，但是我自己开发的MCP Server不会如此）
解决 增加心跳检测，每2分钟ping一次，MCP有相应的方法send_ping()（设置3分钟不行，这个时候已经失效…）
2）虽然有了心跳检测，但是过了几个小时候后，再次调用会报错（大意是验证信息过期）
解决 每2个小时，重启一次MCP Server（自己开发的MCP Server没这个问题，因为也没做鉴权信息，当然我也没针对某个MCP Server这么控制重启）
由此引申而来的异步任务问题
MCP Client底层使用了sse_client用到了task_group，然后FastAPI框架也是异步的。基于task_group的设计，在它下面发起的任务，只能在它这里取消，就很容易报各种cancelscope错误。
比如底层创建了一个sse_client，结果在系统重启MCP Server（需要先关闭再启动连接）的时候，就报错了；心跳同样是在关闭MCP Client的时候需要关闭，也会有一样的问题。
解决的办法，就是套娃，在创建心跳任务、重启任务之外，都加一个task_group或cancel_scope，这样就不会导致一直向上传递导致报错的问题。
async def initialize(self):"""延迟初始化（避免启动时阻塞）"""if not self._initialized:self._task_group = anyio.create_task_group()await self._task_group.__aenter__()self._task_group.start_soon(self.connect_mcp_servers)# await self.connect_mcp_servers()self._initialized = True# 使用 anyio 创建取消范围self._heartbeat_cancel_scope = anyio.CancelScope()self._heartbeat_task = asyncio.create_task(self._run_with_cancel_scope())async def _run_with_cancel_scope(self):"""在取消范围内运行心跳任务"""with self._heartbeat_cancel_scope:await self._start_heartbeat()
如上代码，对于心跳任务，主要是取消，所以创建一个cancelScope就好。对于connect_mcp_servers给他创建一个rask_group。
在disconnect_mcp_servers的时候，还需要每个Client调用各自的task_group去clean_up
for i, client in enumerate(self.mcp_clients):try:log.info(f"开始清理客户端 {i}")# client._session_context._task_groupclient._session_context._task_group.start_soon(client.cleanup)# self._task_group.start_soon(client.cleanup)log.info(f"已启动清理客户端 {i}")except AttributeError as e:log.error(f"清理客户端 {i} 时出现属性错误: {e}")traceback.print_exc()except Exception as e:log.error(f"清理客户端 {i} 时出现异常: {type(e).__name__}: {e}")traceback.print_exc()
MCP周期问题（待讨论）
这个问题，我目前也不确定哪种方式就是合理
1）程序启动的时候，MCP Server就都启动连接，一直到程序关闭的时候关闭连接或定时重启
好处：用到的时候调用速度会快
问题：可能用不到，造成资源浪费
2）在对话用到了某个MCP的时候，才去启动连接
好处：节省资源，但是多个对话用到了某个MCP要不要创建多个Client？
问题：响应可能会慢，需要先去连接，获取tools
3）有一个MCP连接池？在2）的基础上启动，不断放到池子里，通过1）的方式维护
这个问题，欢迎各位讨论交流
结语
今天主要讲了如何将MCP 打造为一个项目实用的Web Chat应用。项目不仅仅是调用MCP进行chat，还支持直接创建通用方法MCPServer，以及基于现有api创建动态tool的MCPServer。
在文章的后半部分，讲了下项目过程中遇到的一些问题以及如何解决，还有一些问题留待各位交流探讨。
相关代码已经开源到Github， 【github地址：loli0123456789/MCPWebChat: 基于MCP的WebChat】。