1. 白皮书

第二天的白皮书探讨了基础模型使用工具的本质：Tools 是什么如何用，接着进一步提供了设计和使用 Tools 的最佳实践原则。然后重点讨论了 MCP 的基本组成以及挑战和安全风险等内容

1.1. Tools and tool calling

1.1.1. Tools 是什么

Tools 是能够完成 LLM 自身能力之外的任务的一种函数或程序。 大模型本身会生成内容来回应用户的问题，而工具则让它能够与其他系统进行交互，分为两类：

• 让模型能够知晓某些信息
• 让模型能够执行某些操作

工具可以通过访问或检索结构化和非结构化数据源来满足用户需求，也可以代表用户执行操作，通常是通过调用外部API，或者执行某些其他代码或函数。
以下为获取天气 Agent，使用了三个外部 Tools 来帮助模型进行此项工作：

1.1.2. Tools 的类型

AI 系统中，Tool 的定义就像传统软件系统中的函数，Tool 的定义声明了模型与它之间协议，至少包括：名字(name)、参数(parameters)以及解释工具的用途和使用方式的自然语言描述(description)，接下来介绍三种类型的 Tools：

1.1.2.1. Function Tools

支持 function_calling 的模型都允许开发人员定义外部函数供给模型调用，工具的定义和使用方式会添加到模型的上下文中，当模型判断需要使用该工具的时候，就会返回她想调用的 Function 名称、参数等信息。
以下示例是一个改变灯光亮度的 Pyhon 函数，函数内的描述文字会提前加载到模型的上下文中：

def set_light_values(
    brightness: int,
    color_temp: str,
    context: ToolContext) -> dict[str, int | str]:
    """该工具用于调节用户当前所在位置房间灯光的亮度和色温。
    参数说明：
        brightness：灯光亮度，取值范围为0到100。0表示关闭灯光，100表示最大亮度
        color_temp：灯具的色温，可选值为`daylight`（日光色）、`cool`（冷光色）或`warm`（暖光色）。
        context：ToolContext对象，用于获取用户的所在位置信息。
    返回值：
        一个字典，包含已设置的灯光亮度和色温信息。
    """
    user_room_id = context.state['room_id']  # 获取用户房间ID
    # 以下为模拟的房间照明控制API（非真实接口）
    room = light_system.get_room(user_room_id)  # 通过房间ID获取对应房间的照明控制实例
    response = room.set_lights(brightness, color_temp)  # 调用接口设置灯光参数
    return {"tool_response": response}  # 返回工具调用结果

1.1.2.2. 内置 tools

一些模型厂商会在模型中内置一些他们家的基本工具，这些工具是隐含在模型内部的。比如谷歌的 Gemini 就内置了 Google Search, Code Execution, URL Context) 和 Computer Use，调用 URL Context示例如下：

from google import genai
from google.genai.types import (
    Tool,
    GenerateContentConfig,
    HttpOptions,
    UrlContext #URL上下文工具
)
client = genai.Client(http_options=HttpOptions(api_version="v1"))
model_id = "gemini-2.5-flash"
# 实例化URL上下文工具，用于让模型获取指定URL的网页内容
url_context_tool = Tool(url_context=UrlContext())

# 定义第一个烤鸡食谱的URL
url1 = "https://www.foodnetwork.com/recipes/ina-garten/perfect-roast-chicken-recipe-1940592"
# 定义第二个烤鸡食谱的URL
url2 = "https://www.allrecipes.com/recipe/70679/simple-whole-roasted-chicken/"

# 调用模型生成内容：让模型对比两个URL中的食谱食材和烹饪时间
response = client.models.generate_content(
    model=model_id,  # 指定使用的模型
    contents=("对比以下两个URL中的食谱食材和烹饪时间："  # 模型任务描述（原代码字符串拼接优化，保持语义一致）
              f"{url1} 和 {url2}"),
    config=GenerateContentConfig(),  # 生成内容的配置（此处使用默认配置）
    tools=[url_context_tool],  # 启用URL上下文工具，让模型读取URL内容
    response_modalities=["TEXT"],  # 指定模型仅返回文本形式的结果
)

for each in response.candidates[0].content.parts:
    print(each.text)
print(response.candidates[0].url_context_metadata)

1.1.2.3. Agent tools

一个 Agent 也可以被作为 Tool 调用，这样可以避免将用户的会话窗口完全移交，主 Agent 仍能保持对整个会话链路的控制，并且根据需要控制子Agent 的输入输出。

from google.adk.agents import LlmAgent
from google.adk.tools import AgentTool

# 创建一个Agent Tool（tool_agent），用于处理“查询首都”任务
tool_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="capital_agent",
    description="返回任意国家或州的首府城市", 
    instruction="""如果用户提供了国家或州的名称（例如田纳西州或新南威尔士州），
    就返回该国家或州的首府城市名称。否则，告知用户无法提供帮助。"""
)

# 创建一个用户交互Agent（user_agent），作为主智能体响应用户问题
user_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="user_advice_agent",
    description="回答用户问题并提供建议",
    instruction="""使用你已有的工具来回答用户的问题""",
    tools=[AgentTool(agent=tool_agent)]  # 为主智能体绑定工具：将“首都查询 Agent”作为可用工具
)

Agent tools 分类方法

• 信息检索：允许 Agent 从各种来源获取数据，例如网络搜索、数据库或非结构化文档。
• 行动执行：允许 Agent 执行现实世界操作：发送电子邮件、发布消息、启动代码执行或控制物理设备。
• API 集成：允许 Agent 连接现有软件系统和API，融入企业工作流程，或与第三方服务交互。
• 人机协作：促进与人类用户的合作：请求澄清、寻求关键行动的批准，或移交任务由人类进行判断。

1.1.3. 最佳实践

随着工具在 AI 应用中的使用日益广泛，以及新的工具类别不断涌现，已被认可的 tool use 最佳实践也在迅速更新。尽管如此，一些具有广泛适用性的指导原则也逐渐形成。

1.1.3.1. 文档很重要

Tool 文档（name、description 和 attributes）均作为请求上下文的一部分传递给模型，因此所有这些对于帮助模型正确使用工具都很重要。

