hello-agents/Co-creation-projects/AstrumPush-Smart-Recipe-Agent/protocol_tools.py

"""
协议工具集合

提供基于协议实现的工具接口：
- MCP Tool: 基于 fastmcp 库，用于连接和调用 MCP 服务器
- A2A Tool: 基于官方 a2a 库，用于 Agent 间通信（需要安装 a2a）
- ANP Tool: 基于概念实现，用于服务发现和网络管理
"""

from typing import Dict, Any, List, Optional
from ..base import Tool, ToolParameter
import os

# todo: 修改by xc
import gc
import asyncio
import sys
if sys.platform == "win32":
    # Windows 10+: 使用 SelectorEventLoop 替代 ProactorEventLoop，
    # 可避免 GetQueuedCompletionStatus 阻塞问题
    if sys.version_info >= (3, 8):
        asyncio.set_event_loop_policy(
            asyncio.WindowsSelectorEventLoopPolicy()
        )
        

# MCP服务器环境变量映射表
# 用于自动检测常见MCP服务器需要的环境变量
MCP_SERVER_ENV_MAP = {
    "server-github": ["GITHUB_PERSONAL_ACCESS_TOKEN"],
    "server-slack": ["SLACK_BOT_TOKEN", "SLACK_TEAM_ID"],
    "server-google-drive": ["GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET", "GOOGLE_REFRESH_TOKEN"],
    "server-postgres": ["POSTGRES_CONNECTION_STRING"],
    "server-sqlite": [],  # 不需要环境变量
    "server-filesystem": [],  # 不需要环境变量
}


