当 Agent 有了工牌：钉钉群里的 Agent IAM 架构设计

用户在群里 @Bot
    ↓
钉钉推送消息到 Bot 服务端
    ↓
Bot 用「应用级 Token」调用钉钉 API
    ↓
结果发回群里

┌──────────────────────────────┐
│      Session Sandbox         │
│  ┌────────────────────────┐  │
│  │   Agent 推理 + 代码    │  │
│  │   (无凭证环境)         │  │
│  └────────┬───────────────┘  │
│           │ 出站请求          │
│           ▼                  │
│  ┌────────────────────────┐  │
│  │    Agent Proxy         │  │
│  │  (凭证边界注入)        │  │
│  └────────┬───────────────┘  │
└───────────┼──────────────────┘
            │ 带凭证的请求
            ▼
      外部服务 (GitHub, CRM...)

Workspace Baseline
  ├── general-read Bundle → 公共频道（只读内部文档 + Issue Tracker）
  ├── eng-triage Bundle  → 工程频道（GitHub Issues + PR + 日志，只读生产）
  ├── eng-pr Bundle      → 工程频道（可开 Branch + PR，不能 Merge）
  └── support-private    → 客服私有频道（CRM + Ticket，仅限私有频道）

App-Only Token（应用凭证）
  └── Client Credentials Grant
  └── 代表「应用自己」
  └── 有效期 2 小时，需缓存

User Token（用户凭证）
  └── OAuth 2.0 Authorization Code
  └── 代表「某个用户」
  └── 需要用户授权

┌─────────────────────────────────────────┐
│           钉钉开放平台                    │
│                                          │
│  ┌─────────────────────────────────┐     │
│  │   Agent Registry（新增）        │     │
│  │                                  │     │
│  │   agent://ops-analysis           │     │
│  │     群: 运营数据群               │     │
│  │     Bundle: ops-readonly         │     │
│  │     凭证: 边界注入               │     │
│  │                                  │     │
│  │   agent://eng-deploy             │     │
│  │     群: 工程发布群               │     │
│  │     Bundle: eng-pr               │     │
│  │     凭证: 边界注入               │     │
│  └─────────────────────────────────┘     │
│                                          │
│  ┌─────────────────────────────────┐     │
│  │   Agent Proxy Layer（新增）      │     │
│  │                                  │     │
│  │   Connection Rules:              │     │
│  │     ops-readonly → AI表格(R)     │     │
│  │     ops-readonly → 工单(R)       │     │
│  │     eng-pr → GitHub(RW-noMerge) │     │
│  │                                  │     │
│  │   Credential Store:              │     │
│  │     github_token: *** (write-only)│    │
│  │     crm_api_key: *** (write-only)│     │
│  └─────────────────────────────────┘     │
└─────────────────────────────────────────┘

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class AccessMode(Enum):
    READ = "read"
    WRITE = "write"
    NONE = "none"


@dataclass
class Connection:
    """一个外部系统的连接配置"""
    name: str                    # 如 "crm", "github", "ai-table"
    host: str                    # 出站域名白名单
    credential_ref: str          # 凭证引用（不存明文）
    mode: AccessMode = AccessMode.READ
    path_prefix: str = "/"       # 可选：限制到特定路径


@dataclass
class AccessBundle:
    """按场景打包的权限集合"""
    name: str
    connections: list[Connection] = field(default_factory=list)
    skills: list[str] = field(default_factory=list)
    standing_instructions: str = ""
    allowed_domains: list[str] = field(default_factory=list)


@dataclass
class AgentIdentity:
    """Agent 在组织中的身份"""
    agent_id: str                # agent://ops-analysis
    display_name: str            # "运营分析 Agent"
    department: str              # "运营部"
    bundle: AccessBundle
    memory_scope: str = "channel"  # channel | thread | workspace


class AgentProxy:
    """凭证边界注入代理——Agent 和外部世界之间的唯一出口"""

    def __init__(self, identity: AgentIdentity, credential_store: dict):
        self.identity = identity
        self.credential_store = credential_store  # {ref: encrypted_credential}

    def forward_request(self, target_host: str, path: str, method: str) -> dict:
        """处理 Agent 的出站请求"""
        bundle = self.identity.bundle

        # 1. 检查是否命中 Connection Rule
        for conn in bundle.connections:
            if target_host == conn.host and path.startswith(conn.path_prefix):
                # 检查访问模式
                if method in ("POST", "PUT", "DELETE") and conn.mode == AccessMode.READ:
                    return {"status": "blocked", "reason": "write_not_allowed"}

                # 注入凭证（Agent 永远看不到这个值）
                credential = self.credential_store.get(conn.credential_ref)
                return {
                    "status": "forwarded",
                    "credential_attached": True,
                    "credential_visible_to_agent": False,
                }

        # 2. 检查网络白名单
        if target_host in bundle.allowed_domains:
            return {"status": "forwarded", "credential_attached": False}

        # 3. 都不匹配——阻断
        return {"status": "blocked", "reason": "host_not_allowed"}
# generated by hugo AI

{
  "timestamp": "2026-06-29T09:30:00+08:00",
  "human_requester": "张三 (运营)",
  "agent_executor": "agent://ops-analysis",
  "session_id": "ses_thread_abc123",
  "action": "query_refund_data",
  "data_source": "ai-table://refund_view",
  "access_bundle": "ops-readonly",
  "result_ref": "report_20260629_001",
  "proxy_log": [
    {"host": "table.dingtalk.com", "status": "forwarded", "credential": "table_token"},
    {"host": "ticket.dingtalk.com", "status": "forwarded", "credential": "ticket_key"},
    {"host": "unknown-api.com", "status": "blocked", "reason": "host_not_allowed"}
  ]
}

Agent: 我需要给客户发送退款确认邮件。这需要「邮件发送」权限。
       [审批请求已发送给运营总监]

运营总监: [钉钉卡片] 运营 Agent 请求发送邮件权限
          目标: customer@example.com
          内容: 退款确认模板 #3
          [批准] [拒绝] [仅限本次]

Agent: 已发送邮件。审计记录: JIT grant #jit_001, 批准人: 运营总监