• 使用清晰的 name：工具的名称应具有明确的描述性、易于人类理解且能精确的帮助模型决定使用哪种工具。例如，create_critical_bug_in_jira_with_priority 比 update_jira 更清晰。这对于治理也很重要；如果工具调用被记录，清晰的名称将使审计日志更具信息性。
• 描述所有输入和输出参数：应清晰描述工具的所有入参，包括所需类型以及工具将如何使用这些参数。
• 简化参数列表：过长的参数列表可能会让模型产生混淆；请保持参数列表简洁，并给参数起清晰的名称。
• 明确工具描述：提供输入和输出参数、工具用途以及有效调用工具所需的其他详细信息的清晰、详细说明。避免使用缩写或技术术语；注重用简单术语进行清晰解释。
• 添加针对性示例：示例有助于解决歧义，展示如何处理棘手请求，或阐明术语上的差异。它们也可以作为一种优化和调整模型行为的方式，而无需采用像微调这样成本更高的方法。你还可以动态检索与当前任务相关的示例，以减少上下文冗余。
• 提供默认值：为关键参数提供默认值，并确保在工具文档中记录和描述这些默认值。如果文档完善，大语言模型通常能正确使用默认值。

优质示例：

def get_product_information(product_id: str) -> dict:
    """根据产品的唯一标识（产品ID），获取该产品的完整信息。
    Args：
        product_id：产品的唯一标识（即产品ID）。
    Returns：
        一个包含产品详细信息的字典。预期包含的键（key）如下：
            - 'product_name'：产品名称
            - 'brand'：产品品牌名称
            - 'description'：一段描述产品的文本
            - 'category'：产品所属类别
            - 'status'：产品当前状态（例如：'active'表示“在售”、'inactive'表示“下架”、'suspended'表示“暂停销售”）
    示例返回值：
        {
            'product_name': 'Astro Zoom 儿童运动鞋',
            'brand': 'Cymbal 运动鞋品牌',
            'description': '...',  
            'category': '儿童鞋',
            'status': 'active' 
        }
    """

劣质示例：

def fetchpd(pid):
    """
    获取产品数据
    Args：
        pid：id
    Returns：
        数据字典
    """

1.1.3.2. 描述行为，而非实现

假设每个工具都有完善的文档说明，那么模型的指令应该描述操作，而非描述具体工具。能消除工具使用说明之间可能存在的冲突很重要（这类冲突可能会让大语言模型产生困惑）。在可用工具会动态变化的情况下（如在MCP中），这一点就更为关键。

• 描述做什么，而非怎么做：解释模型需要做什么，而不是如何去做。例如，说“创建一个BUG来描述该问题”，而不是“使用 create_bug 工具”。
• 不要重复指令：不要重复或重新陈述工具指令或文档。这可能会让模型产生困惑，并在系统指令和工具实现之间造成额外的依赖关系。
• 不要规定工作流程：描述目标，让模型有自主使用工具的空间，而不是规定特定的行动顺序。
• 务必解释工具交互：如果一个工具存在可能影响另一个工具的副作用，需记录这一点。例如，fetch_web_page 工具可能会将检索到的网页存储在文件中；记录下这一点，智能体才能知道如何访问数据。

1.1.3.3. 发布任务，而非 API 调用

工具应封装 Agent 需执行的具体任务，而非单纯包裹外部 API:
API 面向人类开发者设计，使用者具备数据和参数的完整认知，且复杂企业级 API 可能包含数十甚至上百个影响输出的参数；而智能体工具需支持动态调用 —— 智能体需在运行时自主判断使用哪些参数、传递什么数据，二者的使用场景和用户（人类 vs 智能体）能力完全不同。

1.1.3.4. 尽可能精细化

保持函数简洁且功能单一，是标准的最佳编码实践；在定义工具时也应遵循这一原则。这会让工具的文档编写更轻松，并能使 Agent 在判断何时需要使用该工具时更加明确一致性。

• 明确职责：确保每个工具都有清晰且完备的用途描述。它能做什么？应该在何时调用？它有任何副作用吗？会返回什么数据？
• 不要创建多功能工具：一般来说，不要创建需要依次执行多个步骤或包含冗长工作流程的工具。这类工具的文档编写和维护会很复杂，而且大语言模型也难以稳定地使用它们。（不过，在某些场景下，这类工具可能会有用——例如，如果某个经常执行的工作流程需要按顺序进行多次工具调用，那么定义一个单一工具来包含多个操作可能会更高效。在这种情况下，务必清晰地记录该工具的功能，以便大语言模型能有效地使用它。）

1.1.3.5. 设计简洁的输出

设计不佳的工具有时会返回大量数据，这可能会对性能和成本产生不利影响(上下文膨胀)。

• 不要返回冗长的回应：大型数据表或字典、下载的文件、生成的图像等都可能迅速占用大语言模型的输出上下文。这些回应还经常会被存储在智能体的对话历史中，因此冗长的回应也会影响后续的请求。
• 使用外部系统：利用外部系统进行数据存储和访问。例如，与其将大型查询结果直接返回给大模型，不如将其插入临时数据库表并返回表名，这样后续工具就可以直接检索数据。一些人工智能框架本身也提供持久化的外部存储，例如: Artifact Service in Google ADK（PS:谷歌是真爱给自己打广告）

1.1.3.6. 有效使用验证

Agent 框架可能具备对 Tools 的输入和输入进行 schema 验证的能力，充分使用框架的这种能力，即框架代码验证＋模型决策相辅相成。

1.1.3.7. 提供描述性错误消息

工具返回的报错信息是改进和编写工具文档描述时一个被忽视的重点。

• 通常文档完善的工具也只会返回一个错误代码，或者充其量是一条简短、缺乏描述性的错误信息。
• 但是工具回复提供给 LLM 时有助于它进行下一步操作。工具的错误消息还应向大语言模型提供一些关于如何解决特定错误的指示。（例如：一个检索产品数据的工具可以返回：“未找到产品ID为XXX的产品数据。请让客户确认产品名称，并通过名称查找产品ID以确认你拥有正确的ID。”）

1.2. 理解模型上下文协议 MCP

1.2.1. N×M 集成问题与标准化需求

工具是 Agent 或 LLM 与外部世界之间的重要纽带。随着可外部访问的工具、数据源及其他集成的生态系统正变得日益碎片化和复杂。将 LLM 与外部工具集成通常需要为每一对工具和应用程序构建一个定制的一次性连接器。这导致开发工作量激增，这一问题通常被称为 N×M 集成问题，即随着生态系统中每增加一个新模型（N）或工具（M），所需的定制连接数量会呈指数级增长。
Anthropic 于2024年11月推出了 MCP 协议，将其作为一种开放标准来着手解决这一问题。MCP 的目标是用一套统一的 “即插即用” 协议，替代当前分散的自定义集成模式。该协议可作为 AI Agent 与海量外部工具、数据之间的通用接口。通过对 “Agent 与 Tools 的通信层” 进行标准化，MCP 致力于实现 “AI Agent” 与 “Tools 具体实现细节” 的解耦，最终构建一个更具模块化、可扩展性及效率的 AI 工具生态系统。

