深入 ‘Tool Delegation’：如何让一个 Agent 动态地为另一个 Agent 授予临时的工具访问权限？ - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位同学，大家好！

欢迎来到今天的技术讲座。今天我们将深入探讨一个在现代智能体（Agent）系统中至关重要且极具挑战性的主题——工具委托（Tool Delegation）。具体来说，我们将聚焦于一个核心问题：如何让一个智能体（Agent A）动态地、安全地为另一个智能体（Agent B）授予临时的工具访问权限。

在智能体协作日益复杂的今天，我们经常会遇到这样的场景：一个主导智能体负责规划宏观任务，但其中某些子任务需要其他专业智能体来执行。而这些专业智能体可能并不总是拥有完成任务所需的所有工具权限，或者出于安全和效率考虑，我们不希望它们永久拥有这些权限。这时，工具委托机制就显得尤为关键。它允许权限的动态流转，遵循最小权限原则，并确保系统资源的合理利用。

我们将从概念、架构、权限模型、安全考量到具体的代码实现，一步步揭示工具委托的奥秘。

一、智能体协作与工具的基石

在深入工具委托之前，我们先快速回顾一下智能体系统中的两个核心概念：智能体和工具。

1. 智能体 (Agent)

智能体是具备感知、推理、决策和行动能力的自主实体。在我们的语境中，它们可以是：

规划智能体 (Planning Agent)：负责任务分解、策略制定。
执行智能体 (Execution Agent)：负责调用具体工具完成操作。
数据分析智能体 (Data Analysis Agent)：处理数据、生成洞察。
用户界面智能体 (UI Agent)：与用户交互。

每个智能体都有其独特的标识、能力集和预设的角色。

2. 工具 (Tool)

工具是智能体用来与外部世界交互或执行特定操作的函数、API 接口或服务。它们通常封装了特定的功能，例如：

文件操作工具：read_file, write_file, list_directory
网络请求工具：make_http_request, fetch_api_data
数据库工具：query_database, update_record
计算工具：perform_complex_calculation
AI模型工具：generate_image, translate_text

工具是智能体能力的延伸，没有工具，智能体就如同没有手脚，无法真正行动。

二、为什么需要工具委托？

既然智能体可以拥有工具，为什么还需要将工具权限委托给其他智能体呢？这背后有几个核心驱动力：

任务分解与专业化协作
- 一个复杂的大任务往往需要分解成多个子任务，由不同的专业智能体并行或串行执行。主导智能体可能没有所有子任务所需的专业工具，或者不适合直接执行这些子任务。
- 示例：一个“项目管理智能体”需要“代码生成智能体”去编写代码，而“代码生成智能体”又需要“文件写入工具”来保存代码。项目管理智能体可能无权直接写入文件，但它可以授权代码生成智能体在特定目录下写入。
最小权限原则 (Principle of Least Privilege)
- 安全最佳实践要求任何实体（包括智能体）只应拥有完成其任务所需的最小权限集。永久授予所有权限会增加安全风险。
- 通过委托，我们可以为智能体提供临时的、细粒度的权限，任务完成后自动撤销或过期。
资源优化与负载均衡
- 某些工具可能资源密集，或有并发限制。通过委托，可以将工具调用分发给当前空闲或更适合执行的智能体，实现负载均衡。
动态适应性
- 在动态环境中，智能体的任务和所需工具可能会实时变化。工具委托机制允许系统灵活地调整权限，而无需重新配置或部署智能体。
审计与溯源
- 明确的委托链和访问日志可以帮助我们追踪哪个智能体在何时、由谁授权、使用了哪个工具，这对于调试、安全审计和合规性至关重要。

三、工具委托的核心挑战

实现一个健壮的工具委托系统并非易事，我们需要面对以下几个关键挑战：

身份与认证 (Identity and Authentication)：如何唯一识别智能体？如何验证其身份？
授权与权限模型 (Authorization and Permission Model)：如何定义工具的访问权限？如何表示“授予”和“被授予”的关系？
动态性与临时性 (Dynamism and Temporality)：如何实现权限的实时授予、撤销和自动过期？
粒度控制 (Granularity Control)：权限可以细化到什么程度？是整个工具，还是工具的特定功能，甚至特定参数？
安全性 (Security)：如何防止恶意委托、权限滥用或未经授权的访问？
审计与可追溯性 (Auditing and Traceability)：如何记录所有委托事件和工具使用情况？
复杂性管理 (Complexity Management)：随着智能体和工具数量的增加，如何有效管理委托关系？

