使用Trae编辑器与MCP协议构建高德地图定制化服务

发布于:2025-06-22 ⋅ 阅读:(26) ⋅ 点赞:(0)

目录

一、使用Trae编辑器配置高德MCP Server 

1.1 Trae介绍

1.2 从mcp.so中获取配置高德地图mcp server配置信息

1.3 高德地图开发者配置

1.4 添加Filesystem 到Trae

1.5 使用结果展示

1.6 MCP常见命令行工具和包管理说明

1.7 Function Call工具和MCP技术对比

二、本地开发自己的MCP Server

2.1 MCP Server开发

 将 MCP 添加到Python 项目中,使用 uv 来管理Python 项目。

创建虚拟环境和激活

 将 MCP 添加到项目依赖中

创建weather.py

 AI编辑器配置MCP

测试mcp server

2.2 MCP Client 开发

配置项

环境搭建 

 创建 client.py

2.3 整合测试

本文介绍如何使用Trae编辑器与MCP协议配置高德地图服务并开发定制化MCP Server

一、使用Trae编辑器配置高德MCP Server 

1.1 Trae介绍

Trae是字节跳动于2025年推出的AI原生集成开发环境,提供 Claude 3.5、GPT-4o 等模型免费不限量调用

与 Cursor 对比:Trae 更侧重中文支持、企业级协作及全流程自动化,Cursor 则强于代码调试和复杂项目优化

官网地址:https://www.trae.com.cn/

进入官网直接下载 然后next next 安装即可:

下载后 登录 然后打开长这样:

1.2 从mcp.so中获取配置高德地图mcp server配置信息

访问https://mcp.so/ 找到高德地图官方MCP server 

需要copy这个配置出来

然后在Trae中手动添加

添加到这里就可以  高德key的获取在下面章节

1.3 高德地图开发者配置

地址:https://lbs.amap.com/ 访问后进入控制台 

进入控制台后 初次使用需要个人信息认证 按步骤操作就可以

 先创建应用 (初次使用页面上有新手指引)

然后创建key

copy这个key 放到刚才Trae手动添加mcp server的地方

配置成功会有 √ 可使用标识 。出现红色的×  可以点×图标进去看看错误log

注意:安装(有node环境即可,即npm存在就可以)

npm没装 手动配置这步是会错误的!!!

1.4 添加Filesystem 到Trae

从Trae中添加

现在已经加好了2个

1.5 使用结果展示

选择Builder with MCP  然后再输入问题:

我后天去安徽合肥玩3天的旅行。你规划下攻略 ,需要将行程和天气放一起进行规划 再加上交通工具 然后生成kmTravel.html

查看最后结果:

1.6 MCP常见命令行工具和包管理说明

 npx:Node.js 的 “即跑即用” 执行工具
 npx是npm命令的升级版本,功能强大, 侧重于执行命令的,执行某个模块命令,会自动安装模块,但是重在执行某个命令。

在 MCP 开发中,npx主要用于调试 MCP 服务器或执行与 MCP 协议相关的 Node.js 工具


uv:Rust 编写的 Python 极速包管理器
uv是用 Rust 开发的 Python 包与项目管理器,主打极速创建虚拟环境和依赖隔离。

在 MCP 开发中,uv主要用于管理 Python 实现的 MCP 服务器或工具的依赖环境,适合需要快速创建、切换开发环境的场景

uvx:uv 的 “临时环境执行” 扩展工具

uvx是uv工具链的扩展命令,定位为临时环境执行器。

无需提前创建环境,直接为单次命令创建临时虚拟环境,执行后自动清理,适合一次性工具调用

在 MCP 中,uvx主要用于临时运行 Python 实现的 MCP 工具(如官方的mcp-server-fetch)

1.7 Function Call工具和MCP技术对比

MCP解决的核心问题
协议标准化: 通过定义统一的通信规范(类似 USB-C 接口),将传统 Tool/Function Call 的 “定制化适配” 转化为 “标准化接入”
例如,开发一个 MCP 兼容的数据库服务器后,可被 千问、GPT-4 等所有支持 MCP 的模型调用,无需重复开发

效率提升:传统集成中,开发者需为每个工具编写 API 调用代码(如天气查询需处理 HTTP 请求、JSON 解析等)
MCP 提供标准化 SDK,开发者只需关注业务逻辑,接口适配成本趋近于零。

Function Call
每个工具独立开发接口,形成 “烟囱式” 架构(如为 每个API,数据库、 分别编写适配代码)。
手动串联多个工具调用(如先调用天气 API,再调用地图 API),流程硬编码且难以维护。
绑定特定模型,迁移成本高(如从 GPT-3.5 迁移至 Claude 需重写所有工具调用逻辑)。

