目录

前言

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】。