class MCPTool(Tool):
    """MCP (Model Context Protocol) 工具

    连接到 MCP 服务器并调用其提供的工具、资源和提示词。
    
    功能：
    - 列出服务器提供的工具
    - 调用服务器工具
    - 读取服务器资源
    - 获取提示词模板

    使用示例:
        >>> from hello_agents.tools.builtin import MCPTool
        >>>
        >>> # 方式1: 使用内置演示服务器
        >>> tool = MCPTool()  # 自动创建内置服务器
        >>> result = tool.run({"action": "list_tools"})
        >>>
        >>> # 方式2: 连接到外部 MCP 服务器
        >>> tool = MCPTool(server_command=["python", "examples/mcp_example.py"])
        >>> result = tool.run({"action": "list_tools"})
        >>>
        >>> # 方式3: 使用自定义 FastMCP 服务器
        >>> from fastmcp import FastMCP
        >>> server = FastMCP("MyServer")
        >>> tool = MCPTool(server=server)

    注意：使用 fastmcp 库，已包含在依赖中
    """
    
    def __init__(self,
                 name: str = "mcp",
                 description: Optional[str] = None,
                 server_command: Optional[List[str]] = None,
                 server_args: Optional[List[str]] = None,
                 server: Optional[Any] = None,
                 auto_expand: bool = True,
                 env: Optional[Dict[str, str]] = None,
                 env_keys: Optional[List[str]] = None):
        """
        初始化 MCP 工具

        Args:
            name: 工具名称（默认为"mcp"，建议为不同服务器指定不同名称）
            description: 工具描述（可选，默认为通用描述）
            server_command: 服务器启动命令（如 ["python", "server.py"]）
            server_args: 服务器参数列表
            server: FastMCP 服务器实例（可选，用于内存传输）
            auto_expand: 是否自动展开为独立工具（默认True）
            env: 环境变量字典（优先级最高，直接传递给MCP服务器）
            env_keys: 要从系统环境变量加载的key列表（优先级中等）

        环境变量优先级（从高到低）：
            1. 直接传递的env参数
            2. env_keys指定的环境变量
            3. 自动检测的环境变量（根据server_command）

        注意：如果所有参数都为空，将创建内置演示服务器

        示例：
            >>> # 方式1：直接传递环境变量（优先级最高）
            >>> github_tool = MCPTool(
            ...     name="github",
            ...     server_command=["npx", "-y", "@modelcontextprotocol/server-github"],
            ...     env={"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxx"}
            ... )
            >>>
            >>> # 方式2：从.env文件加载指定的环境变量
            >>> github_tool = MCPTool(
            ...     name="github",
            ...     server_command=["npx", "-y", "@modelcontextprotocol/server-github"],
            ...     env_keys=["GITHUB_PERSONAL_ACCESS_TOKEN"]
            ... )
            >>>
            >>> # 方式3：自动检测（最简单，推荐）
            >>> github_tool = MCPTool(
            ...     name="github",
            ...     server_command=["npx", "-y", "@modelcontextprotocol/server-github"]
            ...     # 自动从环境变量加载GITHUB_PERSONAL_ACCESS_TOKEN
            ... )
        """
        self.server_command = server_command
        self.server_args = server_args or []
        self.server = server
        self._client = None
        self._available_tools = []
        self.auto_expand = auto_expand
        self.prefix = f"{name}_" if auto_expand else ""

        # 环境变量处理（优先级：env > env_keys > 自动检测）
        self.env = self._prepare_env(env, env_keys, server_command)

        # 如果没有指定任何服务器，创建内置演示服务器
        if not server_command and not server:
            self.server = self._create_builtin_server()

        # 自动发现工具
        self._discover_tools()

        # 设置默认描述或自动生成
        if description is None:
            description = self._generate_description()

        super().__init__(
            name=name,
            description=description
        )

    def _prepare_env(self,
                     env: Optional[Dict[str, str]],
                     env_keys: Optional[List[str]],
                     server_command: Optional[List[str]]) -> Dict[str, str]:
        """
        准备环境变量

        优先级：env > env_keys > 自动检测

        Args:
            env: 直接传递的环境变量字典
            env_keys: 要从系统环境变量加载的key列表
            server_command: 服务器命令（用于自动检测）

        Returns:
            合并后的环境变量字典
        """
        result_env = {}

        # 1. 自动检测（优先级最低）
        if server_command:
            # 从命令中提取服务器名称
            server_name = None
            for part in server_command:
                if "server-" in part:
                    # 提取类似 "@modelcontextprotocol/server-github" 中的 "server-github"
                    server_name = part.split("/")[-1] if "/" in part else part
                    break

            # 查找映射表
            if server_name and server_name in MCP_SERVER_ENV_MAP:
                auto_keys = MCP_SERVER_ENV_MAP[server_name]
                for key in auto_keys:
                    value = os.getenv(key)
                    if value:
                        result_env[key] = value
                        print(f"🔑 自动加载环境变量: {key}")

        # 2. env_keys指定的环境变量（优先级中等）
        if env_keys:
            for key in env_keys:
                value = os.getenv(key)
                if value:
                    result_env[key] = value
                    print(f"🔑 从env_keys加载环境变量: {key}")
                else:
                    print(f"⚠️  警告: 环境变量 {key} 未设置")

        # 3. 直接传递的env（优先级最高）
        if env:
            result_env.update(env)
            for key in env.keys():
                print(f"🔑 使用直接传递的环境变量: {key}")

        return result_env

    def _create_builtin_server(self):
        """创建内置演示服务器"""
        try:
            from fastmcp import FastMCP

            server = FastMCP("HelloAgents-BuiltinServer")

            @server.tool()
            def add(a: float, b: float) -> float:
                """加法计算器"""
                return a + b

            @server.tool()
            def subtract(a: float, b: float) -> float:
                """减法计算器"""
                return a - b

            @server.tool()
            def multiply(a: float, b: float) -> float:
                """乘法计算器"""
                return a * b

            @server.tool()
            def divide(a: float, b: float) -> float:
                """除法计算器"""
                if b == 0:
                    raise ValueError("除数不能为零")
                return a / b

            @server.tool()
            def greet(name: str = "World") -> str:
                """友好问候"""
                return f"Hello, {name}! 欢迎使用 HelloAgents MCP 工具！"

            @server.tool()
            def get_system_info() -> dict:
                """获取系统信息"""
                import platform
                import sys
                return {
                    "platform": platform.system(),
                    "python_version": sys.version,
                    "server_name": "HelloAgents-BuiltinServer",
                    "tools_count": 6
                }

            return server

        except ImportError:
            raise ImportError(
                "创建内置 MCP 服务器需要 fastmcp 库。请安装: pip install fastmcp"
            )

    def _discover_tools(self):
        """发现MCP服务器提供的所有工具"""
        try:
            from hello_agents.protocols.mcp.client import MCPClient
            import asyncio

            async def discover():
                client_source = self.server if self.server else self.server_command
                async with MCPClient(client_source, self.server_args, env=self.env) as client:
                    tools = await client.list_tools()
                    return tools

            # 运行异步发现
            try:
                loop = asyncio.get_running_loop()
                # 如果已有循环，在新线程中运行
                import concurrent.futures
                def run_in_thread():
                    new_loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(new_loop)
                    try:
                        return new_loop.run_until_complete(discover())
                    finally:
                        new_loop.close()

                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(run_in_thread)
                    self._available_tools = future.result()
            except RuntimeError:
                # 没有运行中的循环
                self._available_tools = asyncio.run(discover())

        except Exception as e:
            # 工具发现失败不影响初始化
            self._available_tools = []

    def _generate_description(self) -> str:
        """生成增强的工具描述"""
        if not self._available_tools:
            return "连接到 MCP 服务器，调用工具、读取资源和获取提示词。支持内置服务器和外部服务器。"

        if self.auto_expand:
            # 展开模式：简单描述
            return f"MCP工具服务器，包含{len(self._available_tools)}个工具。这些工具会自动展开为独立的工具供Agent使用。"
        else:
            # 非展开模式：详细描述
            desc_parts = [
                f"MCP工具服务器，提供{len(self._available_tools)}个工具："
            ]

            # 列出所有工具
            for tool in self._available_tools:
                tool_name = tool.get('name', 'unknown')
                tool_desc = tool.get('description', '无描述')
                # 简化描述，只取第一句
                short_desc = tool_desc.split('.')[0] if tool_desc else '无描述'
                desc_parts.append(f"  • {tool_name}: {short_desc}")

            # 添加调用格式说明
            desc_parts.append("\n调用格式：返回JSON格式的参数")
            desc_parts.append('{"action": "call_tool", "tool_name": "工具名", "arguments": {...}}')

            # 添加示例
            if self._available_tools:
                first_tool = self._available_tools[0]
                tool_name = first_tool.get('name', 'example')
                desc_parts.append(f'\n示例：{{"action": "call_tool", "tool_name": "{tool_name}", "arguments": {{...}}}}')

            return "\n".join(desc_parts)

    def get_expanded_tools(self) -> List['Tool']:  # type: ignore
        """
        获取展开的工具列表

        将MCP服务器的每个工具包装成独立的Tool对象

        Returns:
            Tool对象列表
        """
        if not self.auto_expand:
            return []

        from .mcp_wrapper_tool import MCPWrappedTool

        expanded_tools = []
        for tool_info in self._available_tools:
            wrapped_tool = MCPWrappedTool(
                mcp_tool=self,
                tool_info=tool_info,
                prefix=self.prefix
            )
            expanded_tools.append(wrapped_tool)

        return expanded_tools

    def run(self, parameters: Dict[str, Any]) -> str:
        """
        执行 MCP 操作

        Args:
            parameters: 包含以下参数的字典
                - action: 操作类型 (list_tools, call_tool, list_resources, read_resource, list_prompts, get_prompt)
                  如果不指定action但指定了tool_name，会自动推断为call_tool
                - tool_name: 工具名称（call_tool 需要）
                - arguments: 工具参数（call_tool 需要）
                - uri: 资源 URI（read_resource 需要）
                - prompt_name: 提示词名称（get_prompt 需要）
                - prompt_arguments: 提示词参数（get_prompt 可选）

        Returns:
            操作结果
        """
        from hello_agents.protocols.mcp.client import MCPClient

        timeout = getattr(self, 'timeout', 10)

        # 智能推断action：如果没有action但有tool_name，自动设置为call_tool
        action = parameters.get("action", "").lower()
        if not action and "tool_name" in parameters:
            action = "call_tool"
            parameters["action"] = action

        if not action:
            return "错误：必须指定 action 参数或 tool_name 参数"
        
        try:
            # 使用增强的异步客户端
            import asyncio
            from hello_agents.protocols.mcp.client import MCPClient

            async def run_mcp_operation():
                # 根据配置选择客户端创建方式
                if self.server:
                    # 使用内置服务器（内存传输）
                    client_source = self.server
                else:
                    # 使用外部服务器命令
                    client_source = self.server_command

                async with MCPClient(client_source, self.server_args, env=self.env) as client:
                    if action == "list_tools":
                        tools = await client.list_tools()
                        if not tools:
                            return "没有找到可用的工具"
                        result = f"找到 {len(tools)} 个工具:\n"
                        for tool in tools:
                            result += f"- {tool['name']}: {tool['description']}\n"
                        return result

                    elif action == "call_tool":
                        tool_name = parameters.get("tool_name")
                        arguments = parameters.get("arguments", {})
                        if not tool_name:
                            return "错误：必须指定 tool_name 参数"
                        
                        # todo: 修改by xc
                        result = await asyncio.wait_for(client.call_tool(tool_name, arguments), timeout=timeout)
                        # result = await client.call_tool(tool_name, arguments)
                        return f"工具 '{tool_name}' 执行结果:\n{result}"

                    elif action == "list_resources":
                        resources = await client.list_resources()
                        if not resources:
                            return "没有找到可用的资源"
                        result = f"找到 {len(resources)} 个资源:\n"
                        for resource in resources:
                            result += f"- {resource['uri']}: {resource['name']}\n"
                        return result

                    elif action == "read_resource":
                        uri = parameters.get("uri")
                        if not uri:
                            return "错误：必须指定 uri 参数"
                        content = await client.read_resource(uri)
                        return f"资源 '{uri}' 内容:\n{content}"

                    elif action == "list_prompts":
                        prompts = await client.list_prompts()
                        if not prompts:
                            return "没有找到可用的提示词"
                        result = f"找到 {len(prompts)} 个提示词:\n"
                        for prompt in prompts:
                            result += f"- {prompt['name']}: {prompt['description']}\n"
                        return result

                    elif action == "get_prompt":
                        prompt_name = parameters.get("prompt_name")
                        prompt_arguments = parameters.get("prompt_arguments", {})
                        if not prompt_name:
                            return "错误：必须指定 prompt_name 参数"
                        messages = await client.get_prompt(prompt_name, prompt_arguments)
                        result = f"提示词 '{prompt_name}':\n"
                        for msg in messages:
                            result += f"[{msg['role']}] {msg['content']}\n"
                        return result

                    else:
                        return f"错误：不支持的操作 '{action}'"

            # 运行异步操作
            try:
                # 检查是否已有运行中的事件循环
                try:
                    loop = asyncio.get_running_loop()
                    # 如果有运行中的循环，在新线程中运行新的事件循环
                    import concurrent.futures
                    import threading

                    def run_in_thread():
                        # 在新线程中创建新的事件循环
                        new_loop = asyncio.new_event_loop()
                        asyncio.set_event_loop(new_loop)
                        try:
                            return new_loop.run_until_complete(run_mcp_operation())
                        finally:
                            # todo: 修改by xc
                            # 🔑 关键清理 1：取消所有残留任务，防止 transport 未关闭
                            pending = asyncio.all_tasks(new_loop)
                            for task in pending:
                                task.cancel()
                            if pending:
                                new_loop.run_until_complete(
                                    asyncio.gather(*pending, return_exceptions=True)
                                )

                            new_loop.close()

                    # todo: 修改by xc
                    executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
                    try:
                        future = executor.submit(run_in_thread)
                        # 🔑 关键清理 2：设置超时，避免永久阻塞
                        return future.result(timeout=timeout)
                    finally:
                        # 🔑 关键清理 3：不等待残留线程，直接关闭线程池
                        executor.shutdown(wait=False, cancel_futures=True)

                except RuntimeError:
                    # 没有运行中的循环，直接运行
                    return asyncio.run(run_mcp_operation())
            except Exception as e:
                return f"异步操作失败: {str(e)}"
            finally:
                # todo: 修改by xc
                # 🔑 关键清理 4：强制回收未关闭的管道/文件描述符
                gc.collect()

        except Exception as e:
            return f"MCP 操作失败: {str(e)}"
    
    def get_parameters(self) -> List[ToolParameter]:
        """获取工具参数定义"""
        return [
            ToolParameter(
                name="action",
                type="string",
                description="操作类型: list_tools, call_tool, list_resources, read_resource, list_prompts, get_prompt",
                required=True
            ),
            ToolParameter(
                name="tool_name",
                type="string",
                description="工具名称（call_tool 操作需要）",
                required=False
            ),
            ToolParameter(
                name="arguments",
                type="object",
                description="工具参数（call_tool 操作需要）",
                required=False
            ),
            ToolParameter(
                name="uri",
                type="string",
                description="资源 URI（read_resource 操作需要）",
                required=False
            ),
            ToolParameter(
                name="prompt_name",
                type="string",
                description="提示词名称（get_prompt 操作需要）",
                required=False
            ),
            ToolParameter(
                name="prompt_arguments",
                type="object",
                description="提示词参数（get_prompt 操作可选）",
                required=False
            )
        ]


