# 进阶示例：构建全能出行助手（集成模型/记忆/沙箱/网关/高德mcp） #### 场景概述 在[基础示例](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_056.html)中，我们已创建一个简易对话智能体，并将其托管部署至云上AgentArts环境。但在真实生产环境下，企业级智能体需要具备持久化记忆、安全沙箱执行、外部系统无缝对接等能力。 本示例将基于LangGraph框架构建一个出行规划智能体，核心实现agent.py的TravelAgent类。该类封装LangGraph ReAct工作流（agent 节点完成 LLM 推理，tools 节点负责工具执行）、MCP工具动态发现、云端记忆持久化、代码沙箱调用等特性；同时集成AgentArts平台的Gateway工具调用、代码沙箱执行、云端记忆持久化三大核心能力，部署于AgentArts运行时，对外提供多轮对话式出行规划服务。 用户输入自然语言指令，例如 "查杭州天气"、"帮我算两天行程预算"，Agent 会自主选择工具与调用顺序，最终输出结构化回复。 本示例将构建 "全能出行助手" 智能体，重点演示如下能力： - 记忆库（Memory）：对接AgentArts云端记忆库组件，提供长短期记忆能力，完成跨会话上下文持久化。用户开启新会话时，可加载历史对话记录，接续完成出行规划。 - 网关（Gateway）：作为Agent调用外部工具的统一入口，依托标准MC 协议转换能力对接外部工具；服务启动时动态拉取全部工具定义。示例接入高德地图15个MCP工具，支持天气查询、地理编码、多模式路径规划（驾车 / 步行 / 骑行 / 公交）、周边搜索、距离测量等功能。 - 工具（CodeInterpreter）：依托平台完全隔离的容器环境，创建安全沙箱会话，执行LLM生成的Python代码并返回运算结果。在出行规划场景下，可完成预算核算（门票 + 住宿 + 餐饮 + 交通汇总）、多套方案费用对比、行程增量调整等工作。 - 身份认证（Identity）：与网关协同，完成工具链路的标准化鉴权，保障外部工具调用安全。 #### 架构图与工作流程 #### 整体架构图 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719008917.png "点击放大") #### 工作流程 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689175824.png "点击放大") #### 环境准备 - 操作系统：Ubuntu ARM64，服务器可访问公网。 ![](https://support.huaweicloud.com/highcode-agentarts/public_sys-resources/caution_3.0-zh-cn.png) 需使用ARM64系统制作的Agent镜像，使用X86系统制作的镜像在调用智能体运行时时会调用失败。 可以使用华为云ECS服务购买服务器，购买时需要选择鲲鹏架构（ARM64），操作系统选择Ubuntu。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161240.png "点击放大") - 安装Python：请确保Python 3.10及以上版本已安装。 大多数Linux发行版（如Ubuntu）都预装了Python，您可以先通过python3 --version检查。如未安装，可以使用如下命令安装： ``` sudo apt update sudo apt install python3 ``` - 安装Docker：请确保Docker 18.06及以上版本已安装。如未安装，可以使用如下命令安装： ``` # 查询 Docker 版本 docker --version # 安装Docker sudo apt update sudo apt install docker.io ``` 安装完成后，建议立即配置国内镜像加速器。在后续执行agentarts launch构建镜像时，Docker需要拉取python:3.10-slim基础镜像，默认从docker.io官方仓库下载，可能超时失败。提前配置加速器可避免此问题。 执行以下命令创建Docker配置目录： ``` sudo mkdir -p /etc/docker ``` 执行以下命令配置国内镜像源（直接复制并回车执行）： ``` sudo tee /etc/docker/daemon.json <<-'EOF' { "registry-mirrors":[ "https://docker.m.daocloud.net", "https://dockerproxy.net", "https://mirror.baidubce.com" ] } EOF ``` 依次执行以下命令重启Docker服务，使配置生效： ``` sudo systemctl daemon-reload sudo systemctl restart docker ``` ![](https://support.huaweicloud.com/highcode-agentarts/public_sys-resources/note_3.0-zh-cn.png) 华为云SWR基础版不支持OCI镜像格式，如果您使用的是Docker 27及以上版本，并且需要处理OCI镜像，可以通过设置环境变量来关闭OCI支持。 ``` export DOCKER_BUILDKIT=0 # 或者 export BUILDKIT_USE_OCI_MEDIA_TYPES=0 ``` - 安装SDK。 1. 在Linux系统中创建项目目录并进入。 ``` mkdir ./test2 cd ./test2 ``` 2. 执行以下命令安装SDK（建议在Python虚拟环境中安装，以避免与系统包产生冲突）。 ``` # 安装依赖包 apt install python3.12-venv # 创建并激活虚拟环境 (linux) python3 -m venv venv source venv/bin/activate # 安装sdk pip install agentarts-sdk ``` ![](https://support.huaweicloud.com/highcode-agentarts/public_sys-resources/note_3.0-zh-cn.png) - 执行pip install agentarts-sdk命令下载缓慢、超时可以更改为使用如下命令 ``` pip install agentarts-sdk -i https://repo.huaweicloud.com/repository/pypi/simple --trusted-host repo.huaweicloud.com ``` - 如果系统缺少python3-venv包，导致无法创建虚拟环境，请按照命令回显提示安装python3-venv包。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000819.png "点击放大") - 执行以下命令配置华为云凭证，获取华为云凭证请参考[认证鉴权](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_047.html)。 ``` export HUAWEICLOUD_SDK_AK="your-access-key" export HUAWEICLOUD_SDK_SK="your-secret-key" ``` #### 智能体开发全流程 本示例涉及的代码样例可通过[代码示例](https://res-static.hc-cdn.cn/cloudbu-site/china/zh-cn/agentarts/高代码开发/code_new.zip)全量下载，也可直接从帮助文档中复制使用。 #### 步骤一：平台组件创建与配置 在AgentArts平台依次创建记忆库、沙箱工具和网关，并获取后续代码接入所需的各项配置信息。 1. 在AgentArts平台创建记忆库。 1. 登录[AgentArts智能体平台](https://console.huaweicloud.com/agentarts/#/home/overview)。 2. 在左侧导航栏选择"开发中心 \> 组件库 "，进入"记忆库"界面。 3. 单击"创建记忆库"。填写记忆库名称、描述；勾选"私网访问"、"公网访问"；勾选全部的"长期记忆提取策略"，其余参数使用默认配置。 图1创建记忆库 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321114.png "点击放大") 图2配置记忆策略 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880693.png "点击放大") 4. 记忆库创建完成后，单击记忆库名称，进入"基本信息"界面，**获取记忆库ID** 。 图3获取记忆库ID ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161242.png "点击放大") 5. 在"API Key"处单击"更新"。 图4更新API Key ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000821.png "点击放大") 6. 弹框中单击"确定"后，**复制或下载API Key** 。 图5复制或下载API Key ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321116.png "点击放大") 2. 在AgentArts平台创建沙箱工具。 1. 在AgentArts平台左侧导航栏中选择"开发中心 \> 组件库"，并进入"沙箱工具"页面，单击"创建代码解释器"。 2. 填写代码解释器相关配置。参考下表进行配置。 表1代码解释器配置说明 | 参数 | 说明 | |:---|:---| | 名称 | 可自定义。 | | 描述 | 可自定义。 | | 委托（可选） | （可选）授予的代理权限或代理功能，允许代表智能体与外部系统进行通信和交互。 可不填写，如需手动创建，请单击"创建委托"并参考如下配置创建。 - 委托名称：自定义。 - 信任主体类型：选择"云服务"。 - 云服务：搜索service.WorkloadSandboxMetadata。 - 其余参数使用默认值。 图6创建委托 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880697.png "点击放大") | | 入站身份认证 | 选择"**API Key认证**"。 | | API Key名称 | 可自定义。 | | 日志记录 | 开启。 | | 出网网络配置 | 选择"公网访问"。 | 图7创建沙箱工具 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161244.png "点击放大") 3. 配置完成后，单击"立即创建"创建沙箱。 在沙箱工具列表中，记录名称。 单击沙箱工具，进入"配置信息"页面，获取域名。 在"配置信息"页面，单击URN链接，在新页面获取沙箱API Key。 图8获取沙箱工具名称 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000823.png "点击放大") 图9获取沙箱域名 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321118.png "点击放大") 图10获取沙箱API Key ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880699.png "点击放大") 3. 在AgentArts平台创建网关，配置高德MCP服务。 1. 高德MCP如下，使用前，请参考[高德开放平台文档](https://lbs.amap.com/api/mcp-server/create-project-and-key)，创建获取API Key。 ``` { "mcpServers": { "amap-maps-streamableHTTP": { "url": "https://mcp.amap.com/mcp?key=您在高德官网上申请的key" } } } ``` 2. 在AgentArts平台左侧导航栏中选择"开发中心 \> 组件库"，并进入"网关"页面，单击"创建网关"。 3. 填写网关基础信息并配置权限与身份认证。参考下表进行配置。 表2网关基础信息与权限身份认证配置 | 参数 | 说明 | |:---|:---| | 名称 | 可自定义。 | | 描述（可选） | 网关的描述信息，可自定义。 | | MCP 版本 | 2025-03-26 | | 委托 | 使用平台的默认值。 | | 入站身份认证 | 选择API Key认证。 | | API Key名称 | 可自定义。 | | 日志记录 | 选择开启。 | | 工具检索 | 选择开启。 | | 高级配置 | 选择公网访问，开启会话保持。 | 图11填写网关基础信息与权限身份认证 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161246.png "点击放大") 4. 单击"创建Target"，配置高德MCP服务。 表3高德Target配置 | 参数 | 说明 | |:---|:---| | 名称 | 填写为gaodemap。 | | 描述（可选） | 可自定义。 | | 类型 | 选择"MCP"，使用现成的高德MCP服务，不需要使用直接对接外部REST API的方式。 | | 传输方式 | 选择"Streamable HTTP"，与高德MCP的调用方式保持一致。 | | MCP地址 | 填写：https://mcp.amap.com/mcp | | 出站身份认证 | 选择"API Key"，单击"创建出站身份"，参数配置如下： - 身份名称：填写为gaodemcp。 - 认证类型：选择"API Key"。 - API Key的值：填写高德开发平台中创建的API Key。 出站身份创建完成后，返回"创建Target"页面，选择已创建的出站身份。 - **位置：选择"查询参数"。** - **参数名称：填写为key。** - **前缀：不填。** 图12创建出站身份 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000825.png "点击放大") 图13配置出站身份认证 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321120.png "点击放大") | 5. 配置完成后，单击"确定"创建Target。Target创建完成后，回到"创建网关"页面，高级配置选择"公网访问"后，单击"创建网关"。 6. 网关创建完成后，回到网关列表页面。单击网关名称，记录网关URL及网关的API Key（注意是网关的API Key，不是高德mcp的key）。 图14网关列表 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880701.png "点击放大") 7. 调测网关。 MCP网关配置完成后，建议在"调测"页面进行一次测试，验证MCP工具是否可以正常调用。在网关详情页面单击"调测"标签，确认能够正常返回结果后再进行后续开发。 图15调测网关 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002742073987.png "点击放大") 8. 在网关详情页面获取网关URL。 图16获取网关URL ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000827.png "点击放大") 9. 在网关详情页面，单击URN链接。进入Agent Identity页面，获取网关的API Key。 图17单击URN链接 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321122.png "点击放大") 图18获取网关API Key ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880703.png "点击放大") #### 步骤二：编写Agent核心逻辑 使用LangGraph框架编写智能体的核心对话逻辑，并集成记忆库、MCP 网关、沙箱工具与华为云MaaS模型。 1. 执行如下命令安装langchain、langgraph、langchain-openai。 ``` pip install -U langchain langgraph langchain-openai ``` 2. 创建.env环境变量配置文件。 ``` # ========================================== # 1. 模型配置 (对接华为云MaaS模型) # ========================================== MODEL_NAME=deepseek-v4-flash MODEL_URL=https://api.modelarts-maas.com/openai/v1 MODEL_API_KEY=替换为真实的模型API Key MODEL_TYPE=openai # ========================================== # 2. AgentArts 记忆库配置 # ========================================== MEMORY_SPACE_ID=替换为记忆库 ID HUAWEICLOUD_SDK_MEMORY_API_KEY=替换为记忆库的API Key HUAWEICLOUD_SDK_REGION=cn-southwest-2 # ========================================== # 3. 网关配置 # ========================================== # 网关地址 GATEWAY_ENDPOINT=替换为网关的URL # 网关的 apikey GATEWAY_INBOUND_TOKEN=替换为网关的API Key # ========================================== # 4. 沙箱配置 # ========================================== SANDBOX_NAME=在AgentArts平台中创建的沙箱名称 AGENTARTS_CODEINTERPRETER_DATA_ENDPOINT="http://您的沙箱域名" HUAWEICLOUD_SDK_CODE_INTERPRETER_API_KEY="您的沙箱 API Key" # ========================================== # 5. Agent 全局配置 # ========================================== MAX_ITERATIONS=50 SESSION_TTL=3600 # 本地测试启动 HTTP Server 时绑定的端口 AGENT_RUN_PORT=8080 ``` 获取模型API Key的方法请参考[获取华为云MaaS服务模型API Key](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_053.html)。 agent.py文件的示例中，使用华为云MaaS服务提供的deepseek-v4-flash模型，如果使用其他模型注意MODEL_NAME的值，需填写为模型接口中model参数的值，可以参考如下方式获取。注意需要选择OpenAI兼容接口。 图19获取OpenAI兼容接口的model值 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161250.png "点击放大") 3. 在本地创建agent.py文件，编辑代码如下。 示例代码中创建了出行助手智能体，并对接记忆库、网关、沙箱、身份认证、华为云MaaS服务提供的模型。 ``` """ ================================================================================ AgentArts LangGraph 智能体核心实现 ================================================================================ 启动时自动从Gateway 拉取 tools/list，动态生成 LangChain Tool 对象。 ================================================================================ """ import os import uuid import json import re from typing import List, Optional, Dict, Any, Type import requests import urllib3 from dotenv import load_dotenv from pydantic import create_model, Field urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langgraph.graph import MessagesState from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage, ToolMessage from langchain_core.tools import StructuredTool from langchain_openai import ChatOpenAI try: from agentarts.sdk.memory import MemoryClient from agentarts.sdk.memory.inner.config import TextMessage except ImportError: MemoryClient = None TextMessage = None print("[Warning] 未检测到 agentarts.sdk，将跳过云端记忆同步。") load_dotenv() # ================================================================================ # 常量 # ================================================================================ _MAX_ITERATIONS = 12 _FORCE_FINAL_REPLY_ITERATION = 10 _TOOL_RESULT_MAX_LENGTH = 8000 _HISTORY_K_MESSAGES = 20 _REQUEST_TIMEOUT = 30 _ACTOR_ID = "user_123" _ASSISTANT_ID = "travel_agent" _TOOL_CALL_TAG = "" _DEFAULT_MODEL_NAME = "deepseek-v4-flash" _DEFAULT_MODEL_URL = "https://api.modelarts-maas.com/openai/v1" _DEFAULT_REGION = "cn-southwest-2" _DEFAULT_MAX_ITERATIONS = 50 _SYSTEM_PROMPT = ( "你是一个专业的出行规划助手，具备以下能力：\n\n" "【地图服务 - 高德MCP】\n" "- 天气查询：帮用户判断出行日期是否合适\n" "- 地理编码：将地址转为坐标\n" "- 路径规划：计算驾车/步行/骑行/公交路线\n" "- 周边搜索：查找餐厅、酒店、景点等\n\n" "【代码沙箱】\n" "- 仅在用户明确要求计算费用/预算/总花费时使用\n" "- 不适用于：查位置、查路线、查天气、展示信息\n\n" "【云端记忆】\n" "- 记住用户提到的城市、日期、偏好\n" "- 跨轮次累积规划，支持增量调整\n\n" "工作原则：\n" "1. 规划出行时，先查天气判断可行性\n" "2. **何时调用 execute_python**：只有当用户明确要求计算费用、预算、总价、多方案费用对比时才调用。" "查位置、查路线、查天气、展示距离等任务直接用工具返回的结果回答，**不要调用 execute_python**。\n" " - 对于不确定的价格（如门票、住宿、餐饮），用合理估计值\n" " - 在 Python 代码中定义这些变量，用代码做加法/乘法计算总费用\n" " - 禁止直接给出估算数字，必须通过沙箱计算\n" "3. 记住用户之前提到的所有信息，避免重复询问\n" "4. 多轮对话中，基于已有上下文做增量调整\n" "5. 路径规划时，尽量同时查询所有出行方式（驾车、步行、骑行、公交），给用户完整的对比" ) # ================================================================================ # 消息清洗（防止 ModelArts 81001 错误） # ================================================================================ def clean_messages_for_llm(messages: List[BaseMessage]) -> List[BaseMessage]: cleaned = [] expecting_tool_results = False for msg in messages: if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None): cleaned.append(msg) expecting_tool_results = True elif isinstance(msg, ToolMessage): if expecting_tool_results: cleaned.append(msg) else: print("[MaaS Guard] 检测到孤立的 ToolMessage，已自动过滤。") else: cleaned.append(msg) expecting_tool_results = False return cleaned # ================================================================================ # 配置 # ================================================================================ class ModelConfig: def __init__(self): self.name = os.getenv("MODEL_NAME", _DEFAULT_MODEL_NAME) self.url = os.getenv("MODEL_URL", _DEFAULT_MODEL_URL) self.api_key = os.getenv("MODEL_API_KEY") if not self.api_key: raise ValueError("[错误] 未找到 MODEL_API_KEY！请在 .env 文件中进行配置。") class AgentConfig: def __init__(self): self.model = ModelConfig() self.memory_space = os.getenv("MEMORY_SPACE_ID") if not self.memory_space: print("[Warning] 未配置 MEMORY_SPACE_ID，云端记忆库可能无法正常工作。") self.region = os.getenv("HUAWEICLOUD_SDK_REGION", _DEFAULT_REGION) self.max_iterations = int(os.getenv("MAX_ITERATIONS", str(_DEFAULT_MAX_ITERATIONS))) config = AgentConfig() # ================================================================================ # Gateway 交互层（tools/list + tools/call） # ================================================================================ def _mcp_headers() -> Dict[str, str]: return { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "mcp-session-id": str(uuid.uuid4()), "Authorization": f"Bearer {os.getenv('GATEWAY_INBOUND_TOKEN', '')}" } def _parse_mcp_response(raw_text: str) -> dict: """解析 MCP 响应：兼容 SSE (data: {...}) 和纯 JSON 两种格式""" if "data:" in raw_text: match = re.search(r'\{.*\}', raw_text, re.DOTALL) if match: return json.loads(match.group()) return json.loads(raw_text) def _mcp_request(method: str, params: dict = None) -> dict: """通用 MCP JSON-RPC 请求""" gateway_url = os.getenv("GATEWAY_ENDPOINT") if not gateway_url: raise RuntimeError("未配置 GATEWAY_ENDPOINT") payload = { "jsonrpc": "2.0", "id": str(uuid.uuid4()), "method": method, "params": params or {} } resp = requests.post(gateway_url, headers=_mcp_headers(), json=payload, verify=False, timeout=_REQUEST_TIMEOUT) return _parse_mcp_response(resp.text) def list_mcp_tools() -> List[dict]: """第一步：调 tools/list 拉取 Gateway 上所有工具定义（支持 cursor 分页）""" all_tools = [] cursor = None try: while True: params = {"cursor": cursor} if cursor else {} result = _mcp_request("tools/list", params) result_data = result.get("result", {}) all_tools.extend(result_data.get("tools", [])) cursor = result_data.get("nextCursor") if not cursor: break print(f"[MCP] 从 Gateway 发现 {len(all_tools)} 个工具") return all_tools except Exception as e: print(f"[MCP] tools/list 失败: {e}") return all_tools _TRUNCATION_SUFFIX = "\n...(结果已截断，原始数据过长)" _MIN_TRUNCATION_RATIO = 0.8 def _truncate_safely(text: str, max_length: int) -> str: """安全截断：在 JSON 结构边界处截断，避免破坏数据完整性""" if len(text) <= max_length: return text truncated = text[:max_length] last_brace = max(truncated.rfind('}'), truncated.rfind(']')) if last_brace > max_length * _MIN_TRUNCATION_RATIO: return truncated[:last_brace + 1] + _TRUNCATION_SUFFIX return truncated + _TRUNCATION_SUFFIX def call_mcp_tool(tool_name: str, arguments: Dict[str, Any]) -> str: """第四步：模型决定调某工具后，透传到 MCP tools/call""" try: result = _mcp_request("tools/call", {"name": tool_name, "arguments": arguments}) text = json.dumps(result.get("result", result), ensure_ascii=False) return _truncate_safely(text, _TOOL_RESULT_MAX_LENGTH) except Exception as e: return f"网关调用异常: {e}" # ================================================================================ # 第二步：把 MCP 工具定义动态转成 LangChain StructuredTool # ================================================================================ _JSON_TYPE_MAP = { "string": str, "number": float, "integer": int, "boolean": bool, "array": list, "object": dict, } def _json_schema_to_pydantic(schema: dict, model_name: str = "Args") -> Type: """将 MCP inputSchema (JSON Schema) 转为 Pydantic Model 类""" properties = schema.get("properties", {}) required = set(schema.get("required", [])) fields = {} for param_name, param_info in properties.items(): json_type = param_info.get("type", "string") py_type = _JSON_TYPE_MAP.get(json_type, str) desc = param_info.get("description", "") if param_name in required: fields[param_name] = (py_type, Field(..., description=desc)) else: fields[param_name] = (Optional[py_type], Field(None, description=desc)) return create_model(model_name, **fields) if fields else create_model(model_name) def _make_tool_caller(tool_name: str): """闭包工厂：每个工具绑定自己的 tool_name，防止循环变量捕获问题""" def caller(**kwargs): clean_args = {k: v for k, v in kwargs.items() if v is not None} for k, v in clean_args.items(): if isinstance(v, (int, float)): clean_args[k] = str(v) print(f"\n[Gateway] 调用工具: {tool_name} | 参数: {clean_args}") return call_mcp_tool(tool_name, clean_args) return caller def build_langchain_tool(mcp_tool_def: dict) -> StructuredTool: """将单个 MCP 工具定义转为 LangChain StructuredTool""" name = mcp_tool_def["name"] desc = mcp_tool_def.get("description", "") or f"MCP tool: {name}" schema = mcp_tool_def.get("inputSchema", {}) args_model = _json_schema_to_pydantic(schema, f"{name}_Args") caller = _make_tool_caller(name) return StructuredTool.from_function( func=caller, name=name, description=desc, args_schema=args_model, ) def discover_mcp_tools() -> List[StructuredTool]: """完整的工具发现流程：list -> build""" mcp_tools = list_mcp_tools() return [build_langchain_tool(t) for t in mcp_tools] # ================================================================================ # 本地工具：代码解释器（非 MCP，保持独立定义） # ================================================================================ def _execute_python(code: str) -> str: from agentarts.sdk.tools import code_session print(f"\n[Sandbox] 正在安全沙箱中执行代码...") sandbox_name = os.getenv("SANDBOX_NAME") api_key = os.getenv("SANDBOX_API_KEY") or os.getenv("HUAWEICLOUD_SDK_CODE_INTERPRETER_API_KEY") if not sandbox_name: return "沙箱调用失败: 未在环境变量中配置 SANDBOX_NAME" try: with code_session(config.region, sandbox_name, api_key=api_key) as client: res = client.invoke(operate_type="execute_code", api_key=api_key, arguments={"code": code, "language": "python"}) return json.dumps(res.get("result")) except Exception as e: return f"沙箱执行失败: {e}" execute_python = StructuredTool.from_function( func=_execute_python, name="execute_python", description="在平台沙箱环境中安全执行 Python 代码。Args: code: 待执行的Python代码", args_schema=create_model("execute_python_Args", code=(str, Field(..., description="待执行的Python代码"))) ) # ================================================================================ # LangGraph 工作流 # ================================================================================ class AgentState(MessagesState): iteration: int def should_continue(state: AgentState) -> str: if state.get("iteration", 0) >= _MAX_ITERATIONS: print(f"[Safety] 达到最大迭代次数({_MAX_ITERATIONS})，强制结束") return "end" last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "continue" return "end" class TravelAgent: def __init__(self): self._llm = ChatOpenAI( model=config.model.name, base_url=config.model.url, api_key=config.model.api_key, temperature=0.1, ) self._mcp_tools = discover_mcp_tools() self._all_tools = self._mcp_tools + [execute_python] print(f"[Agent] 共加载 {len(self._all_tools)} 个工具（MCP: {len(self._mcp_tools)}, 本地: 1）") self.graph = self._build_graph() def _build_graph(self): def model_node(state: AgentState) -> AgentState: messages = state["messages"] current_iteration = state.get("iteration", 0) if current_iteration >= _FORCE_FINAL_REPLY_ITERATION: print("[Safety] 接近最大迭代次数，强制生成最终回复") final_prompt = SystemMessage(content=( "你已达到工具调用次数上限。请基于已有工具返回的结果，" "直接给出最终的文字回复，不要再调用任何工具。" "如果信息不足，请如实告知用户已获取的部分结果。" )) safe_messages = clean_messages_for_llm(messages + [final_prompt]) response = self._llm.invoke(safe_messages) else: llm_with_tools = self._llm.bind_tools(self._all_tools) safe_messages = clean_messages_for_llm(messages) response = llm_with_tools.invoke(safe_messages) return {"messages": [response], "iteration": current_iteration + 1} workflow = StateGraph(AgentState) workflow.add_node("agent", model_node) workflow.add_node("tools", ToolNode(self._all_tools)) workflow.set_entry_point("agent") workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END}) workflow.add_edge("tools", "agent") # 不使用 checkpointer，每轮从云端记忆库加载历史 return workflow.compile() def run(self, user_input: str, session_id: str) -> str: thread_config = {"configurable": {"thread_id": session_id}} self._ensure_session(session_id) history = self._load_cloud_history(session_id) messages = self._build_messages(history, user_input) prev_count = len(messages) - 1 result = self.graph.invoke( {"messages": messages, "iteration": 0}, config=thread_config, recursion_limit=config.max_iterations ) new_messages = result["messages"][prev_count:] self._save_to_cloud_memory(session_id, new_messages) return self._clean_response(result["messages"][-1].content) def _ensure_session(self, session_id: str): """确保云端记忆会话存在""" if MemoryClient is None or not config.memory_space: return try: MemoryClient().create_memory_session( space_id=config.memory_space, id=session_id, actor_id=_ACTOR_ID ) except Exception as e: if "Session id already exists" not in str(e): print(f"[Warning] 云端记忆会话同步失败: {e}") def _build_messages(self, history: List[BaseMessage], user_input: str) -> List[BaseMessage]: """构建送入 LLM 的消息列表：系统提示 + 历史 + 当前输入""" messages = [SystemMessage(content=_SYSTEM_PROMPT)] messages.extend(history) messages.append(HumanMessage(content=user_input)) return messages def _clean_response(self, response: str) -> str: """清理可能幻觉的工具调用标签""" if _TOOL_CALL_TAG in response: return response[:response.index(_TOOL_CALL_TAG)].strip() return response def _load_cloud_history(self, session_id: str) -> List[BaseMessage]: """从云端记忆加载历史消息""" if MemoryClient is None or not config.memory_space: return [] try: client = MemoryClient() response = client.get_last_k_messages( space_id=config.memory_space, session_id=session_id, k=_HISTORY_K_MESSAGES ) msgs = None if isinstance(response, list): msgs = response elif hasattr(response, 'messages'): msgs = response.messages if msgs is None: print(f"[Memory] 无法提取消息列表，{type(response).__name__}") return [] history = [] for m in msgs: role = getattr(m, "role", "") content = getattr(m, "content", "") if not content and hasattr(m, 'parts') and m.parts: part = m.parts[0] content = part.get('text', '') if isinstance(part, dict) else getattr(part, 'text', '') if role == "user" and content: history.append(HumanMessage(content=content)) elif role == "assistant" and content: history.append(AIMessage(content=content)) if history: print(f"[Memory] 从云端加载 {len(history)} 条历史消息") return history except Exception as e: print(f"[Memory] 加载历史失败: {e}") return [] def _save_to_cloud_memory(self, session_id: str, new_messages: List[BaseMessage]): """增量同步：只写入本轮新增的消息到 AgentArts 记忆库""" if MemoryClient is None or TextMessage is None or not config.memory_space: return try: text_messages = [] for msg in new_messages: if isinstance(msg, HumanMessage) and msg.content: text_messages.append(TextMessage(role="user", content=msg.content, actor_id=_ACTOR_ID, assistant_id=_ASSISTANT_ID)) elif isinstance(msg, AIMessage) and msg.content: text_messages.append(TextMessage(role="assistant", content=msg.content, actor_id=_ACTOR_ID, assistant_id=_ASSISTANT_ID)) if text_messages: MemoryClient().add_messages( space_id=config.memory_space, session_id=session_id, messages=text_messages ) print(f"[Memory] 增量同步 {len(text_messages)} 条新消息到云端") except Exception as e: print(f"[Memory] 云端同步失败（不影响对话）: {e}") ``` #### 步骤三：本地功能验证 在本地直接运行智能体，确认智能体可以正常运行。 1. 创建main.py文件，用于本地运行智能体，并进行测试。 测试脚本为交互式对话，可直接输入问题测试agent.py中集成的基础对话、网关、沙箱工具、记忆恢复等能力。 ``` """ 出行规划助手 - 交互式场景演示 """ import sys import uuid from agent import TravelAgent _EXIT_COMMANDS = {"exit", "quit", "退出"} def safe_input(prompt="你: "): """ 兼容 Cloud Shell 终端编码的输入函数。 Cloud Shell 可能以 GBK 编码发送中文字节，而 Python 默认按 UTF-8 解码会报 UnicodeDecodeError。 策略：先尝试 UTF-8，失败后回退到 GBK，再失败则用 replace 模式兜底。 """ print(prompt, end="", flush=True) raw = sys.stdin.buffer.readline() if not raw: return "" try: return raw.decode("utf-8").strip() except UnicodeDecodeError: try: return raw.decode("gbk").strip() except UnicodeDecodeError: return raw.decode("utf-8", errors="replace").strip() def print_guide(): print("\n" + "=" * 60) print(" 杭州周末出行规划助手") print("=" * 60) print(""" 我可以帮你： - 查天气，判断出行是否合适 - 查地点、规划路线 - 计算行程预算 - 记住你的偏好，持续优化方案 建议对话流程： 1. "我周末想去杭州玩，先帮我看看天气" 2. "查一下西湖和灵隐寺的位置" 3. "这两个地方之间怎么走？多远？" 4. "帮我算一下两天行程的总预算" 5. "如果改成三天呢？住宿多一晚" """) print("=" * 60 + "\n") def main(): print("正在初始化智能体助手（集成网关、代码解释器与云端记忆）...") agent = TravelAgent() session_id = str(uuid.uuid4()) print(f"会话 ID: {session_id}") print_guide() while True: try: user_input = safe_input() if not user_input: continue if user_input.lower() in _EXIT_COMMANDS: print("再见！祝旅途愉快～") break print("\n" + "-" * 40) response = agent.run(user_input, session_id=session_id) print(f"\n助手: {response}\n") print("-" * 40 + "\n") except KeyboardInterrupt: print("\n再见！") break if __name__ == "__main__": main() ``` 2. 将agent.py、.env、main.py上传至服务器项目目录中。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000829.png "点击放大") 3. 执行python main.py命令对智能体进行本地测试。 由于main.py是交互式对话程序，运行后可直接输入问题，智能体会调用大模型进行回复。有正常的回复即表示本地测试成功。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718993883.png "点击放大") **交互场景与组件调用** **问题 1："我周末想去杭州玩，先帮我看看天气"** - 用户意图：查询目的地天气，判断出行日期是否合适 - 调用组件：网关、记忆库 - 组件作用： - MCP Gateway：LLM 决策调用 maps_weather 工具，Gateway 转发请求到高德天气服务，返回杭州未来几天的天气数据（温度、天气状况、风向）。 - 记忆库：首次对话，MemoryClient().create_memory_session() 在云端创建记忆会话，为跨会话持久化做准备；get_last_k_messages(k=20) 加载 0 条历史（首轮无历史）；对话后增量同步 3 条新消息（用户输入 + AI 工具调用 + AI 最终回复）到云端记忆库。 **问题 2："查一下西湖和灵隐寺的位置"** - 用户意图：获取两个景点的地理坐标和距离 - 调用组件：网关、记忆库 - 组件作用： - MCP Gateway：LLM 连续调用 3 次工具------maps_geo（西湖坐标）、maps_geo（灵隐寺坐标）、maps_distance（两点驾车距离 4.4 公里）。LLM 自主决策调用顺序和参数，Gateway 透传执行。 - 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 3 条历史消息（Turn 1 的对话），LLM 能记住"之前查了杭州天气"的语境；对话后增量同步 4 条新消息到云端记忆库。 **问题 3："这两个地方之间怎么走？多远？"** - 用户意图：获取多种出行方式的路线规划 - 调用组件：网关、记忆库 - 组件作用： - MCP Gateway：LLM 一次性调用 4 个路径规划工具------maps_direction_driving（驾车）、maps_direction_walking（步行）、maps_direction_bicycling（骑行）、maps_direction_transit_integrated（公交）。Gateway 并行执行，返回 4 种出行方式的路线数据。安全截断机制检查每个结果长度，超过 8000 字符时会在 JSON 结构边界处截断，避免破坏数据完整性，防止云端记忆写入失败。 - 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 7 条历史消息（Turn 1-2 的对话），LLM 记住前两轮的景点坐标和天气数据，回复中引用"结合周末雨天情况"给出出行建议；对话后增量同步 3 条新消息到云端记忆库。 **问题 4："帮我算一下两天行程的总预算"** - 用户意图：精确计算两日行程的费用明细 - 调用组件：代码解释器、记忆库 - 组件作用： - 代码沙箱：LLM 不调用 MCP 工具，直接调用 execute_python，在沙箱中执行 Python 代码------定义住宿（经济型 300 元/晚）、餐饮、交通等变量，用代码做加法计算总费用，返回精确结果。LLM 基于沙箱返回的数值生成预算明细表格。 - 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 10 条历史消息（Turn 1-3 的对话），LLM 记住前 3 轮的景点和路线信息，在预算计算中引用之前查询的结果；对话后增量同步 3 条新消息到云端记忆库。 **问题 5："如果改成三天呢？住宿多一晚"** - 用户意图：基于已有两日预算做增量调整 - 调用组件：代码解释器、记忆库 - 组件作用： - 代码沙箱：LLM 生成新的 Python 代码，复用 Turn 4 的变量定义，增加第三天的费用项（+1 晚住宿 300 元 +1 天餐饮 280 元 +1 天交通 40 元 + 额外景点 80 元），沙箱执行增量计算，返回 +700 元的差额。 - 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 13 条历史消息（Turn 1-4 的对话），LLM 记住 Turn 4 的两天预算方案，能做增量调整而非重新计算；对话后增量同步 3 条新消息到云端记忆库，保存完整 5 轮对话上下文，下次开新会话能加载历史继续规划。 #### 步骤四：SDK封装与本地接口验证 将智能体封装为符合AgentArts平台规范的HTTP服务，并在本地模拟云端接口调用，确认封装正确后再执行后续的部署上云操作。 1. SDK封装，AgentArts平台通过标准HTTP协议与智能体通信，您需要用AgentArts SDK将Agent逻辑包一层，将其暴露为符合平台规范的/invocations接口，平台才能正确调度和管理您的智能体。 创建app.py文件，并上传至服务器中。该文件会将本地的Agent逻辑封装为符合AgentArts平台规范的Web服务。 ``` """ AgentArts 平台 HTTP 封装层 将 TravelAgent 暴露为符合平台规范的 /invocations 接口 """ import socket import uuid from typing import Dict, Any # ========================================== # DNS 补丁：容器内无法解析 AgentArts 公网域名 # 必须在 import agent / sdk 之前执行 # 通过 nslookup <域名> 获取对应 IP，替换下面的 XXX.XXX.XXX.XXX # ========================================== _DNS_OVERRIDES = { "memory.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX", "gateway-gaodemcp-defaultgw-ge3wkzbthf.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX", "defaultgw-ge3wkzbthf.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX", } _orig_getaddrinfo = socket.getaddrinfo def _patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): if host and host in _DNS_OVERRIDES: return _orig_getaddrinfo(_DNS_OVERRIDES[host], port, family, type, proto, flags) return _orig_getaddrinfo(host, port, family, type, proto, flags) socket.getaddrinfo = _patched_getaddrinfo from agentarts.sdk import AgentArtsRuntimeApp, RequestContext from agent import TravelAgent app = AgentArtsRuntimeApp() try: my_agent = TravelAgent() except Exception as e: print(f"Agent 初始化失败: {e}") my_agent = None def _resolve_session_id(context: RequestContext = None, payload: Dict[str, Any] = None) -> str: raw = None if context: raw = getattr(context, "session_id", None) if not raw and payload: raw = payload.get("session_id") if not raw: raw = str(uuid.uuid4()) return str(raw) @app.entrypoint def invoke(payload: Dict[str, Any], context: RequestContext = None) -> Dict[str, Any]: message = payload.get("message", "") session_id = _resolve_session_id(context, payload) if my_agent is None: return {"response": "Agent 初始化失败", "status": "error", "session_id": session_id} try: response = my_agent.run(message, session_id=session_id) return {"response": response, "status": "success", "session_id": session_id} except Exception as e: return {"response": f"处理失败: {str(e)}", "status": "error", "session_id": session_id} if __name__ == "__main__": app.run(host="0.0.0.0", port=8080) ``` 2. 接口验证，在正式推送镜像之前，先在本地模拟云端HTTP调用环境，验证Agent的接口封装是否正确、通信是否正常，可以大幅降低因接口问题导致云端部署后才发现错误的调试成本。 执行python app.py启动http server，执行以下命令调用验证Agent的HTTP接口是否已经被正确封装且能正常通信。 执行python app.py回显效果如下。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689156478.png "点击放大") 打开一个新的终端窗口（保持原窗口运行），使用curl命令进行测试。测试完成后，可以使用Ctrl + C停止运行的进程。 ``` curl --location --request POST 'http://localhost:8080/invocations' \ --header 'Content-Type: application/json' \ --data-raw '{"message": "你好，请用一句话做个自我介绍，并查询下杭州的天气。"}' ``` ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718996429.png "点击放大") #### 步骤五：云端部署 将本地智能体打包为Docker镜像并一键部署至AgentArts云端运行时。 1. 部署智能体运行时。 按上述步骤完成后，基本代码开发已经完成，接下来准备将本地创建好的智能体部署托管到AgentArts平台。 首先准备依赖文件requirements.txt，内容可参考如下： ``` # ================================================================================ # AgentArts LangGraph Agent Demo - 依赖清单 # ================================================================================ # # 本项目基于 AgentArts 平台，使用 LangGraph 框架构建 AI Agent # # 安装方式: # pip install -r requirements.txt # # ================================================================================ # 指定国内 PyPI 镜像源 --index-url https://repo.huaweicloud.com/repository/pypi/simple --trusted-host repo.huaweicloud.com # ================================================================================ # LangGraph & LangChain 核心框架 # ================================================================================ langgraph>=0.2.0 langchain>=0.3.0 langchain-core>=0.3.0 langchain-openai>=0.1.0 # ================================================================================ # LLM Provider（ModelArts MaaS OpenAI 兼容接口） # ================================================================================ openai>=1.0.0 # ================================================================================ # HTTP & 网络 # ================================================================================ requests>=2.31.0 urllib3>=2.0.0 # ================================================================================ # 工具与配置 # ================================================================================ python-dotenv>=1.0.0 pydantic>=2.0.0 # ================================================================================ # AgentArts 平台 SDK # ================================================================================ agentarts-sdk>=0.1.0 ``` 2. 执行如下命令配置智能体。 ``` agentarts configure --entrypoint app:app ``` 执行后按照操作指引进行配置。 配置智能体名称（以小写字母开头，以小写字母或数字结尾，可以包含小写字母、数字和中划线）、服务部署区域（使用cn-southwest-2，仅支持此区域）、requirements.txt依赖文件、SWR Organization镜像组织名称（如果使用自定义的镜像组织名，需要在SWR服务控制台贵阳一region创建）。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718997907.png "点击放大") 3. 执行命令部署智能体。 ``` agentarts launch ``` 该命令会自动完成以下步骤： 1. 本地构建Docker镜像。 2. 将Docker镜像推送到华为云SWR镜像仓库。 3. 部署到AgentArts运行时托管环境。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689318368.png "点击放大") 4. 调用云端Agent进行会话。 ``` agentarts invoke --agent <你的agent名称> -s "<会话ID>" '{"message": "查一下西湖和灵隐寺的位置"}' ``` ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002706375024.png "点击放大") #### 常见问题 - **执行agentarts launch命令时 出现AK/SK认证报错** 执行以下命令配置华为云凭证，获取华为云凭证请参考[认证鉴权](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_047.html)。 ``` export HUAWEICLOUD_SDK_AK="your-access-key" export HUAWEICLOUD_SDK_SK="your-secret-key" ``` ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321124.png "点击放大") - **执行agentarts launch命令** **出现运行时Runtime名称错误** 智能体的名称需要以小写字母开头，以小写字母或数字结尾，可以包含小写字母、数字和中划线，长度为2-48个字符。 请重新执行agentarts configure --entrypoint app:app命令进行配置。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002718880705.png "点击放大") - **执行命令，出现SWR.4040010,error_msg:Repository does not exists,ERROR: Organization 'xxx' not found报错。** 问题现象及原因 SWR服务镜像组织不存在，出现该报错的原因通常是在执行agentarts configure --entrypoint app:app命令时，手动指定了一个SWR镜像组织名称，但是未在SWR控制台创建同名组织导致。 解决方案 登录[SWR服务控制台](https://console.huaweicloud.com/swr/?locale=zh-cn#/dashboard)（贵阳一区域），在"组织管理"中手动创建一个与agentarts configure --entrypoint app:app命令中同名的镜像组织。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689161252.png "点击放大") - **执行python main.py或者python app.py命令出现Memory记忆库报错。** 请检查记忆库ID以及记忆库API Key配置是否正确，建议重新配置该值。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002719000831.png "点击放大") - **执行agentarts invoke与智能体进行会话出现invoke_agent failed、Failed to create sandbox且平台运行时日志出现python: exec format error报错。** 问题现象及原因 当前AgentArts服务部署于贵阳一区域，在创建智能体镜像时需要使用ARM64机器打镜像，使用X86机器打镜像会出现此报错。 ![](https://support.huaweicloud.com/highcode-agentarts/zh-cn_image_0000002689321126.png "点击放大") 解决方案 请更换为ARM64机器制作智能体镜像。 - **ECS DNS 解析失败** 问题现象 在ECS上运行python main.py本地测试时，AgentArts公网域名无法解析，报错NameResolutionError: Failed to resolve。涉及的域名包括（具体域名因region和实例不同而异，请在 AgentArts 控制台查看）： - memory.\.huaweicloud-agentarts.com（记忆库服务） - gateway-\<网关名称\>-\<网关实例ID\>.\.huaweicloud-agentarts.com（网关） - defaultgw-\<网关实例ID\>.\.huaweicloud-agentarts.com（沙箱代码解释器） 根因分析 AgentArts域名是**公网域名**，但ECS默认只配置了华为云内网DNS（100.125.x.x）。内网DNS对这些公网域名返回NXDOMAIN（域名不存在）。此外，ECS使用systemd-resolved管理 DNS，eth0网卡的 +DefaultRoute标志会用 DHCP 分配的内网DNS覆盖全局DNS配置，导致即使修改了/etc/systemd/resolved.conf也无法生效。 解决方案 绕过systemd-resolved，直接写静态/etc/resolv.conf： ``` # 删除 systemd-resolved 的符号链接 sudo rm /etc/resolv.conf # 写入静态 DNS 配置（单行命令，不用 heredoc） echo -e "nameserver 114.114.114.114\nnameserver 8.8.8.8\nnameserver 100.125.1.250\nsearch openstacklocal" | sudo tee /etc/resolv.conf ``` DNS 服务器选择说明： - 114.114.114.114（中国公共 DNS）作为首选------国内访问延迟低，解析速度快 - 8.8.8.8（Google 公共 DNS）作为备选------但国内访问有间歇性超时，不能作为首选 - 100.125.1.250（华为云内网 DNS）------用于解析华为云内部域名 验证方式： 修改后用 nslookup（不带 DNS 服务器参数）逐个测试AgentArts 域名，全部返回 IP 地址即为成功。 #### 更多业务场景 #### 场景配置说明 本智能体采用解耦架构设计：工具动态发现、配置通过环境变量注入、System Prompt 独立定义。移植至其他业务场景时，仅需调整少量配置项，核心代码无需改动。 | 修改类别 | 具体内容 | |:---|:---| | 必须修改 | System Prompt --- 定义智能体角色、能力描述与工作原则。当前为出行规划场景定制，移植时需根据新场景重写。 | | 必须修改 | .env 中的 Gateway 配置 --- 若新场景接入不同的 MCP Gateway（如数据库查询工具集），需更新 GATEWAY_ENDPOINT 与 GATEWAY_INBOUND_TOKEN。工具采用动态发现机制，业务代码无需修改。 | | 建议修改 | main.py 引导文本 --- print_guide() 中的场景描述与建议对话流程。属界面展示层，不影响核心逻辑。 | | 建议修改 | .env 模型配置 --- 若新场景对推理能力有更高要求，可调整 MODEL_NAME（如 deepseek-v3）。代码层自动适配。 | | 可选修改 | 追加本地工具 --- execute_python 具备通用性，建议保留。如需扩展其他本地工具（如发邮件、文件操作），定义后追加 StructuredTool 即可。 | #### 典型业务场景 - 客服助手 重写System Prompt为客服话术规范 → MCP Gateway替换为工单/订单查询工具 → 保留execute_python用于退款金额计算 → 其余配置无需调整 - 数据分析助手 重写System Prompt为数据分析流程 → MCP Gateway替换为SQL查询/数据导出工具 → 保留execute_python用于统计分析 → 其余配置无需调整 - 代码审查助手 重写System Prompt为审查标准 → MCP Gateway替换为代码仓库/CI状态查询工具 → 保留execute_python用于复杂度/覆盖率计算 → 其余配置无需调整