四、委托系统架构设计

为了应对上述挑战，我们需要一个清晰、模块化的系统架构。以下是一个典型的工具委托系统组件图及其职责：

graph TD
    A[Agent A (Delegator)] --> R{Delegation Request};
    R --> DM[Delegation Manager Service];
    DM --> PS[Permission Store];
    DM --> AS[Audit Service];
    DM --> PE[Permission Evaluator];

    B[Agent B (Delegatee)] --> TE[Tool Executor];
    TE --> PE;
    PE --> PS;
    TE --> TS[Tool Service/Registry];
    TS --> PT[Actual Tool Implementations];

    subgraph Core Services
        DM
        PS
        PE
        AS
        TS
    end

核心组件职责：

智能体 (Agent A & Agent B)
- Agent A (Delegator)：发起委托请求，拥有对特定工具的原始权限，并希望将其临时授予 Agent B。
- Agent B (Delegatee)：接收委托，在获得权限后尝试使用工具。
工具注册与服务 (Tool Service/Registry)
- 负责注册所有可用的工具，提供工具的元数据（名称、描述、所需参数、原始权限要求等）。
- 封装实际的工具实现，提供统一的调用接口。
权限存储 (Permission Store)
- 持久化存储所有智能体的权限信息，包括原始权限和通过委托获得的临时权限。
- 可以是数据库、键值存储或分布式配置系统。
授权评估器 (Permission Evaluator)
- 根据智能体身份、请求的工具和操作，查询权限存储，判断智能体是否拥有执行该操作的权限。
- 需要能够识别并处理委托权限。
委托管理器服务 (Delegation Manager Service)
- 这是整个委托机制的核心。它负责：
  - 接收来自 Agent A 的委托请求。
  - 验证 Agent A 是否有权委托该工具。
  - 在权限存储中创建或更新委托记录。
  - 管理委托的生命周期（过期、撤销）。
  - 生成和验证委托凭证（如果使用）。
工具执行器 (Tool Executor)
- 智能体通过此组件来调用工具。
- 在调用前，它会先请求授权评估器检查权限。
- 如果权限通过，则调用工具注册服务执行实际工具。
审计服务 (Audit Service)
- 记录所有关键事件：委托请求、批准、拒绝、撤销、过期、工具调用尝试（成功/失败）。
- 提供可追溯性，帮助诊断问题和满足合规性要求。

五、权限模型与委托凭证

在实现委托时，权限模型的选择至关重要。我们将采用一种基于访问控制列表（ACL）和角色基访问控制（RBAC）的混合模型，并引入委托凭证（Delegation Token）的概念来处理临时性和动态性。

1. 权限定义

我们首先需要定义工具的抽象表示和权限的结构。

import uuid
import time
from datetime import datetime, timedelta
from typing import Callable, Dict, Any, Optional, List

# --- 1. 工具定义 ---
class Tool:
    """
    表示一个智能体可以使用的工具。
    """
    def __init__(self, name: str, description: str, func: Callable, required_permissions: Optional[List[str]] = None):
        self.name = name
        self.description = description
        self.func = func
        self.required_permissions = required_permissions if required_permissions is not None else ["execute"] # 默认执行权限

    def __str__(self):
        return f"Tool(name='{self.name}', description='{self.description}')"

    def execute(self, **kwargs) -> Any:
        """
        执行工具的实际功能。
        """
        print(f"Executing tool: {self.name} with args: {kwargs}")
        return self.func(**kwargs)

# 示例工具函数
def _read_file_content(path: str) -> str:
    print(f"Reading file: {path}")
    # 模拟文件读取
    if path == "sensitive_data.txt":
        return "This is sensitive data."
    return f"Content of {path}"

def _write_file_content(path: str, content: str) -> str:
    print(f"Writing '{content}' to file: {path}")
    # 模拟文件写入
    return f"Successfully wrote to {path}"

def _search_web(query: str) -> str:
    print(f"Searching web for: {query}")
    # 模拟网络搜索
    return f"Search results for '{query}': Example search result."

# --- 2. 智能体定义 (简化) ---
class Agent:
    """
    表示一个智能体，具有唯一ID。
    """
    def __init__(self, agent_id: str, name: str):
        self.agent_id = agent_id
        self.name = name

    def __str__(self):
        return f"Agent(id='{self.agent_id}', name='{self.name}')"