1.2.2. 核心架构组件：Host、Clients和Servers

MCP 采用了 client - server 模型，其灵感源自软件开发领域的语言服务协议（LSP）。这一架构将 Agent 与 Tools 分离开来，为工具开发提供了更具模块化和可扩展性的方法。MCP 的核心组件包括主机（你开发的 Agent 应用）、客户端和服务端。

• MCP Host：承载 AI 应用的执行环境，既可以是独立应用（如 AI 聊天助手、智能 IDE），也可以是大型系统（如多 Agent 平台）的子组件，是连接用户、AI 模型与外部工具的核心桥梁。
• MCP Client：嵌入在 Host 内部的轻量级软件组件，每个 Client 与一个特定的 MCP Server 建立一对一连接，是两者之间数据传输与协议交互的唯一通道。
• MCP Server：提供特定功能的程序，通常作为外部工具、数据源或 API 的代理，将外部资源的能力标准化为 AI 应用可调用的接口，可部署为本地进程或远程服务。

下图展示了这些组件之间的关系以及它们如何通信：

1.2.3. 通信层：JSON-RPC、Transports和消息类型

MCP clients 与 servers 之间的所有通信都建立在标准化的技术基础之上，以确保一致性和互操作性。
1. 基础协议：MCP采用 JSON-RPC 2.0 作为其基础消息格式。这为所有通信提供了一种轻量级、基于文本且与语言无关的结构。
2. 消息类型：该协议定义了四种基本消息类型，用于规范交互流程：

• Requests：一方发送给另一方的 RPC 调用，期望得到响应。
• Results：包含相应请求成功结果的消息。
• Errors：指示请求失败的消息，包含 Code 码和错误描述。
• Notifications：一种单向消息，不需要响应，也无法回复。

3. 传输机制：MCP 还需要一个用于客户端和服务器之间通信的标准协议，称为“传输协议”，以确保每个组件都能够解析对方的消息。MCP 支持两种传输协议：一种用于本地通信，另一种用于远程连接

• stdio（标准输入/输出）：在工具需要访问用户文件系统等本地资源时使用，MCP server 作为 Agent 的子进程启用。
• Streamable HTTP：远程 client - server 协议。它支持SSE流式响应，同时也允许无状态服务并在HTTP服务中实现。

1.2.4. 核心原语：Tools 及其他

在基本通信框架之上，MCP 定义了几个关键概念或实体类型，以增强基于LLM的应用程序与外部系统交互的能力。
服务端：Tools、Resources和Prompts
客户端：Sampling、Elicitation和Roots
在 MCP 规范所定义的这些功能中，只有 Tools 得到了广泛支持，其他的所用甚少，一带而过。

1.2.4.1. Tools

MCP 的 Tool 是服务端向客户端描述其可用功能的标准化方式。例如read_file、get_weather、execute_sql 或 create_ticket。MCP服务器会发布其可用工具列表，其中包含name、descriptions 和 parameter 结构，供 Agent 发现并调用。

1.2.4.1.1. Tool 定义

工具定义必须符合JSON 结构，包含以下字段：

• name：工具的唯一标识符
• title(可选) ：用于显示的人类可读名称
• description：人类&模型可理解的功能说明
• inputSchema：JSON结构的入参说明
• outputSchema(可选)：JSON结构的出参说明
• annotations(可选)：描述工具行为的属性

name、description 和 inputSchema 为核心参数，决定了 Agent 发现并调用 Tool 的时机和效果。一个包含所有字段的完整示例如下：

{
  "name": "get_stock_price",
  "title": "股票价格查询工具",
  "description": "根据特定股票代码查询股票价格。若提供“date”（日期）参数，则返回该日期的最新价格或收盘价；若未提供，则返回最新股价。",
  "inputSchema": {
    "type": "object",
    "properties": {
      "symbol": {
        "type": "string",
        "description": "股票代码"
      },
      "date": {
        "type": "string",
        "description": "查询日期（格式为YYYY-MM-DD）"
      }
    },
    "required": ["symbol"]
  },
  "outputSchema": {
    "type": "object",
    "properties": {
      "price": {
        "type": "number",
        "description": "股票价格"
      },
      "date": {
        "type": "string",
        "description": "股票价格对应日期"
      }
    },
    "required": ["price", "date"]
  },
  "annotations": {
    "readOnlyHint": "true"
  }
}

1.2.4.1.2. Tool 结果

MCP tools 可以返回结构化或非结构化结果，且可以包含多种不同的内容类型。结果可以链接到服务的其他资源，可以返回流式回复或者非流式回复。

1.2.4.1.2.1 非结构化内容

• 文本类型代表非结构化字符串数据
• 音频和图像内容类型包含带有适当MIME类型标记的base64编码数据
• MCP 还允许工具返回指定的资源，资源既可以作为指向存储在其他 URI 的资源实体的链接返回，包括标题、描述、大小和 MIME 类型；也可以完全嵌入到工具结果中。无论哪种情况，客户端开发者应使用来自可信来源的资源。

1.2.4.1.2.2 结构化内容

• 结构化内容需始终以 JSON 对象形式返回；
• 工具实现者应借助 outputSchema 参数提供 JSON schema，供客户端验证工具结果，客户端开发者需依据该 schema 验证结果；
• 与标准函数调用类似，已定义的输出 schema 有双重作用：一是帮助客户端有效解析输出，二是告知调用的 LLM 如何及为何使用该工具。

1.2.4.1.3. 错误处理

MCP 定义两种标准错误上报机制：

• Server 针对协议问题（如工具未知、参数无效、服务器错误）返回标准 JSON-RPC 错误
• 工具运行中出现问题（如后端 API 故障、数据无效、业务逻辑错误）时，在结果对象中设 “isError”: true 参数返回错误信息。

错误信息是向调用的 LLM 提供额外上下文的重要渠道（常被忽视），MCP 工具开发者需思考如何利用该渠道助力 Agent 从错误中恢复。以下示例展示开发者如何用这两类错误为客户端 LLM 提供额外指引：
协议错误示例：

