网站推广培训中考网络应用服务平台

主要记录自己在调用MCP服务中遇见的问题，这是单个agent，主要作用是调用高德地图的MCP去解决问题。后续会构建多智能体去解决问题。

大模型的API申请，这儿我就懒得写了。我用的是Qwen

高德地图的API地址高德地图MCP的API申请

直接上代码：高德地图MCP服务有多个工具，每个工具都有name和描述。我循环打印出来了。大模型会根据问题以及工具描述去调用相应工具

from datetime import datetime
import time
from langgraph.checkpoint.memory import InMemorySaver
from mcp.server.fastmcp import FastMCP
# from mcp.tools import Tool, ToolsRoute
import sys
import os
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from langchain_mcp_adapters.client import MultiServerMCPClient
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
import asyncio
load_dotenv("/home/sevnce/zy/sevnce_AI/.env")import loggingfrom logging.handlers import TimedRotatingFileHandlerdef set_logger(name, log_path):logHandler = TimedRotatingFileHandler(log_path, when="D",interval=30,backupCount=1,utc=False,encoding="utf-8")#日志中时间显示格式 年-月-天 时：分：秒logFormatter = logging.Formatter('%(asctime)s %(levelname)-8s %(message)s','%Y-%m-%d %H:%M:%S')logHandler.setFormatter(logFormatter)logger = logging.getLogger(name)logger.addHandler(logHandler)#设置日志显示等级logger.setLevel(logging.INFO)return loggerlogger = set_logger("weather", "./log/weather_server.log")
mcp = FastMCP("Weather", port=8001)
gaode_map_api_key = os.getenv("GaoDe-map_API_KEY")client = MultiServerMCPClient({"amap-amap-sse": {"url": f"https://mcp.amap.com/sse?key={gaode_map_api_key}","transport": "sse"}}   # type: ignore
)llm = ChatOpenAI(base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",model="qwen-plus",temperature=0,api_key=os.getenv("Qwen-plus_API_KEY")  # type: ignore
)async def get_weather(imput: str) -> str:sys_prompt = "你是一个高德地图工具，你需要根据用户的输入，返回对应的信息"checkpointer = InMemorySaver()from langchain_core.messages import SystemMessage, HumanMessagetry:tools = await client.get_tools()agent = create_react_agent(llm, tools, prompt=sys_prompt, checkpointer=checkpointer)logger.info(f"Weather_server: 获取到的工具列表: {[[tool.name, tool.description] for tool in tools]}")config = {"configurable": {"thread_id": "1"  }}async for chunk in agent.astream({"messages": [HumanMessage(content=imput)]}, config=config):# 你可以在这里处理每个 chunk，比如打印、yield、写文件等print(chunk)return "流式输出结束"# logger.info(response)# logger.info(response["messages"][-1].content)# # logger.info(f"Weather_server: 天气服务查询结果是：{response_content}")# return str(response["messages"][-1].content)except Exception as e:logger.error(f"Weather_server: 调用天气服务时出错: {str(e)}")response = f"抱歉，天气服务调用失败: {str(e)}"return responseif __name__ == "__main__":start_time = time.time()result = asyncio.run(get_weather("我现在在成都的红牌楼，想要去重庆洪崖洞旅游，请帮我整理一条旅游路线以及攻略"))print("-"*100)# print(result)end_time = time.time()print(f"处理时间: {end_time - start_time:.2f}秒")