# --- 3. 权限记录 ---
class PermissionRecord:
    """
    表示一个智能体对某个工具的权限。
    可以是原始权限或委托权限。
    """
    def __init__(self,
                 agent_id: str,
                 tool_name: str,
                 action: str, # 例如 "execute", "read", "write"
                 granted_by: str, # 授予者ID，如果是系统授予则是 "system"
                 expires_at: Optional[datetime] = None,
                 delegation_id: Optional[str] = None): # 如果是委托权限，则有委托ID

        self.record_id = str(uuid.uuid4())
        self.agent_id = agent_id
        self.tool_name = tool_name
        self.action = action
        self.granted_by = granted_by
        self.expires_at = expires_at
        self.delegation_id = delegation_id # 关联到 DelegationToken 或 DelegationRequest

    def is_expired(self) -> bool:
        return self.expires_at is not None and self.expires_at < datetime.now()

    def __str__(self):
        expiry_info = f", Expires: {self.expires_at.strftime('%Y-%m-%d %H:%M:%S')}" if self.expires_at else ""
        delegation_info = f", Delegation ID: {self.delegation_id[:8]}..." if self.delegation_id else ""
        return (f"PermissionRecord(Agent='{self.agent_id}', Tool='{self.tool_name}', "
                f"Action='{self.action}', GrantedBy='{self.granted_by}'{expiry_info}{delegation_info})")

# --- 4. 委托凭证 (Delegation Token) ---
class DelegationToken:
    """
    一个临时的、可分发的凭证，代表了对特定工具的委托权限。
    它包含足够的信息供授权评估器验证。
    """
    def __init__(self,
                 delegator_id: str, # 委托方智能体ID
                 delegatee_id: str, # 被委托方智能体ID
                 tool_name: str,
                 action: str,
                 expires_at: datetime,
                 reason: str = "temporary task"):

        self.token_id = str(uuid.uuid4())
        self.delegator_id = delegator_id
        self.delegatee_id = delegatee_id
        self.tool_name = tool_name
        self.action = action
        self.expires_at = expires_at
        self.reason = reason
        # 实际系统中，这里可能还需要一个签名来防止篡改
        self.signature = self._generate_signature() # 简化处理，实际中会用加密算法

    def _generate_signature(self) -> str:
        # 模拟签名，实际应使用私钥加密
        return f"signed_by_{self.delegator_id}_at_{int(time.time())}"

    def is_valid(self) -> bool:
        return datetime.now() < self.expires_at

    def __str__(self):
        return (f"DelegationToken(ID='{self.token_id[:8]}...', Delegator='{self.delegator_id}', "
                f"Delegatee='{self.delegatee_id}', Tool='{self.tool_name}', Action='{self.action}', "
                f"Expires='{self.expires_at.strftime('%Y-%m-%d %H:%M:%S')}', Valid={self.is_valid()})")

权限存储结构：

我们使用一个简单的字典来模拟 Permission Store，它存储 PermissionRecord 对象。

六、核心服务实现

现在，我们来实现上面提到的核心服务组件。

1. 工具注册表 (ToolRegistry)

class ToolRegistry:
    """
    负责注册和管理所有可用的工具。
    """
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register_tool(self, tool: Tool):
        if tool.name in self._tools:
            raise ValueError(f"Tool '{tool.name}' already registered.")
        self._tools[tool.name] = tool
        print(f"Tool '{tool.name}' registered.")

    def get_tool(self, tool_name: str) -> Optional[Tool]:
        return self._tools.get(tool_name)

    def list_tools(self) -> List[Tool]:
        return list(self._tools.values())

# 初始化并注册一些工具
tool_registry = ToolRegistry()
tool_registry.register_tool(Tool("read_file", "Reads content from a specified file path.", _read_file_content))
tool_registry.register_tool(Tool("write_file", "Writes content to a specified file path.", _write_file_content))
tool_registry.register_tool(Tool("search_web", "Performs a web search for a given query.", _search_web))

2. 权限存储 (PermissionStore)