{
  "jsonrpc": "2.0",
  "id": 3,
  "error": {
    "code": -32602,
    "message": "Unknown tool: invalid_tool_name. 该工具名可能存在拼写错误，或该工具在当前服务器中不存在。请检查工具名，必要时请求获取最新的工具列表。"
  }
}

工具错误示例:

{
  "jsonrpc": "2.0",
  "id": 4,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "获取天气数据失败：API 调用次数限制已超出。请等待 15 秒后再调用此工具。"
      }
    ],
    "isError": true
  }
}

1.2.4.2. 其他五个功能

除了 Tools 之外，MCP 规范还定义了服务器和客户端可以提供的其他五种功能。不过，正如我们上面所指出的，只有少数 MCP 实现支持这些功能，因此它们是否会在基于 MCP 的部署中发挥重要作用还有待观察(了解一下就可以)。

1.2.4.2.1. Resources

资源是 MCP 服务器端提供的能力，核心作用是向 Agent 应用程序提供可访问的上下文数据，包括文件内容、数据库记录、数据库模式、图像等静态数据（如日志文件、PDF、市场统计信息等）。由于将外部内容引入 LLM 上下文存在显著安全风险，规范要求 LLM 客户端使用的资源必须经过验证，且需从可信 URL 获取，以保障数据安全。

1.2.4.2.2. Prompts

提示词是 MCP 另一项服务器端能力，主要功能是提供与工具、资源相关的可重用提示示例或模板。客户端可检索这些提示词，直接用于与 LLM 交互，服务器通过提示词能向客户端传递工具使用的高阶说明，辅助客户端更准确调用工具。但在分布式企业环境中，提示词存在安全隐患（如第三方服务注入任意指令），规范建议在更完善的安全模型落地前，尽量少用或不用该能力。

1.2.4.2.3. Sampling

采样是 MCP 客户端能力，允许服务器向客户端请求 LLM 补全结果。当服务器功能需 LLM 输入时，无需自行调用 LLM，而是通过采样请求让客户端执行 LLM 调用，反转了传统控制流程，可借助主机核心 AI 模型完成子任务（如总结服务器获取的大型文档）。该能力让客户端开发者掌控 LLM 提供商选择、成本承担及内容防护措施，还支持插入人工审批环节，但也存在提示词注入风险，需过滤验证请求提示词并确保人工介入有效。

1.2.4.2.4. Elicitation

引导是与采样类似的客户端能力，核心是允许服务器向客户端请求额外用户信息。使用引导功能的 MCP 工具不调用 LLM，而是动态查询主机应用程序获取补充数据以完成工具请求，为服务器提供 “暂停操作 - 经客户端 UI 与用户交互” 的正式机制，既让客户端保持对用户交互和数据共享的控制，也帮服务器获取用户输入。规范明确服务器不得用其请求敏感信息，需告知用户信息用途并允许用户自主选择（批准 / 拒绝 / 取消），但该限制缺乏系统强制力，客户端需做好防护以避免敏感信息泄露。

1.2.4.2.5. Roots

根目录是 MCP 第三种客户端能力，作用是定义服务器在文件系统中的可操作边界，其定义包含标识根目录的 URI（当前规范仅允许 file: URI，未来或调整）。接收根目录规范的服务器应将操作范围限定在该边界内，但目前生产环境中根目录的应用情况尚不明确 —— 规范未对服务器相关行为设置防护措施，仅建议服务器 “应尊重根目录边界”，客户端开发者不宜过度依赖服务器在根目录使用上的合规性。

1.3. MCP 支持与反对

MCP 为 Agent 开发者的工具箱增添了多项重要的新功能。但它也存在一些重要的局限性和缺点，尤其是当其使用场景从本地部署的开发者场景扩展到远程部署的企业集成应用时。在本节中，我们将探讨 MCP 的优势和功能，然后分析 MCP 带来的隐患、不足、挑战和风险。

1.3.1. 能力与战略优势

• 加速开发并培育可复用的生态系统
  • 通用协议，简化集成流程
  • 即插即用，可复用形成MCP注册中心和市场
  • MCP 项目近期启动 MCP Registry，该注册中心既是公共 MCP 服务器的核心权威来源，也提供 OpenAPI 规范以标准化 MCP 服务器声明
• 动态增强 Agent 的能力与自主性
  • 动态工具发现
  • 工具描述标准化和结构化
  • 极大扩展了LLM能力和可获取的信息
• 架构灵活性与未来适应性
  • MCP 将 Agent 架构和能力分离开，促进了模块化和可组合性
  • 可以使 Agent 系统易于调试、升级和重构
• 治理与控制的基础
  • 为后续实施更完善的治理提供了基础
  • 协议规范本身通过明确建议用户同意和控制权，为安全的 Agent 设计奠定了基础

1.3.2. 关键风险与挑战

• 性能和可扩展性瓶颈
  • 上下文窗口膨胀：MCP Tools 的源数据信息需要包含在窗口中(RAG是一个解决方案)
  • 影响LLM推理能力：工具信息太多分散模型注意力
  • 有状态的连接会导致架构复杂性
• 企业就绪差距
  • 认证与授权达不到企业级要求
  • 身份管理能力尚未达标
  • 没有原生的可观测性能力

1.4. MCP 的安全性

1.4.1. 新的威胁格局

• 基础 MCP 协议本身并不包含传统 API 端点和其他系统中已实现的许多安全功能与控制措施，通过MCP 暴露现有 API 或后端系统可能会引发新的漏洞。
• MCP 正被广泛应用于各类场景，包括许多涉及敏感个人或企业信息的场景，以及智能体与后端系统交互以执行某些现实世界操作的应用。这种广泛的适用性增加了安全问题出现的可能性及其潜在严重性，其中最突出的是未授权操作和数据泄露。

1.4.2. 风险与缓解措施

1.4.2.1. 动态能力注入

1. 风险：MCP 服务器可能动态变更提供的工具、资源或提示，未通知 / 获客户端批准，导致 AI Agent 意外获得未授权高危能力（如低风险代理获金融交易权限）。
2. 缓解措施：

• 客户端设工具 / 服务器白名单；
• 强制服务器变更工具时设 listChanged 标识并允许客户端重验证；
• 固定工具版本 / 哈希，变更即告警 / 断开；
• 用 API 网关过滤工具列表；
• 在受控环境部署 MCP 服务器。

1.4.2.2. 工具影子

1. 风险：恶意工具用宽泛触发条件（如 “save/store”）覆盖合法工具，诱使 Agent 调用，拦截 / 篡改用户敏感数据（如替换加密存储工具为恶意存储工具）。
2. 缓解措施：