优先选择 MCP 的场景
复杂企业级应用:需集成多个异构系统(如 ERP、CRM、邮件、数据库),且未来可能扩展新工具。
跨模型部署:需支持多 LLM 供应商(如同时使用 Claude 和 GPT)。
敏感数据处理:涉及医疗、金融等受监管数据,需严格权限控制和审计

优先选择 Function Call 的场景
快速原型开发:简单功能验证(如天气查询、计算器),追求最短开发周期。
封闭环境应用:工具逻辑完全内置于当前系统(如内部知识库问答),无需外部交互。
深度依赖特定模型:需利用专有功能(如 GPT-4 的代码解释器)。

混合架构策略
对于大型项目,可采用 “MCP+Function Call” 混合架构:
核心业务流程:通过 MCP 协议实现标准化集成(如数据库、API)。
边缘功能模块:使用 Function Call 快速实现轻量级工具调用(如简单计算器)。
这种策略既能享受 MCP 的扩展性与安全性,又能保留 Function Call 的敏捷性。

二、本地开发自己的MCP Server

基于高德地图接口,开发定制化的MCP Server

参考官方文档:https://modelcontextprotocol.io/introduction

@mcp.tool() 装饰器

用于标记工具函数,将普通Python函数注册为MCPServer可调用的服务方法。

通常配合协议解析器使用,自动暴露函数到服务接口。

@mcp.resource 装饰器

用于定义资源型接口(类似RESTful API),处理HTTP请求或自定义协议的资源操作。

通常与URL路由或协议动作绑定,支持CRUD操作。

2.1 MCP Server开发

本节内容是在window11上操作的 。MacOS/Linux 有些命令不一样 。见官网demo

 将 MCP 添加到Python 项目中,使用 uv 来管理Python 项目。

#安装uv环境

pip install uv

uv init nnw-mcp-server-demo
cd nnw-mcp-server-demo

 然后再vscode中选择文件夹打开即可

创建虚拟环境和激活

uv venv

# 激活虚拟环境
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

 目录前面是项目名称就可以

 将 MCP 添加到项目依赖中

uv add mcp[cli] httpx

创建weather.py
from typing import Any, Optional
import httpx
from mcp.server.fastmcp import FastMCP


mcp = FastMCP("amap-weather")

# 高德地图API常量 就是上个章节的那个Key
AMAP_API_BASE = "https://restapi.amap.com/v3"
AMAP_API_KEY = "2f92a69cxxxxxxxxxxxxxxxxxxx"
USER_AGENT = "amap-weather-app/1.0"


async def make_amap_request(endpoint: str, params: dict) -> dict[str, Any] | None:
    """
    通用高德API请求函数(带错误处理)
    Args:
        endpoint: API接口路径(如"/weather/weatherInfo")
        params: 请求参数(自动附加key和基础参数)
    """
    print("正在发起高德API请求...make_amap_request")
    base_params = {"key": AMAP_API_KEY, "output": "JSON"}
    full_params = {**base_params, **params}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                f"{AMAP_API_BASE}{endpoint}",
                params=full_params,
                headers={"User-Agent": USER_AGENT},
                timeout=30.0,
            )
            response.raise_for_status()  # 检查HTTP状态码(非200抛异常)
            data = response.json()

            # 处理高德API业务错误码(如key无效、参数错误)
            if data.get("status") != "1":
                err_msg = data.get("info", "未知错误")
                print(f"高德API请求失败:{err_msg}")
                return None
            return data
        except Exception as e:
            print(f"请求异常:{str(e)}")
            return None


def format_realtime_weather(data: dict) -> str:
    """格式化实时天气数据为可读文本"""
    realtime = data["lives"][0]  # 高德实时天气数据在lives数组中
    return f"""
    城市:{realtime['city']}
    实时天气:{realtime['weather']}({realtime['temperature']}°C)
    湿度:{realtime['humidity']}%
    风向:{realtime['winddirection']}
    风速:{realtime['windpower']}级
    更新时间:{realtime['reporttime']}
    """


def format_forecast(data: dict) -> str:
    """格式化未来天气预报数据为可读文本"""
    forecasts = []
    for day_data in data["forecasts"][0]["casts"]:  # 高德预报数据在casts数组中
        forecast = f"""
        日期:{day_data['date']}
        白天天气:{day_data['dayweather']}
        夜间天气:{day_data['nightweather']}
        最高温:{day_data['daytemp']}°C
        最低温:{day_data['nighttemp']}°C
        风向:{day_data['daywind']}
        风力:{day_data['daypower']}级
        """
        forecasts.append(forecast)
    return "\n---\n".join(forecasts)