class PermissionStore:
    """
    管理所有智能体的原始权限和委托权限。
    """
    def __init__(self):
        # {agent_id: {tool_name: {action: [PermissionRecord, ...]}}}
        self._permissions: Dict[str, Dict[str, Dict[str, List[PermissionRecord]]]] = {}
        # {delegation_id: DelegationToken}
        self._active_delegations: Dict[str, DelegationToken] = {}

    def add_permission(self, record: PermissionRecord):
        self._permissions.setdefault(record.agent_id, {}).setdefault(record.tool_name, {}).setdefault(record.action, []).append(record)
        print(f"Added permission: {record}")

    def remove_permission(self, record_id: str):
        for agent_perms in self._permissions.values():
            for tool_perms in agent_perms.values():
                for action_perms in tool_perms.values():
                    # 查找并移除
                    initial_len = len(action_perms)
                    action_perms[:] = [p for p in action_perms if p.record_id != record_id]
                    if len(action_perms) < initial_len:
                        print(f"Removed permission record: {record_id}")
                        return

    def get_permissions_for_agent(self, agent_id: str, tool_name: str, action: str) -> List[PermissionRecord]:
        agent_tool_action_perms = self._permissions.get(agent_id, {}).get(tool_name, {}).get(action, [])
        # 过滤掉已过期的权限
        return [p for p in agent_tool_action_perms if not p.is_expired()]

    def add_delegation_token(self, token: DelegationToken):
        if not token.is_valid():
            raise ValueError("Cannot add an expired delegation token.")
        self._active_delegations[token.token_id] = token
        print(f"Added active delegation token: {token}")

        # 同时为被委托方创建一条临时的 PermissionRecord
        delegated_record = PermissionRecord(
            agent_id=token.delegatee_id,
            tool_name=token.tool_name,
            action=token.action,
            granted_by=token.delegator_id,
            expires_at=token.expires_at,
            delegation_id=token.token_id
        )
        self.add_permission(delegated_record)

    def get_delegation_token(self, token_id: str) -> Optional[DelegationToken]:
        token = self._active_delegations.get(token_id)
        if token and not token.is_valid():
            self.revoke_delegation_token(token_id) # 自动清理过期token
            return None
        return token

    def revoke_delegation_token(self, token_id: str):
        if token_id in self._active_delegations:
            token = self._active_delegations.pop(token_id)
            print(f"Revoked delegation token: {token_id}")

            # 移除所有关联的 PermissionRecord
            for agent_id, agent_perms in list(self._permissions.items()):
                for tool_name, tool_perms in list(agent_perms.items()):
                    for action, records in list(tool_perms.items()):
                        initial_len = len(records)
                        # 注意：这里需要创建一个新列表来避免在迭代时修改原列表
                        tool_perms[action] = [p for p in records if p.delegation_id != token_id]
                        if not tool_perms[action]: # 如果某个action下的权限为空，则清理
                            del tool_perms[action]
                if not agent_perms: # 如果某个agent的权限为空，则清理
                    del self._permissions[agent_id]
            print(f"Removed associated permission records for delegation {token_id}")

    def cleanup_expired_delegations(self):
        """
        定期清理已过期的委托令牌和权限记录。
        """
        expired_token_ids = [tid for tid, token in self._active_delegations.items() if not token.is_valid()]
        for token_id in expired_token_ids:
            self.revoke_delegation_token(token_id)
        print(f"Cleaned up {len(expired_token_ids)} expired delegation tokens.")

permission_store = PermissionStore()

3. 授权评估器 (PermissionEvaluator)

class PermissionEvaluator:
    """
    负责检查智能体是否拥有执行特定工具的权限。
    """
    def __init__(self, permission_store: PermissionStore, tool_registry: ToolRegistry):
        self.permission_store = permission_store
        self.tool_registry = tool_registry

    def check_permission(self, agent_id: str, tool_name: str, action: str = "execute") -> bool:
        """
        检查智能体是否拥有对指定工具的指定操作权限。
        """
        tool = self.tool_registry.get_tool(tool_name)
        if not tool:
            print(f"Permission check failed: Tool '{tool_name}' not found.")
            return False

        # 1. 检查是否有直接的非过期权限
        direct_permissions = self.permission_store.get_permissions_for_agent(agent_id, tool_name, action)
        if direct_permissions:
            print(f"Permission granted for agent '{agent_id}' to '{action}' tool '{tool_name}' via direct permission.")
            return True

        # 2. 如果没有直接权限，理论上这里也可以检查是否存在通过有效委托获得的权限
        # 但我们设计的 PermissionStore 在 add_delegation_token 时已经为 delegatee_id 写入了 PermissionRecord，
        # 所以第一步的 direct_permissions 已经包含了委托权限。
        # 故此，如果第一步返回False，则意味着没有权限。

        print(f"Permission denied for agent '{agent_id}' to '{action}' tool '{tool_name}'. No valid permission found.")
        return False