• 新增工具前检查与可信工具的语义冲突；
• 高敏感连接用双向 TLS 验证身份；
• 关键交互节点强制执行安全策略；
• 高风险操作（如数据存储）需人工确认；
• 禁止 Agent 访问未授权 MCP 服务器。

1.4.2.3. 恶意工具定义与所消费内容

1. 风险：工具描述 / API 签名操纵 Agent 执行恶意操作；工具消费的外部内容含注入提示；工具返回数据泄露敏感信息。
2. 缓解措施：

• 验证用户输入，阻止恶意命令；
• 净化工具返回数据（移除令牌、隐私信息等）；
• 分离用户输入与系统提示，可设双规划器（可信 / 不可信）；
• 第三方资源需经白名单验证 + 用户明确同意；
• 净化工具描述后再注入 LLM 上下文。

1.4.2.4. 敏感信息泄露

1. 风险：工具无意 / 故意接收敏感信息，会话上下文存储并传输至未授权工具；Elicitation 能力可能被滥用请求敏感数据（无强制限制）。
2. 缓解措施：

• 工具结构化输出，用标注标识敏感字段；
• 对输入（用户自由文本、外部数据）和输出（外部邮件发送、公共数据库写入）做 “污点标记”，追踪敏感数据流向。

1.4.2.5. 不支持限制访问范围

1. 风险：MCP 仅支持粗粒度客户端 - 服务器授权，无工具 / 资源级授权；无法原生传递客户端凭证，导致权限过度或身份模糊。
2. 缓解措施：

• 工具调用用含受众（audience）和权限范围（scope）的短期凭证；
• 遵循最小权限原则（如只读工具不赋写权限）；
• 凭证 / 密钥不进入代理上下文，通过侧信道传输。

1.5. 总结

模型在孤立状态下，仅限于基于其训练数据进行模式预测。它们自身无法感知新信息，也不能对外部世界采取行动；而工具赋予了它们这些能力。

工具的有效性取决于精心的设计：清晰的文档能直接指导模型、必须体现具体的、面向用户的任务，而不仅仅是照搬复杂的内部API。

提供简洁的输出和描述性的错误信息，对于引导智能体的推理过程也必不可少。

MCP 作为开放标准，核心目标是解决 AI 工具集成的 “N×M 问题” 并构建可复用生态，其动态发现工具的能力为更自主的 AI 奠定了架构基础，但企业应用时面临显著风险 —— 因 MCP 源于去中心化、以开发者为核心的设计，目前缺乏企业级的安全、身份管理及可观测性功能，由此催生了动态能力注入、工具影子、“困惑的代理人” 漏洞等新型安全威胁。

MCP 在企业中的未来应用大概率不会是 “纯开放协议” 形态，而是需集成集中化治理与控制层。这为能强化 MCP 原生缺失的安全和身份策略的平台创造了机会，企业使用者则必须实施多层防御措施：借助 API 网关执行策略、强制使用含明确白名单的加固 SDK、遵循安全的工具设计规范。简言之，MCP 提供了工具互操作性标准，而构建其运行所需的安全、可审计、可靠框架的责任则由企业承担。

2. 代码实验室

2.1. Agent Tools

2.1.1 货币转换 Agent – 构造自定义 Function Tools

该 Agent 可以将一种面额的货币转换为另一种面额，并计算转换所需的费用，该 Agent 有两个自定义工具，并遵循以下工作流程：

1. Fee Lookup Tool - 查找转换的交易费用（模拟）
2. Exchange Rate Tool - 获取货币转换汇率（模拟）
3. Calculation Step - 计算包括费用在内的总转换成本
   

Fee Lookup Tool 代码：

	def get_fee_for_payment_method(method: str) -> dict:
	    """查询指定支付方式对应的交易手续费百分比。
	    该工具会根据用户提供的支付方式名称，模拟查询某公司内部的手续费结构。
	    Args：
	        method: 支付方式的名称，需具备描述性，
	                例如："platinum credit card"（白金信用卡）或 "bank transfer"（银行转账）。
	    Returns：
	        包含状态（status）和手续费信息的字典（dict）。
	        成功场景：{"status": "success", "fee_percentage": 0.02}
	        错误场景：{"status": "error", "error_message": "Payment method not found"}
	    """
	    # 此部分代码用于模拟查询公司内部的手续费结构
	    fee_database = {
	        "platinum credit card": 0.02,  # 对应手续费：2%（白金信用卡）
	        "gold debit card": 0.035,      # 对应手续费：3.5%（黄金借记卡）
	        "bank transfer": 0.01,         # 对应手续费：1%（银行转账）
	    }
	
	    # 将输入的支付方式名称转为小写后，在手续费数据库中匹配查询
	    fee = fee_database.get(method.lower())
	    if fee is not None:  # 若查询到对应支付方式的手续费（即支付方式存在）
	        return {"status": "success", "fee_percentage": fee}
	    else:  # 若未查询到对应支付方式（即支付方式不存在）
	        return {
	            "status": "error",
	            "error_message": f"Payment method '{method}' not found",  # 错误提示：指定支付方式未找到
	        }
	# 打印函数创建成功的提示信息
	print("✅ 手续费查询函数创建完成")
	# 测试：查询“白金信用卡”的手续费，并打印测试结果
	print(f"💳 测试结果：{get_fee_for_payment_method('platinum credit card')}")

Exchange Rate Tool 代码：

def get_exchange_rate(base_currency: str, target_currency: str) -> dict:
    """查询并返回两种货币之间的汇率。

    Args：
        base_currency: 你要转换的“基础货币”对应的ISO 4217货币代码（例如："USD"，代表美元）。
        target_currency: 你要转换到的“目标货币”对应的ISO 4217货币代码（例如："EUR"，代表欧元）。

    Returns：
        包含状态（status）和汇率信息（rate）的字典（dict）。
        成功场景（Success）：{"status": "success", "rate": 0.93}
        错误场景（Error）：{"status": "error", "error_message": "Unsupported currency pair"}
    """
    # 模拟实时汇率API的静态数据
    # 生产环境中，此处应替换为真实API请求，例如：requests.get("api.exchangerates.com")
    rate_database = {
        "usd": {  # 基础货币：美元（USD）
            "eur": 0.93,  # 目标货币：欧元（EUR），1美元=0.93欧元
            "jpy": 157.50,  # 目标货币：日元（JPY），1美元=157.50日元
            "inr": 83.58,  # 目标货币：印度卢比（INR），1美元=83.58印度卢比
        }
    }

    # 输入验证与预处理：将货币代码转为小写，避免大小写敏感导致的查询失败
    base = base_currency.lower()
    target = target_currency.lower()

    # 返回结构化结果（含状态标识）：先查基础货币，再查该基础货币下的目标货币汇率
    rate = rate_database.get(base, {}).get(target)
    if rate is not None:  # 若查询到有效汇率（即货币对存在）
        return {"status": "success", "rate": rate}
    else:  # 若未查询到有效汇率（即货币对不支持）
        return {
            "status": "error",
            "error_message": f"Unsupported currency pair: {base_currency}/{target_currency}",
            # 错误提示：明确告知不支持的货币对（保留原始输入的货币代码格式）
        }
        
