普通智能体接入高德MCP
mcp协议官网：https://modelcontextprotocol.info/zh-cn/
关于什么是mcp建议大家看一下MCP官网就可以，首页的描述就非常贴切
“AI应用的USB-C接口” 接口通用，功能强大
接入流程
官网示例：https://docs.langchain.com/oss/python/langchain/mcp
- 添加 langchain-mcp-adapters 包来将 MCP 服务中的工具适配成 agent 使用的工具
- getTools 方法是异步的
接入代码实现
""" 【简单创建智能体,并将MCP服务工具，注入给智能体】 可参考官方文档地址：https://docs.langchain.com/oss/python/langchain/mcp """importasyncioimportosfromdotenvimportload_dotenvfromlangchain_core.messagesimportHumanMessagefromlangchain_core.toolsimporttoolfrompydanticimportBaseModel,Fieldfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportcreate_agent load_dotenv()# 1. 创建本地工具classAddInputArgs(BaseModel):a:float=Field(description="第一个数字",example=1)b:float=Field(description="第二个数字",example=2)@tool(description="对输入的两个数字执行加法运算",args_schema=AddInputArgs,return_direct=False)defadd(a,b)->float:returna+b# 2. 获取 MCP 工具asyncdefget_mcp_tools():"""异步获取 MCP 工具,这里get_tools是异步的"""mcp_client=MultiServerMCPClient({"amap-maps-streamableHTTP":{"url":"https://mcp.amap.com/mcp?key="+os.getenv("AMAP_KEY"),"transport":"http"},})returnawaitmcp_client.get_tools()# 3. 主异步函数asyncdefmain():# 获取 MCP 工具print("正在获取 MCP 工具...")mcp_tools=awaitget_mcp_tools()# 汇总工具列表total_tools=[add]+mcp_toolsprint(f"✅ 所有工具列表（共{len(total_tools)}个）:")fori,toolinenumerate(total_tools,1):print(f"{i}.{tool.name}")# 4. 初始化大模型llm=ChatOpenAI(model="qwen-max-latest",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key=os.getenv("DASHSCOPE_API_KEY"),streaming=True,)# 5. 创建智能体agent=create_agent(model=llm,tools=total_tools,system_prompt="你是一个智能助手，可以调用高德 MCP 工具来帮助用户完成日常的一些需求")# 6. 调用智能体（使用异步调用！）print("\n🚀 开始调用智能体...")# ✅ 关键修改：使用 ainvoke() 而不是 invoke()result=awaitagent.ainvoke(input={"messages":[HumanMessage(content="我现在在上海市闵行区正矩科技大厦，我的车牌是‘浙AFQ5247’,我想开车导航去人民广场，请你给我规划下路线和时间。因为有禁行时间段！")]})# 7. 完成调用后打印，如果需要实时流式输出，则需要创建 create_react_agent 这里的普通智能体，只能事后打印消息记录print("\n"+"="*60)print("💬 对话记录:")print("="*60)formsginresult['messages']:t=type(msg).__name__ift=='HumanMessage':print(f"👤{msg.content}")elift=='AIMessage':ifhasattr(msg,'tool_calls')andmsg.tool_calls:fortool_callinmsg.tool_calls:args_str=', '.join(f'{k}={v}'fork,vintool_call['args'].items())print(f"🤖 调用{tool_call['name']}({args_str})")elifmsg.content:print(f"🤖{msg.content}")elift=='ToolMessage':print(f"🔧{msg.name}→{msg.content[:100]}...")# 只显示前100字符print("="*60)returnresult# 8. 运行主函数if__name__=="__main__":# 使用 asyncio.run() 启动异步主函数asyncio.run(main())创建自己的MCP服务
通过 FastMCP 创建
官网链接：https://docs.langchain.com/oss/python/langchain/mcp#custom-servers
""" 【自定义MCP】 官网：https://docs.langchain.com/oss/python/langchain/mcp#custom-servers """fromfastmcpimportFastMCP mcp=FastMCP("Math")@mcp.tool()defadd(a:int,b:int)->int:"""Add two numbers"""returna+b@mcp.tool()defmultiply(a:int,b:int)->int:"""Multiply two numbers"""returna*bif__name__=="__main__":mcp.run(transport="streamable-http")使用mcp时的配置 JSON
{"mcpServers":{"fastmcp":{"url":"http://127.0.0.1:8000/mcp"}}}