permission_evaluator = PermissionEvaluator(permission_store, tool_registry)

4. 委托管理器服务 (DelegationManagerService)

这是实现“Agent A 动态地为 Agent B 授予临时工具访问权限”的关键。

class DelegationManagerService:
    """
    管理工具的委托生命周期，包括请求、批准、撤销。
    """
    def __init__(self, permission_store: PermissionStore, permission_evaluator: PermissionEvaluator, tool_registry: ToolRegistry):
        self.permission_store = permission_store
        self.permission_evaluator = permission_evaluator
        self.tool_registry = tool_registry

    def request_and_grant_delegation(
        self,
        delegator_agent_id: str, # 委托方 Agent A
        delegatee_agent_id: str, # 被委托方 Agent B
        tool_name: str,
        action: str = "execute",
        duration_seconds: int = 3600, # 默认1小时
        reason: str = "temporary task execution"
    ) -> Optional[DelegationToken]:
        """
        由 delegator_agent_id 发起请求，为 delegatee_agent_id 授予对指定工具的临时权限。
        此方法封装了请求和授予的逻辑，假设 delegator_agent_id 有权进行委托。
        """
        print(f"n--- Delegation Request ---")
        print(f"Delegator: {delegator_agent_id}, Delegatee: {delegatee_agent_id}, Tool: {tool_name}, Action: {action}")

        # 1. 验证委托方 (delegator_agent_id) 是否拥有原始权限来委托此工具
        # 委托方必须拥有它想要委托的权限，这是安全的基础。
        if not self.permission_evaluator.check_permission(delegator_agent_id, tool_name, action):
            print(f"Delegation rejected: Delegator '{delegator_agent_id}' does not have original permission to '{action}' tool '{tool_name}'.")
            return None

        # 2. 验证工具是否存在
        if not self.tool_registry.get_tool(tool_name):
            print(f"Delegation rejected: Tool '{tool_name}' does not exist.")
            return None

        # 3. 生成委托令牌
        expires_at = datetime.now() + timedelta(seconds=duration_seconds)
        delegation_token = DelegationToken(
            delegator_id=delegator_agent_id,
            delegatee_id=delegatee_agent_id,
            tool_name=tool_name,
            action=action,
            expires_at=expires_at,
            reason=reason
        )

        # 4. 将委托令牌添加到权限存储，同时为被委托方创建 PermissionRecord
        self.permission_store.add_delegation_token(delegation_token)
        print(f"Delegation granted and token issued: {delegation_token}")
        return delegation_token

    def revoke_delegation(self, token_id: str, revoker_agent_id: str) -> bool:
        """
        撤销一个委托令牌。只有原始委托方或系统管理员才能撤销。
        """
        print(f"n--- Delegation Revocation Request ---")
        print(f"Revoker: {revoker_agent_id}, Token ID: {token_id}")

        token = self.permission_store.get_delegation_token(token_id)
        if not token:
            print(f"Revocation failed: Delegation token '{token_id}' not found or already expired.")
            return False

        # 验证撤销方是否有权撤销此委托 (简化：只有原始委托方或'system'可以撤销)
        if revoker_agent_id != token.delegator_id and revoker_agent_id != "system":
            print(f"Revocation failed: Agent '{revoker_agent_id}' is not authorized to revoke this delegation.")
            return False

        self.permission_store.revoke_delegation_token(token_id)
        print(f"Delegation token '{token_id}' successfully revoked by '{revoker_agent_id}'.")
        return True

    def get_delegation_status(self, token_id: str) -> Optional[DelegationToken]:
        """
        获取委托令牌的状态。
        """
        return self.permission_store.get_delegation_token(token_id)

delegation_manager = DelegationManagerService(permission_store, permission_evaluator, tool_registry)

5. 智能体执行器 (AgentExecutor)

智能体通过这个执行器来调用工具。