class A2ATool(Tool):
    """A2A (Agent-to-Agent Protocol) 工具

    连接到 A2A Agent 并进行通信。
    
    功能：
    - 向 Agent 提问
    - 获取 Agent 信息
    - 发送自定义消息

    使用示例:
        >>> from hello_agents.tools.builtin import A2ATool
        >>> # 连接到 A2A Agent（使用默认名称）
        >>> tool = A2ATool(agent_url="http://localhost:5000")
        >>> # 连接到 A2A Agent（自定义名称和描述）
        >>> tool = A2ATool(
        ...     agent_url="http://localhost:5000",
        ...     name="tech_expert",
        ...     description="技术专家，回答技术相关问题"
        ... )
        >>> # 提问
        >>> result = tool.run({"action": "ask", "question": "计算 2+2"})
        >>> # 获取信息
        >>> result = tool.run({"action": "get_info"})
    
    注意：需要安装官方 a2a-sdk 库: pip install a2a-sdk
    详见文档: docs/chapter10/A2A_GUIDE.md
    官方仓库: https://github.com/a2aproject/a2a-python
    """
    
    def __init__(self, agent_url: str, name: str = "a2a", description: str = None):
        """
        初始化 A2A 工具

        Args:
            agent_url: Agent URL
            name: 工具名称（可选，默认为 "a2a"）
            description: 工具描述（可选）
        """
        if description is None:
            description = "连接到 A2A Agent，支持提问和获取信息。需要安装官方 a2a-sdk 库。"

        super().__init__(
            name=name,
            description=description
        )
        self.agent_url = agent_url
        
    def run(self, parameters: Dict[str, Any]) -> str:
        """
        执行 A2A 操作
        
        Args:
            parameters: 包含以下参数的字典
                - action: 操作类型 (ask, get_info)
                - question: 问题文本（ask 需要）
        
        Returns:
            操作结果
        """
        try:
            from hello_agents.protocols.a2a.implementation import A2AClient, A2A_AVAILABLE
            if not A2A_AVAILABLE:
                return ("错误：需要安装 a2a-sdk 库\n"
                       "安装命令: pip install a2a-sdk\n"
                       "详见文档: docs/chapter10/A2A_GUIDE.md\n"
                       "官方仓库: https://github.com/a2aproject/a2a-python")
        except ImportError:
            return ("错误：无法导入 A2A 模块\n"
                   "安装命令: pip install a2a-sdk\n"
                   "详见文档: docs/chapter10/A2A_GUIDE.md\n"
                   "官方仓库: https://github.com/a2aproject/a2a-python")

        action = parameters.get("action", "").lower()
        
        if not action:
            return "错误：必须指定 action 参数"
        
        try:
            client = A2AClient(self.agent_url)
            
            if action == "ask":
                question = parameters.get("question")
                if not question:
                    return "错误：必须指定 question 参数"
                response = client.ask(question)
                return f"Agent 回答:\n{response}"
                
            elif action == "get_info":
                info = client.get_info()
                result = "Agent 信息:\n"
                for key, value in info.items():
                    result += f"- {key}: {value}\n"
                return result
                
            else:
                return f"错误：不支持的操作 '{action}'"
                
        except Exception as e:
            return f"A2A 操作失败: {str(e)}"
    
    def get_parameters(self) -> List[ToolParameter]:
        """获取工具参数定义"""
        return [
            ToolParameter(
                name="action",
                type="string",
                description="操作类型: ask(提问), get_info(获取信息)",
                required=True
            ),
            ToolParameter(
                name="question",
                type="string",
                description="问题文本（ask 操作需要）",
                required=False
            )
        ]