@mcp.tool()
async def get_realtime_weather(city_adcode: str) -> str:
    """
    获取中国城市实时天气(通过高德地图API)
    Args:
        city_adcode: 城市编码(如北京110000,上海310000,可通过高德城市编码表查询)
    """
    print("正在获取实时天气...get_realtime_weather")
    data = await make_amap_request(
        endpoint="/weather/weatherInfo",
        params={
            "city": city_adcode,
            "extensions": "base",
        },  # extensions=base表示实时天气
    )

    if not data or not data.get("lives"):
        return "无法获取实时天气数据(请检查城市编码或API Key)"

    return format_realtime_weather(data)


@mcp.tool()
async def get_forecast(city_adcode: str) -> str:
    """
    获取中国城市未来多天天气预报(通过高德地图API)
    Args:
        city_adcode: 城市编码(如北京110000,上海310000,可通过高德城市编码表查询)
    """
    data = await make_amap_request(
        endpoint="/weather/weatherInfo",
        params={"city": city_adcode, "extensions": "all"},  # extensions=all表示预报
    )

    if not data or not data.get("forecasts"):
        return "无法获取天气预报数据(请检查城市编码或API Key)"

    return format_forecast(data)


if __name__ == "__main__":
    # 启动MCP服务器(通过标准输入输出通信)
    mcp.run(transport="stdio")
 AI编辑器配置MCP
{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": [
        "--directory",
        "C:\\soft\\test\\nnw-mcp-server-demo",
        "run",
        "weather.py"
      ]
    }
  }
}

配置成功后是这样的  

测试mcp server

正常调用实时工具

正常调用预报工具

 到这mcp server demo就开发结束啦 ~

2.2 MCP Client 开发

配置项

DASHSCOPE_API_KEY = "sk-xxx"  # 阿里云DashScope平台申请的API Key
TONGYI_MODEL = "qwen-plus"    # 使用的通义千问模型

环境搭建 

同上面mcp server的一致 

 创建 client.py

代码流程参考MCP官方案例:https://modelcontextprotocol.io/quickstart/client

import asyncio
import json
import sys
from typing import Optional
from contextlib import AsyncExitStack
import httpx

# ClientSession:MCP客户端会话;StdioServerParameters:标准输入输出服务端配置参数
from mcp import ClientSession, StdioServerParameters

# 标准输入输出客户端,用于通过控制台与服务端通信(适用于本地进程间通信)
from mcp.client.stdio import stdio_client

DASHSCOPE_API_KEY = "sk-090303xxxxxxxxxxxxxxxx"  # 阿里云DashScope平台API Key
TONGYI_MODEL = "qwen-plus"  # 使用qwen-plus模型