class AgentExecutor:
    """
    智能体使用此执行器来调用工具。
    它会在调用前检查权限。
    """
    def __init__(self, agent: Agent, permission_evaluator: PermissionEvaluator, tool_registry: ToolRegistry):
        self.agent = agent
        self.permission_evaluator = permission_evaluator
        self.tool_registry = tool_registry

    def execute_tool(self, tool_name: str, **kwargs) -> Any:
        print(f"n--- Agent {self.agent.name} attempting to execute tool '{tool_name}' ---")
        # 1. 权限检查
        if not self.permission_evaluator.check_permission(self.agent.agent_id, tool_name, "execute"):
            print(f"Execution failed: Agent '{self.agent.name}' lacks permission for tool '{tool_name}'.")
            return f"Error: Permission denied for tool '{tool_name}'."

        # 2. 获取工具实例
        tool = self.tool_registry.get_tool(tool_name)
        if not tool:
            print(f"Execution failed: Tool '{tool_name}' not found in registry.")
            return f"Error: Tool '{tool_name}' not found."

        # 3. 执行工具
        try:
            result = tool.execute(**kwargs)
            print(f"Execution successful: Agent '{self.agent.name}' executed tool '{tool_name}'. Result: {result}")
            return result
        except Exception as e:
            print(f"Execution failed for tool '{tool_name}': {e}")
            return f"Error executing tool '{tool_name}': {e}"

七、实践案例：主智能体委托任务给辅助智能体

现在，让我们通过一个具体的场景来演示这套委托系统的工作流程。

场景描述：

Agent_P (Planner Agent)：项目规划智能体，拥有“读取文件”和“委托工具”的原始权限。它负责协调任务。
Agent_D (Developer Agent)：开发者智能体，初始没有任何工具权限。它需要根据指令编写代码并写入文件。
Goal: Agent_P 发现需要 Agent_D 来完成一个“写入文件”的任务。Agent_P 将临时授予 Agent_D “写入文件”的权限。

步骤：

初始化智能体和系统权限。
Agent_P 尝试直接写入文件（失败，因为它没有写权限）。
Agent_P 授权给 Agent_D 写入文件的临时权限。
Agent_D 尝试写入文件（成功，因为获得了临时权限）。
Agent_D 尝试读取文件（失败，因为它没有读权限）。
模拟委托过期或撤销。
Agent_D 再次尝试写入文件（失败，因为权限已过期/撤销）。

# --- 1. 初始化智能体 ---
agent_p = Agent("agent-p-123", "Planner Agent")
agent_d = Agent("agent-d-456", "Developer Agent")
agent_s = Agent("system-admin", "System Admin") # 用于模拟系统权限

# --- 2. 初始化智能体的原始权限 ---
# Planner Agent (agent_p) 拥有读取文件和委托权限
permission_store.add_permission(PermissionRecord(agent_p.agent_id, "read_file", "execute", "system"))
permission_store.add_permission(PermissionRecord(agent_p.agent_id, "write_file", "execute", "system", expires_at=datetime.now() + timedelta(days=1))) # Planner Agent 自身拥有写权限，才能委托
permission_store.add_permission(PermissionRecord(agent_p.agent_id, "search_web", "execute", "system"))

# Developer Agent (agent_d) 初始没有任何权限

# 创建执行器
executor_p = AgentExecutor(agent_p, permission_evaluator, tool_registry)
executor_d = AgentExecutor(agent_d, permission_evaluator, tool_registry)

print("n--- 初始权限设置完成 ---")

# --- 2. Agent_P 尝试直接写入文件 (成功，因为自身有权限) ---
print("n--- Agent_P 尝试直接写入文件 ---")
executor_p.execute_tool("write_file", path="project_plan.txt", content="Initial project plan.")

# --- 3. Agent_D 尝试写入文件 (失败，因为它没有写权限) ---
print("n--- Agent_D 尝试直接写入文件 (预期失败) ---")
executor_d.execute_tool("write_file", path="code_output.py", content="print('Hello World')")

# --- 4. Agent_P 委托 Agent_D 写入文件的临时权限 ---
print("n--- Agent_P 委托 Agent_D 写入文件的临时权限 ---")
# 委托 60 秒的权限
delegation_token_for_write = delegation_manager.request_and_grant_delegation(
    delegator_agent_id=agent_p.agent_id,
    delegatee_agent_id=agent_d.agent_id,
    tool_name="write_file",
    action="execute",
    duration_seconds=60, # 授予 60 秒的临时权限
    reason="To write generated code to file."
)

if delegation_token_for_write:
    print(f"Delegation successful. Agent_D received token: {delegation_token_for_write.token_id[:8]}...")
else:
    print("Delegation failed.")

# --- 5. Agent_D 尝试写入文件 (成功，因为获得了临时权限) ---
print("n--- Agent_D 再次尝试写入文件 (预期成功) ---")
executor_d.execute_tool("write_file", path="generated_script.py", content="import osnprint('Script executed!')")

