Langchain mcp 可视化界面

界面

依赖包

"gradio>=6.13.0","langchain>=1.2.15","langchain-mcp-adapters>=0.2.2","langchain-openai>=1.1.12","langgraph-cli[inmem]>=0.4.21",

代码

importasyncioimportuuidfromtypingimportList,DictimportgradioasgrfromgradioimportChatMessagefromlangchain.agentsimportcreate_agentfromlangchain_core.messagesimportHumanMessage,AIMessage,ToolMessagefromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlanggraph.typesimportCommandfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.chat_modelsimportinit_chat_modelfromdotenvimportload_dotenv,find_dotenvimportos load_dotenv(find_dotenv())ZHIPU_API_KEY=os.getenv("ZHIPU_API_KEY")# ModelScope 上的 Qwen3-32B（OpenAI 接口兼容）QWEN3_32B="Qwen/Qwen3-32B"# 优先读取 ModelScope 的 API Key；若未配置则回退到 OPENAI_API_KEY（兼容常见环境）API_KEY=os.getenv("MODELSCOPE_API_KEY")oros.getenv("OPENAI_API_KEY")or""model=init_chat_model(model=QWEN3_32B,model_provider="openai",base_url="https://api-inference.modelscope.cn/v1/",api_key=API_KEY,temperature=0,# 通过 extra_body 传入思考/推理开关与流式选项（content_blocks 将包含 reasoning 与 text）extra_body={# 常见思考模式参数的并集，未识别的键会被后端忽略"enable_thinking":False},)# 网络搜索MCP服务端（工具的配置）zhipuai_mcp_server_config={"url":"https://open.bigmodel.cn/api/mcp/web_search/sse?Authorization="+ZHIPU_API_KEY,"transport":"sse",}# https://modelscope.cn/mcp/servers/@Joooook/12306-mcp 部署mcp_12306_server_config={"url":"https://mcp.api-inference.modelscope.net/xxx/sse",'transport':'sse'}# https://modelscope.cn/mcp/servers/antvis/mcp-server-chart 部署mcp_chart_server_config={"url":"https://mcp.api-inference.modelscope.net/xxx/sse",'transport':'sse'}# 创建一个MCP的客户端去连接mcp_client=MultiServerMCPClient({"zhipuai_mcp_server_config":zhipuai_mcp_server_config,"my12306_mcp_server_config":mcp_12306_server_config,"chart_mcp_server_config":mcp_12306_server_config,})# 这些MCP服务端的工具，只支持异步模式，所以只能用MCP的异步模式去调用asyncdefcreate_new_agent():mcp_tools=awaitmcp_client.get_tools()# 获取MCP的所有toolsreturncreate_agent(model,tools=mcp_tools,system_prompt='你是一个智能助手，尽可能的调用工具回答用户的问题',checkpointer=InMemorySaver()# 创建一个内存的保存器： 保存对话上下文)agent=asyncio.run(create_new_agent())# 配置参数，包含会话IDconfig={"configurable":{# 检查点由session_id访问"thread_id":str(uuid.uuid4()),}}# res = agent.invoke(input={'messages': [HumanMessage(content='你好！')]}, config=config)# print(res)defadd_message(chat_history,user_message):ifuser_message:chat_history.append({"role":"user","content":user_message})returnchat_history,gr.Textbox(value=None,interactive=False)asyncdefsubmit_messages(chat_history:List[Dict]):"""流式处理消息的核心函数"""user_input=chat_history[-1]['content']current_state=agent.get_state(config)full_response=""# 累积完整响应tool_calls=[]# 记录工具调用# 处理中断恢复或正常消息inputs=Command(resume={'answer':user_input})ifcurrent_state.nextelse{'messages':[HumanMessage(content=user_input)]}asyncforchunkinagent.astream(inputs,config,stream_mode=["messages","updates"],# 同时监听消息和状态更新):if'messages'inchunk:formessageinchunk[1]:# 处理AI消息流式输出ifisinstance(message,AIMessage)andmessage.content:full_response+=message.content# 更新最后一条消息而非追加ifchat_historyandisinstance(chat_history[-1],ChatMessage)and'title'notinchat_history[-1].metadata:chat_history[-1].content=full_responseelse:chat_history.append(ChatMessage(role="assistant",content=message.content))yieldchat_history# 处理工具调用消息elifisinstance(message,ToolMessage):tool_msg=f"🔧 调用工具:{message.name}\n{message.content}"chat_history.append(ChatMessage(role="assistant",content=tool_msg,metadata={"title":f"🛠️ Used tool{message.name}"}))yieldchat_history# 创建Gradio界面withgr.Blocks(title='我的智能小秘书',theme=gr.themes.Soft(),css=".system {color: #666; font-style: italic;}"# 自定义系统消息样式)asdemo:# 聊天历史记录组件chatbot=gr.Chatbot(height=500,render_markdown=True,# 支持Markdown格式line_breaks=False# 禁用自动换行符)# 输入组件chat_input=gr.Textbox(placeholder="请输入您的消息...",label="用户输入",max_lines=5,container=False)# 控制按钮withgr.Row():submit_btn=gr.Button("发送",variant="primary")clear_btn=gr.Button("清空对话")# 消息提交处理链msg_handler=chat_input.submit(fn=add_message,inputs=[chatbot,chat_input],outputs=[chatbot,chat_input],queue=False).then(fn=submit_messages,inputs=chatbot,outputs=chatbot,api_name="chat_stream"# API端点名称)# 按钮点击处理链btn_handler=submit_btn.click(fn=add_message,inputs=[chatbot,chat_input],outputs=[chatbot,chat_input],queue=False).then(fn=submit_messages,inputs=chatbot,outputs=chatbot)# 清空对话clear_btn.click(fn=lambda:[],inputs=None,outputs=chatbot,queue=False)# 重置输入框状态msg_handler.then(lambda:gr.Textbox(interactive=True),None,[chat_input])btn_handler.then(lambda:gr.Textbox(interactive=True),None,[chat_input])if__name__=='__main__':demo.launch()