class MCPClient:
    """MCP客户端,用于连接服务端并通过LLM调用工具"""

    def __init__(self):
        """初始化MCP客户端和通义千问API连接"""
        # MCP客户端会话(用于与服务端通信,初始为None,连接后赋值)
        self.session: Optional[ClientSession] = None
        # 异步资源栈(用于自动管理后续创建的异步上下文,如连接、会话等,确保资源正确释放)
        self.exit_stack = AsyncExitStack()
        self.tools = {}  # 存储可用工具的字典
        self.tool_descriptions = []  # LLM可用的工具描述

    async def connect_to_server(self, server_script_path: str):
        """
        连接到MCP服务端

        Args:
            server_script_path: 服务端脚本路径(.py或.js)
        """
        # 验证脚本类型
        is_python = server_script_path.endswith(".py")
        is_js = server_script_path.endswith(".js")

        if not (is_python or is_js):
            raise ValueError("服务端脚本必须是.py或.js文件")

        # 确定执行命令
        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command, args=[server_script_path], env=None
        )

        # 建立与服务端的连接
        print(f"正在连接到服务端: {server_script_path}...")
        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )

        await self.session.initialize()

        # 获取并展示可用工具
        response = await self.session.list_tools()
        self.tools = {tool.name: tool for tool in response.tools}

        # 美化工具列表输出
        print("\n✓ 已成功连接到服务端")
        print("🔧 可用工具:")
        for tool_name in self.tools:
            print(f"  - {tool_name}")

        # 为LLM准备工具描述
        self.tool_descriptions = [
            {
                "name": tool.name,
                "parameters": (
                    json.loads(tool.inputSchema)
                    if isinstance(tool.inputSchema, str)
                    else tool.inputSchema
                ),
                "description": tool.description,
            }
            for tool in response.tools
        ]

    async def call_tool(self, tool_name: str, parameters: dict) -> str:
        """
        调用MCP服务端的工具

        Args:
            tool_name: 工具名称
            parameters: 工具参数

        Returns:
            工具执行结果
        """
        if tool_name not in self.tools:
            raise ValueError(f"未知工具: {tool_name}")

        # 调用前显示进度
        print(f"🚀 正在调用工具: {tool_name}")
        print(f"🔧 参数: {json.dumps(parameters, ensure_ascii=False, indent=2)}")

        result = await self.session.call_tool(tool_name, parameters)
        return result.content

    async def query_llm(self, user_query: str) -> dict:
        """
        调用通义千问API解析用户查询为工具调用指令

        Args:
            user_query: 用户查询内容

        Returns:
            解析后的工具调用指令字典
        """
        print(f"🤖 正在分析查询: {user_query}")
        api_url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"

        # 系统提示词:指导模型使用工具
        system_prompt = f"""
        你是一个天气信息查询助手。用户的问题将由以下工具回答:
        
        {json.dumps(self.tool_descriptions, indent=2, ensure_ascii=False)}
        
        请直接生成JSON格式的工具调用指令,不要包含其他解释。
        
        示例:
        用户输入:"上海明天的天气如何?" → 
        {{
            "name": "get_forecast", 
            "parameters": {{
                "city_adcode": "310000"
            }}
        }}
        """.strip()

        headers = {
            "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
            "Content-Type": "application/json",
        }

        payload = {
            "model": TONGYI_MODEL,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_query},
            ],
            "parameters": {
                "temperature": 0.1,  # 极低随机性
                "max_tokens": 128,  # 限制生成长度
            },
        }

        # 发送请求并处理响应
        async with httpx.AsyncClient() as client:
            print("🌐 正在请求LLM服务...")
            response = await client.post(api_url, headers=headers, json=payload)
            response.raise_for_status()

            result = response.json()
            tool_call = (
                result.get("choices", [{}])[0].get("message", {}).get("content", "{}")
            )

            try:
                print("✓ 成功获取工具调用指令")
                return json.loads(tool_call)
            except json.JSONDecodeError:
                print(f"⚠️ 无效的工具调用格式: {tool_call}")
                return None

    async def process_query(self, user_query: str) -> str:
        """
        处理用户查询,调用LLM并执行工具

        Args:
            user_query: 用户查询内容

        Returns:
            最终响应结果
        """
        # 获取工具调用指令
        tool_call = await self.query_llm(user_query)

        if not tool_call or "name" not in tool_call or "parameters" not in tool_call:
            return "抱歉,无法理解您的查询。"

        tool_name = tool_call["name"]
        parameters = tool_call["parameters"]

        try:
            return await self.call_tool(tool_name, parameters)
        except Exception as e:
            return f"调用工具失败: {str(e)}"

    async def chat_loop(self):
        """交互式聊天循环"""
        # 显示欢迎信息
        print("\n" + "=" * 30)
        print("🌦️ 天气查询助手")
        print("=" * 30)
        print("输入示例:北京明天天气如何?/ 北京今天天气怎么样?")
        print("输入'quit'、'退出'或'bye'结束对话")

        while True:
            try:
                query = input("\n查询: ").strip()

                if query.lower() in ["quit", "退出", "bye"]:
                    print("👋 感谢使用,再见!")
                    break

                if not query:
                    continue

                response = await self.process_query(query)

                # 美化结果输出
                print("\n" + "-" * 30)
                print("📋 查询结果:")
                print("-" * 30)
                print(response)
                print("-" * 30)

            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")

    async def cleanup(self):
        """清理资源"""
        print("🧹 正在清理资源...")
        await self.exit_stack.aclose()
        print("✓ 资源清理完成")


async def main():
    """程序入口点"""
    if len(sys.argv) < 2:
        print("使用方法: python client.py <服务端脚本路径>")
        sys.exit(1)

    print("🚀 天气查询助手启动中...")
    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

2.3 整合测试

本地运行 python client.py weather.py

提问:合肥明天天气怎么样?

输入自然查询内容 就会去调用相应的工具  询问今天会去调用get_realtime_weather 询问明天预报类的则会去调用get_forecast

工具的调用流程

用户输入 → process_query → 调用query_llm获取工具调用指令(如get_forecast)

通过 MCP 协议向服务端发送指令,服务端执行具体工具逻辑(如调用天气 API)

服务端返回结果,客户端解析并展示

总结:

本文介绍了如何利用Trae编辑器与MCP协议,快速配置并开发基于高德地图API的定制化服务。通过具体的步骤指导,我们了解了从Trae环境搭建、MCP Server的配置到实际开发应用的全过程。此外,还探讨了MCP Server和Client的构建方法,以及如何使用Python进行本地测试和集成。


网站公告

今日签到

点亮在社区的每一天
去签到