# --- 6. Agent_D 尝试读取文件 (失败，因为它没有被委托读权限) ---
print("n--- Agent_D 尝试读取文件 (预期失败) ---")
executor_d.execute_tool("read_file", path="some_file.txt")

# --- 7. 模拟时间流逝，委托过期 ---
print("n--- 模拟时间流逝，等待委托过期 (65秒) ---")
import time
time.sleep(65) # 等待 60 秒 + 5 秒，确保过期

# 清理过期委托，确保PermissionStore中的记录也更新
permission_store.cleanup_expired_delegations()

# --- 8. Agent_D 再次尝试写入文件 (失败，因为权限已过期) ---
print("n--- Agent_D 再次尝试写入文件 (预期失败，权限已过期) ---")
executor_d.execute_tool("write_file", path="another_script.py", content="print('This should fail now!')")

# --- 9. Agent_P 撤销对 Agent_D 的委托 (如果未过期) ---
# 假设我们想提前撤销
print("n--- 演示委托撤销 ---")
# 先授予一个新权限进行演示
new_delegation_token = delegation_manager.request_and_grant_delegation(
    delegator_agent_id=agent_p.agent_id,
    delegatee_agent_id=agent_d.agent_id,
    tool_name="search_web",
    action="execute",
    duration_seconds=300, # 授予 5 分钟
    reason="Temporary web search"
)
if new_delegation_token:
    print(f"New delegation for search_web: {new_delegation_token.token_id[:8]}...")
    executor_d.execute_tool("search_web", query="latest AI trends")
    # 立即撤销
    delegation_manager.revoke_delegation(new_delegation_token.token_id, agent_p.agent_id)
    print("n--- Agent_D 再次尝试搜索 (预期失败，权限已撤销) ---")
    executor_d.execute_tool("search_web", query="latest AI trends after revocation")

八、高级考量与未来方向

上述系统提供了一个坚实的基础，但在生产环境中，我们还需要考虑更高级的场景和优化：

嵌套委托与委托链
- Agent A 委托 Agent B，Agent B 又能否将它获得的权限（或一部分）委托给 Agent C？这需要明确的策略和严格的审计。通常，直接通过委托获得的权限是不能再被二次委托的，除非有明确的can_delegate权限。
细粒度权限控制
- 目前我们仅控制到工具级别。更精细的控制可能包括：
  - 参数级别：例如，write_file工具只能写入/tmp目录，不能写入/etc。
  - 数据级别：只能访问数据库中特定表或特定行的数据。
- 这需要更复杂的策略引擎（例如基于属性的访问控制 ABAC）和工具接口设计。
分布式与高可用
- 在分布式系统中，PermissionStore 和 DelegationManager 需要是高可用和可扩展的。这可能涉及分布式数据库、消息队列和共识机制。
安全加固
- 凭证签名与加密：委托令牌应进行数字签名，以防篡改，并可能加密敏感信息。
- TLS/SSL：所有服务间通信都应通过加密通道进行。
- 速率限制：防止智能体恶意请求大量委托或滥用工具。
- 权限审查：定期审查智能体的权限，清理不必要的或长期未使用的权限。
用户界面与管理
- 需要一个管理界面，允许人类管理员查看、管理和审计智能体的权限和委托。
与现有身份和访问管理 (IAM) 系统集成
- 将智能体的身份管理与现有的企业 IAM 解决方案（如 OAuth 2.0, OpenID Connect, LDAP）集成，以利用其成熟的认证和授权基础设施。

展望未来智能体系统的协作模式

通过今天对工具委托的深入探讨，我们看到了一个强大的机制，它能够赋能智能体系统实现更安全、更灵活、更高效的协作。动态、临时的工具访问权限，不仅是遵循最小权限原则的关键实践，更是构建复杂、自适应智能体生态系统的基石。随着智能体技术的不断演进，工具委托将成为连接不同智能体能力、优化资源分配、确保系统韧性的核心组件。理解并掌握这一技术，对于设计和部署未来的智能体系统至关重要。

一、 智能体协作与工具的基石

二、 为什么需要工具委托？

三、 工具委托的核心挑战

四、 委托系统架构设计

五、 权限模型与委托凭证

六、 核心服务实现

七、 实践案例：主智能体委托任务给辅助智能体

八、 高级考量与未来方向