# 打印函数创建成功的提示
print("✅ 汇率查询函数创建完成")
# 测试：查询“美元（USD）兑换欧元（EUR）”的汇率，并打印测试结果
print(f"💱 测试结果：{get_exchange_rate('USD', 'EUR')}")

货币转换 Agent 代码：

# 带有自定义函数工具的货币 Agent
currency_agent = LlmAgent(
    name="currency_agent", 
    model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
    instruction="""你是一个智能货币转换助手。

    处理货币转换请求时，请遵循以下步骤：
    1. 调用 `get_fee_for_payment_method()` 函数查询交易手续费
    2. 调用 `get_exchange_rate()` 函数获取货币转换汇率
    3. 检查每个工具返回结果中的 "status" 字段，确认是否存在错误
    4. 根据 `get_fee_for_payment_method` 和 `get_exchange_rate` 两个函数的输出，计算扣除手续费后的最终金额，并清晰列出计算明细
    5. 输出格式要求：
        首先，明确告知最终转换金额；
        然后，通过展示中间计算过程解释结果由来，解释内容必须包含：手续费百分比及其对应的原始货币金额、扣除手续费后的剩余金额，以及用于最终转换的汇率。

    若任意工具返回 "status" 为 "error"（错误状态），请向用户清晰说明具体问题。
    """,
    # 智能体可调用的工具列表：包含手续费查询函数和汇率查询函数
    tools=[get_fee_for_payment_method, get_exchange_rate],
)

# 打印智能体创建成功的提示信息
print("✅ 已创建带有自定义函数工具的货币智能体")
# 打印智能体可用的工具列表及功能说明
print("🔧 可用工具：")
print("  • get_fee_for_payment_method - 查询企业手续费结构")
print("  • get_exchange_rate - 获取当前货币汇率")

2.1.2 增强计算货币转换 Agent – 使用代码提高 Agent 的可靠性

在上一个例子的基础上，使用 Gemini 内置的代码执行器 BuiltInCodeExecutor 叫 Agent 自己生成 Python 代码来做数学运算规避 LLM 计算能力弱的问题。

calculation_agent 代码：

# 计算智能体（用于生成计算所需的Python代码）
calculation_agent = LlmAgent(
    name="CalculationAgent", 
    model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
    instruction="""你是一个专用计算器，**仅能以Python代码块作为响应**。严禁提供任何文本说明、解释或对话式回复。

    你的任务是接收计算请求，并将其转换为一段可计算出结果的Python代码（单个代码块）。
    
    **规则（RULES）：**
    1. 输出内容**必须仅为一个Python代码块**，不得包含其他内容。
    2. 不得在代码块之前或之后编写任何文本。
    3. 这段Python代码**必须能计算出结果**。
    4. 这段Python代码**必须将最终结果打印到标准输出（stdout）** 。
    5. 严禁你自行执行计算操作。你的唯一工作是生成用于执行计算的代码。

    若未遵守这些规则，将导致错误。
       """,
    # 代码执行器配置：使用内置代码执行器工具（BuiltInCodeExecutor）
    # 该配置为智能体赋予代码执行能力
    code_executor=BuiltInCodeExecutor(),
)

更新货币转换 Agent 代码，加入 calculation_agent ：

# 增强版货币智能体
enhanced_currency_agent = LlmAgent(
    name="enhanced_currency_agent",
    model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
    instruction="""你是一个智能货币转换助手，必须严格遵循以下步骤并使用可用工具。

对于任何货币转换请求，请执行以下操作：

1. 获取交易手续费：使用 get_fee_for_payment_method() 工具确定交易手续费。
2. 获取汇率：使用 get_exchange_rate() 工具获取货币转换汇率。
3. 错误检查：每次调用工具后，必须检查返回结果中的 "status" 字段。若状态为 "error"，必须停止操作，并向用户清晰说明问题所在。
4. 计算最终金额（关键步骤）：严禁你自行执行任何算术计算。必须使用 calculation_agent（计算智能体）工具生成Python代码，通过该代码计算最终转换金额。
   生成的代码需使用步骤1中的手续费信息和步骤2中的汇率信息。
5. 提供详细明细：在总结内容中，必须包含：
    * 明确最终转换金额；
    * 解释结果的计算过程，包括：
        * 手续费百分比及以原始货币计价的手续费金额；
        * 扣除手续费后的剩余金额；
        * 所使用的汇率。
    """,
    # 智能体可调用的工具列表
    tools=[
        get_fee_for_payment_method,  # 手续费查询工具（函数工具）
        get_exchange_rate,           # 汇率查询工具（函数工具）
        AgentTool(agent=calculation_agent),  # 计算工具（将另一个智能体作为工具使用）
    ],
)


print("✅ Enhanced currency agent created")  # 输出：✅ 增强版货币智能体创建完成
print("🎯 New capability: Delegates calculations to specialist agent")  # 输出：🎯 新增功能：将计算任务委托给专业智能体
print("🔧 Tool types used:")  # 输出：🔧 所用工具类型：
print("  • Function Tools (fees, rates)")  # 输出：  • 函数工具（手续费查询、汇率查询）
print("  • Agent Tool (calculation specialist)")  # 输出：  • 智能体工具（专业计算智能体）

2.2. Agent Tool Patterns and Best Practices

2.2.1 简单 MCP 服务示例

添加 MCP 服务 demo (npx方式)：