class ANPTool(Tool):
    """ANP (Agent Network Protocol) 工具

    提供智能体网络管理功能，包括服务发现、节点管理和消息路由。
    这是一个概念性实现，用于演示 Agent 网络管理的核心理念。
    
    功能：
    - 注册和发现服务
    - 添加和管理网络节点
    - 消息路由
    - 网络统计

    使用示例:
        >>> from hello_agents.tools.builtin import ANPTool
        >>> tool = ANPTool()
        >>> # 注册服务
        >>> result = tool.run({
        ...     "action": "register_service",
        ...     "service_id": "calc-1",
        ...     "service_type": "calculator",
        ...     "endpoint": "http://localhost:5001"
        ... })
        >>> # 发现服务
        >>> result = tool.run({
        ...     "action": "discover_services",
        ...     "service_type": "calculator"
        ... })
        >>> # 添加节点
        >>> result = tool.run({
        ...     "action": "add_node",
        ...     "node_id": "agent-1",
        ...     "endpoint": "http://localhost:5001"
        ... })
    
    注意：这是概念性实现，不需要额外依赖
    详见文档: docs/chapter10/ANP_CONCEPTS.md
    """
    
    def __init__(self, name: str = "anp", description: str = None, discovery=None, network=None):
        """初始化 ANP 工具

        Args:
            name: 工具名称
            description: 工具描述
            discovery: 可选的 ANPDiscovery 实例，如果不提供则创建新实例
            network: 可选的 ANPNetwork 实例，如果不提供则创建新实例
        """
        if description is None:
            description = "智能体网络管理工具，支持服务发现、节点管理和消息路由。概念性实现。"

        super().__init__(
            name=name,
            description=description
        )
        from hello_agents.protocols.anp.implementation import ANPDiscovery, ANPNetwork
        self._discovery = discovery if discovery is not None else ANPDiscovery()
        self._network = network if network is not None else ANPNetwork()
        
    def run(self, parameters: Dict[str, Any]) -> str:
        """
        执行 ANP 操作
        
        Args:
            parameters: 包含以下参数的字典
                - action: 操作类型 (register_service, discover_services, add_node, route_message, get_stats)
                - service_id, service_type, endpoint: 服务信息（register_service 需要）
                - node_id, endpoint: 节点信息（add_node 需要）
                - from_node, to_node, message: 路由信息（route_message 需要）
        
        Returns:
            操作结果
        """
        from hello_agents.protocols.anp.implementation import ServiceInfo

        action = parameters.get("action", "").lower()
        
        if not action:
            return "错误：必须指定 action 参数"
        
        try:
            if action == "register_service":
                service_id = parameters.get("service_id")
                service_type = parameters.get("service_type")
                endpoint = parameters.get("endpoint")
                metadata = parameters.get("metadata", {})
                
                if not all([service_id, service_type, endpoint]):
                    return "错误：必须指定 service_id, service_type 和 endpoint 参数"
                
                service = ServiceInfo(service_id, service_type, endpoint, metadata)
                self._discovery.register_service(service)
                return f"✅ 已注册服务 '{service_id}'"

            elif action == "unregister_service":
                service_id = parameters.get("service_id")
                if not service_id:
                    return "错误：必须指定 service_id 参数"

                # 使用 ANPDiscovery 的 unregister_service 方法
                success = self._discovery.unregister_service(service_id)

                if success:
                    return f"✅ 已注销服务 '{service_id}'"
                else:
                    return f"错误：服务 '{service_id}' 不存在"

            elif action == "discover_services":
                service_type = parameters.get("service_type")
                services = self._discovery.discover_services(service_type)

                if not services:
                    return "没有找到服务"

                result = f"找到 {len(services)} 个服务:\n\n"
                for service in services:
                    result += f"服务ID: {service.service_id}\n"
                    result += f"  名称: {service.service_name}\n"
                    result += f"  类型: {service.service_type}\n"
                    result += f"  端点: {service.endpoint}\n"
                    if service.capabilities:
                        result += f"  能力: {', '.join(service.capabilities)}\n"
                    if service.metadata:
                        result += f"  元数据: {service.metadata}\n"
                    result += "\n"
                return result
                
            elif action == "add_node":
                node_id = parameters.get("node_id")
                endpoint = parameters.get("endpoint")
                metadata = parameters.get("metadata", {})
                
                if not all([node_id, endpoint]):
                    return "错误：必须指定 node_id 和 endpoint 参数"
                
                self._network.add_node(node_id, endpoint, metadata)
                return f"✅ 已添加节点 '{node_id}'"
                
            elif action == "route_message":
                from_node = parameters.get("from_node")
                to_node = parameters.get("to_node")
                message = parameters.get("message", {})
                
                if not all([from_node, to_node]):
                    return "错误：必须指定 from_node 和 to_node 参数"
                
                path = self._network.route_message(from_node, to_node, message)
                if path:
                    return f"消息路由路径: {' -> '.join(path)}"
                else:
                    return "无法找到路由路径"
                
            elif action == "get_stats":
                stats = self._network.get_network_stats()
                result = "网络统计:\n"
                for key, value in stats.items():
                    result += f"- {key}: {value}\n"
                return result
                
            else:
                return f"错误：不支持的操作 '{action}'"
                
        except Exception as e:
            return f"ANP 操作失败: {str(e)}"
    
    def get_parameters(self) -> List[ToolParameter]:
        """获取工具参数定义"""
        return [
            ToolParameter(
                name="action",
                type="string",
                description="操作类型: register_service, unregister_service, discover_services, add_node, route_message, get_stats",
                required=True
            ),
            ToolParameter(
                name="service_id",
                type="string",
                description="服务 ID（register_service, unregister_service 需要）",
                required=False
            ),
            ToolParameter(
                name="service_type",
                type="string",
                description="服务类型（register_service 需要）",
                required=False
            ),
            ToolParameter(
                name="endpoint",
                type="string",
                description="端点地址（register_service, add_node 需要）",
                required=False
            ),
            ToolParameter(
                name="node_id",
                type="string",
                description="节点 ID（add_node 需要）",
                required=False
            ),
            ToolParameter(
                name="from_node",
                type="string",
                description="源节点 ID（route_message 需要）",
                required=False
            ),
            ToolParameter(
                name="to_node",
                type="string",
                description="目标节点 ID（route_message 需要）",
                required=False
            ),
            ToolParameter(
                name="message",
                type="object",
                description="消息内容（route_message 需要）",
                required=False
            ),
            ToolParameter(
                name="metadata",
                type="object",
                description="元数据（register_service, add_node 可选）",
                required=False
            )
        ]