mcp_image_server = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",  # 通过 npx 运行 MCP 服务器
            args=[
                "-y",  # npx 用于自动确认安装的参数
                "@modelcontextprotocol/server-everything",  # Everything 服务器对应的 MCP 协议包
            ],
            tool_filter=["getTinyImage"],  # 工具过滤器，仅启用 getTinyImage 工具
        ),
        timeout=30,  # 连接超时时间，单位为秒
    )
)
print("✅ MCP 工具创建完成")  # 打印提示信息，确认 MCP 工具已成功创建

将 MCP 工具添加到 Agent:

# 集成 MCP 创建图像代理
image_agent = LlmAgent(
    model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config), 
    name="image_agent",  # 图像代理的名称
    instruction="使用 MCP 工具为用户需求生成图像",  # 对图像代理的指令：
    tools=[mcp_image_server],  # 配置 mcp tool
)

2.2.2 需要人工确认的长时间运行操作

一个货运 Agent ：在下达大额订单(>5)前应请求人工批准

1. 依赖到的技术：

• Long-Running Operation (LRO) – 长时间运行操作
• Human-in-the-loop – 人机交互
• 状态持久化和恢复机制

2. Google ADK 依赖于以下技术实现：

• ToolContext

  • 请求审批方法：调用 tool_context.request_confirmation ()
  • 检查审批状态：读取 tool_context.tool_confirmation

• ResumabilityConfig ：保存对话历史、保存工具调用状态和暂停位置

• adk_request_confirmation：标记暂定点

• InMemorySessionService：管理持久化会话

3. 这里我们使 LangGrap 框架复刻：

#!/usr/bin/env python3
"""
LangGraph 1.0.3 实现 Google ADK 长时间运行操作示例
使用 LangGraph 1.0.3 的新特性复刻 Google ADK 的人机交互审批流程

技术要点：
1. interrupt() - 实现暂停等待人工输入（对应 ADK 的 request_confirmation）
2. Command - 控制图的执行流程和状态更新
3. MemorySaver - 实现状态持久化和恢复（对应 ADK 的 ResumabilityConfig）
4. Human-in-the-Loop 模式 - 完整的暂停/恢复工作流
"""

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
import uuid
import re

LARGE_ORDER_THRESHOLD = 5  # 大订单阈值

# 状态定义 - 对应 ADK 的 Session State
class ShippingState(TypedDict):
    """货运代理的状态结构"""
    query: str  # 用户原始请求
    num_containers: int  # 集装箱数量
    destination: str  # 目的地
    order_status: str  # 订单状态：pending, approved, rejected, auto_approved
    order_id: str  # 订单ID
    message: str  # 返回消息


# 图节点实现 - 对应 ADK 的工具和 Agent 逻辑
def parse_user_input(state: ShippingState) -> dict:
    """
    解析用户输入，提取集装箱数量和目的地

    实际生产环境应该使用 LLM 来提取参数，这里为了演示简化处理
    对应 ADK 中 Agent 接收消息并理解用户意图的过程
    """
    text = state["query"].lower()

    # 提取数字（集装箱数量）
    numbers = re.findall(r'\d+', text)
    num_containers = int(numbers[0]) if numbers else 3

    # 提取目的地
    destination_map = {
        "singapore": "Singapore",
        "新加坡": "Singapore",
        "rotterdam": "Rotterdam",
        "鹿特丹": "Rotterdam",
        "los angeles": "Los Angeles",
        "洛杉矶": "Los Angeles",
    }

    destination = "Unknown"
    for key, value in destination_map.items():
        if key in text:
            destination = value
            break

    print(f"📋 解析结果: {num_containers} 个集装箱 → {destination}")

    return {
        "num_containers": num_containers,
        "destination": destination,
    }


def check_order_size(state: ShippingState) -> dict:
    """
    检查订单大小，决定是否需要审批
    对应 ADK 中 place_shipping_order 工具的场景 1 和场景 2

    场景 1: 小订单（≤5）→ 自动批准
    场景 2: 大订单（>5）→ 设置状态为 pending，后续节点会请求审批
    """
    num_containers = state["num_containers"]
    destination = state["destination"]

    # 场景 1: 小订单自动批准
    if num_containers <= LARGE_ORDER_THRESHOLD:
        order_id = f"ORD-{num_containers}-AUTO"
        message = f"✅ 订单自动批准: {num_containers} 个集装箱发往 {destination}，订单号: {order_id}"

        print(f"🚀 小订单，自动批准")

        return {
            "order_status": "auto_approved",
            "order_id": order_id,
            "message": message,
        }

    # 场景 2: 大订单需要审批
    else:
        message = f"⚠️ 大订单检测: {num_containers} 个集装箱发往 {destination}，需要人工审批"

        print(f"⏸️  大订单，需要审批")

        return {
            "order_status": "pending",
            "message": message,
        }


def request_approval(state: ShippingState) -> dict:
    """
    请求人工审批 - 这是关键节点！
    使用 interrupt() 暂停图执行，等待外部输入

    对应 ADK 中的：
    - tool_context.request_confirmation()
    - 创建 adk_request_confirmation 事件

    LangGraph 1.0.3 的 interrupt() 机制：
    1. interrupt() 被调用时，图执行立即暂停
    2. 函数返回值是暂停时传递的提示信息
    3. 当外部调用 graph.invoke(Command(resume=value)) 时，
       interrupt() 返回 value 值，函数继续执行
    """
    num_containers = state["num_containers"]
    destination = state["destination"]

    # 暂停执行，等待人工决策
    # interrupt() 的参数是展示给人工的提示信息
    approval_hint = {
        "type": "approval_request",
        "num_containers": num_containers,
        "destination": destination,
        "question": f"⚠️ 大订单: {num_containers} 个集装箱发往 {destination}。是否批准？",
    }

    print(f"⏸️  图执行已暂停，等待人工审批...")

    # 关键！interrupt() 会暂停执行
    # 返回值将在 graph.invoke(Command(resume=...)) 时获得
    approval_decision = interrupt(approval_hint)

    # 执行到这里时，说明已经恢复执行，approval_decision 是人工的决策
    print(f"▶️  图执行已恢复，收到审批决策: {'批准 ✅' if approval_decision else '拒绝 ❌'}")

    # 场景 3: 根据审批结果返回不同状态
    if approval_decision:
        # 批准
        order_id = f"ORD-{num_containers}-HUMAN"
        message = f"✅ 订单已批准: {num_containers} 个集装箱发往 {destination}，订单号: {order_id}"
        return {
            "order_status": "approved",
            "order_id": order_id,
            "message": message,
        }
    else:
        # 拒绝
        message = f"❌ 订单已拒绝: {num_containers} 个集装箱发往 {destination}"
        return {
            "order_status": "rejected",
            "message": message,
        }


def route_after_check(state: ShippingState) -> Literal["request_approval", "complete"]:
    """
    条件路由：根据订单状态决定下一步

    - 如果是 pending（大订单），路由到 request_approval 节点
    - 如果是 auto_approved（小订单），直接路由到 complete 节点

    对应 ADK 中检测 adk_request_confirmation 事件的逻辑
    """
    if state["order_status"] == "pending":
        return "request_approval"
    return "complete"


def complete(state: ShippingState) -> dict:
    """完成节点，打印最终结果"""
    print(f"✨ {state['message']}")
    return state

# 构建图 - 对应 ADK 的 App
def create_shipping_graph():
    """
    创建货运代理图

    对应 ADK 中的：
    App(
        root_agent=shipping_agent,
        resumability_config=ResumabilityConfig(is_resumable=True)
    )

    图结构：
    START → parse_input → check_order → [条件路由]
                                          ├─ auto_approved → complete → END
                                          └─ pending → request_approval → complete → END
    """
    # 创建状态图
    workflow = StateGraph(ShippingState)

    # 添加节点
    workflow.add_node("parse_input", parse_user_input)
    workflow.add_node("check_order", check_order_size)
    workflow.add_node("request_approval", request_approval)
    workflow.add_node("complete", complete)

    # 定义边
    workflow.add_edge(START, "parse_input")
    workflow.add_edge("parse_input", "check_order")

    # 条件路由：根据订单状态决定是否需要审批
    workflow.add_conditional_edges(
        "check_order",
        route_after_check,
        {
            "request_approval": "request_approval",
            "complete": "complete"
        }
    )

    # 审批完成后跳转到 complete
    workflow.add_edge("request_approval", "complete")
    workflow.add_edge("complete", END)

    # 编译图并添加 Checkpointer（实现状态持久化）
    # 这对应 ADK 的 ResumabilityConfig(is_resumable=True)
    checkpointer = MemorySaver()
    graph = workflow.compile(checkpointer=checkpointer)

    return graph


# 工作流执行函数 - 对应 ADK 的 run_shipping_workflow
def run_shipping_workflow(query: str, auto_approve: bool = True):
    """
    运行货运工作流，支持人机交互审批

    对应 ADK 中的 run_shipping_workflow 函数

    执行流程：
    1. 创建图和唯一的 thread_id（会话ID）
    2. 首次调用 graph.invoke() 执行图
    3. 检查图是否被 interrupt() 暂停
    4. 如果暂停，使用 Command(resume=decision) 恢复执行

    Args:
        query: 用户请求
        auto_approve: 是否自动批准大订单（模拟人工决策）
    """
    print(f"{'='*60}")
    print(f"User : {query}")
    print(f"\n")

    # 创建图
    graph = create_shipping_graph()

    # 创建唯一的 thread_id（对应 ADK 的 session_id）
    thread_id = f"order_{uuid.uuid4().hex[:8]}"
    config = {"configurable": {"thread_id": thread_id}}

    # 初始状态
    initial_state = {
        "query": query,
        "num_containers": 0,
        "destination": "",
        "order_status": "",
        "order_id": "",
        "message": "",
    }

    # STEP 1: 首次运行图
    # 对应 ADK 中第一次调用 shipping_runner.run_async()
    result = graph.invoke(initial_state, config)

    # STEP 2: 检查图是否被中断（等待审批）
    # 对应 ADK 中的 check_for_approval(events)
    state_snapshot = graph.get_state(config)

    # 如果 state_snapshot.next 不为空，说明图被 interrupt() 暂停了
    if state_snapshot.next:
        print(f"\n⏸️  工作流已暂停，等待人工审批")
        print(f"🤔 模拟人工决策: {'批准 ✅' if auto_approve else '拒绝 ❌'}\n")

        # STEP 3: 恢复执行，传入审批决策
        # 对应 ADK 中第二次调用 run_async() 并传入 approval response
        # 使用 Command(resume=value) 恢复执行
        # 注意：这里传入 None 作为第一个参数，因为我们不需要添加新的输入
        result = graph.invoke(Command(resume=auto_approve), config)

    return result



# 主程序 - 测试三种场景
def main():
    """
    主测试函数 - 复刻 ADK 文档中的三个 Demo

    Demo 1: 小订单（≤5）→ 自动批准
    Demo 2: 大订单（>5）+ 人工批准
    Demo 3: 大订单（>5）+ 人工拒绝
    """

    # Demo 1: 小订单（自动批准）
    run_shipping_workflow("发送 3 个集装箱到 Singapore")

    # Demo 2: 大订单（人工批准）
    run_shipping_workflow("发送 10 个集装箱到 Rotterdam", auto_approve=True)

    # Demo 3: 大订单（人工拒绝）
    run_shipping_workflow("发送 8 个集装箱到 Los Angeles", auto_approve=False)


if __name__ == "__main__":
    main()

运行结果如下：

============================================================
User : 发送 3 个集装箱到 Singapore


📋 解析结果: 3 个集装箱 → Singapore
🚀 小订单，自动批准
✨ ✅ 订单自动批准: 3 个集装箱发往 Singapore，订单号: ORD-3-AUTO
============================================================
User : 发送 10 个集装箱到 Rotterdam


📋 解析结果: 10 个集装箱 → Rotterdam
⏸️  大订单，需要审批
⏸️  图执行已暂停，等待人工审批...

⏸️  工作流已暂停，等待人工审批
🤔 模拟人工决策: 批准 ✅

⏸️  图执行已暂停，等待人工审批...
▶️  图执行已恢复，收到审批决策: 批准 ✅
✨ ✅ 订单已批准: 10 个集装箱发往 Rotterdam，订单号: ORD-10-HUMAN
============================================================
User : 发送 8 个集装箱到 Los Angeles


📋 解析结果: 8 个集装箱 → Los Angeles
⏸️  大订单，需要审批
⏸️  图执行已暂停，等待人工审批...

⏸️  工作流已暂停，等待人工审批
🤔 模拟人工决策: 拒绝 ❌

⏸️  图执行已暂停，等待人工审批...
▶️  图执行已恢复，收到审批决策: 拒绝 ❌
✨ ❌ 订单已拒绝: 8 个集装箱发往 Los Angeles

参考文献：
5-Day AI Agents Intensive Course with Google
Agent Tools & Interoperability with MCP
Day 2a - Agent Tools
Day 2b - Agent Tools Best Practices