代理身份安全
人工智能代理的身份治理和管理（IGA）。使用LangGraph构建。
该系统将SailPoint用于人类身份的相同治理原则（发现、所有权、认证、补救）应用于跨多云环境部署的AI代理。
它的作用
问题： 组织正在跨AWS、Azure、GCP和Salesforce部署AI代理。许多是未注册的、不受监控的、过度许可的。安全团队不知道那里有什么，谁拥有它，或者它可以访问什么。
该系统：
- 发现 跨云平台的AI代理——显式（Bedrock代理、Copilot Studio）和隐式（调用LLM API的Lambda函数）
- 寄存器 每个代理具有唯一的身份，并评估治理的完整性
- 丰富 代理缺少关键元数据——分类、所有权、影子AI评估
- 评论 每个代理对安全策略的访问，标记违规行为，并检测人类通过人工智能代理获得提升权限的间接访问路径
- 证明 审计报告中的结果，包括优先补救建议
建筑
两个LangGraph图，每个图通过Send API使用动态编排器驱动的扇出。
图1：发现与丰富
START
│
▼
discovery_orchestrator (LLM analyzes environment, generates scanner AgentSpecs)
│
├──── Send ────┬──── Send ────┬──── Send ────┬──── Send ────┐
▼ ▼ ▼ ▼ ▼
execute_scanner execute_scanner execute_scanner execute_scanner execute_scanner
(AWS) (Azure) (GCP) (Salesforce) (MCP Connector)
│ │ │ │ │
└──────────────┴──────────────┴──────────────┴──────────────┘
│
▼
register_agents (assign unique IDs, deduplicate)
│
▼
enrichment_orchestrator (analyze gaps, dynamically generate AgentSpecs)
│ Only assigns agents with missing fields
│
├──── Send ────┬──── Send ────┬──── Send ────┐
▼ ▼ ▼ ▼
execute_ execute_ execute_ (however many the
enrichment enrichment enrichment orchestrator decides)
│ │ │
└──────────────┴──────────────┘
│
▼
critique_enrichments (validate updates — are classifications accurate?)
│
├── rejected → execute_enrichment (retry with feedback, max 2 retries)
└── approved ↓
│
▼
merge_enrichments (apply approved updates to registry by agent ID)
│
▼
summarize
│
▼
END图2：访问审查和认证
START
│
▼
review_orchestrator (LLM analyzes registry + policies, dynamically generates reviewer AgentSpecs)
│
├──── Send ────┬──── Send ────┬──── Send ────┐
▼ ▼ ▼ ▼
execute_ execute_ execute_ (however many the
reviewer reviewer reviewer orchestrator decides)
│ │ │
└──────────────┴──────────────┘
│
▼
generate_certification_report
│
▼
END关键设计决策
动态编排 --编排器由LLM驱动。他们分析输入并决定生成哪些代理、生成多少代理以及使用什么指令。没有硬编码的管道。
单一真相来源 --所有代理的定义见 data_sources/agents.py 无论平台提供什么字段。有些完整地回来了，有些稀疏。可以在运行时从外部MCP服务器中提取其他代理。
确定性差距分析+健全性检查+LLM智能 --Python逻辑确定哪些代理需要丰富（没有遗漏的代理），并根据代理自己的元数据验证预填充的扫描程序数据（不盲目信任平台报告的值）。LLM编排器生成专业提示，并可以添加代理进行价值审查。三个层次：代码的精度、规则的验证、人工智能的判断。
结构化工具调用 --代理通过键入的工具（例如。， report_discovered_agent, flag_policy_violation).结构化数据直接从工具调用中提取，无需对自由文本进行JSON解析。
两层发现 -扫描仪检查显式代理服务（Bedrock、Copilot Studio）和隐式信号（具有LLM API键的Lambda函数）。该系统撒下一张宽网，然后在富集过程中过滤误报。
MCP作为数据源 --扫描仪可以作为客户端连接到外部MCP服务器，实时从公司平台拉取代理。同样的管道，实时数据。
每个字段的评论循环 --在合并之前，逐个字段验证扩展更新。正确的字段会立即合并；只有被拒绝的字段才能通过反馈重新丰富。评论日志跟踪每次拒绝，以便手动审查。
选项A状态管理 --丰富更新通过以下方式累积在列表中 operator.add，然后合并节点通过代理ID将批准的更新应用于注册表。这保留了每个专家所写内容的完整审计跟踪。
治理检查表 --必填字段的静态列表定义了“良好信誉”的含义。富集协调器只为有缺口的代理配备专家。
项目结构
.
├── state.py # State schemas: DiscoveryState, AccessReviewState, all TypedDicts
├── tools.py # Action tools (scanner, enrichment, review) + reporting tools
├── prompts.py # System prompts for 3 orchestrators + background context
├── run.py # Full pipeline — runs both graphs end to end
├── config/
│ └── environment.py # Graph inputs (environment config, team directory, policies)
├── data_sources/
│ ├── agents.py # Agent definitions — single source of truth for all agents
│ ├── scanners.py # Scanner tool responses (reads from agents.py by platform)
│ ├── mcp_connector.py # Live MCP client — pulls agents from external MCP servers
│ ├── enrichment.py # Reference data (agent inventory, deployment records)
│ └── review.py # Reference data (invokers, invoker permissions)
├── discovery/
│ ├── graph.py # Graph 1 wiring — nodes, edges, Send fan-out, critique loop
│ ├── run.py # Entry point with streaming output
│ └── nodes/
│ ├── orchestrator.py # discovery_orchestrator + route_to_scanners
│ ├── scanners.py # execute_scanner (tool-call loop)
│ ├── registry.py # register_agents (multi-key dedup + ID assignment)
│ ├── enrichment.py # enrichment_orchestrator + route + execute + critique + merge
│ └── summarize.py # Final scan summary
├── access_review/
│ ├── graph.py # Graph 2 wiring — nodes, edges, Send fan-out
│ ├── run.py # Entry point with streaming output
│ └── nodes/
│ ├── review.py # review_orchestrator + route + execute
│ └── report.py # generate_certification_report
├── mcp_server/
│ ├── server.py # Main MCP server — stdio transport
│ ├── tools.py # 5 MCP tools (discovery, review, registry queries, update)
│ ├── resources.py # 4 MCP resources (registry, policies, directory, checklist)
│ ├── prompts.py # 3 MCP prompts (risk analysis, executive summary, remediation)
│ └── types.py # ToolDefinition, ResourceDefinition, PromptDefinition
├── evals/
│ ├── dataset.py # Test cases with ground truth
│ ├── evaluators.py # Custom evaluators (recall, accuracy, false positive rate)
│ └── run_eval.py # Evaluation harness
├── docs/
│ ├── architecture_journey.md # Design decisions and tradeoffs
│ ├── mcp_guide.md # MCP walkthrough with screenshots
│ ├── test_results.md # Enrichment test results log
│ └── original_blueprint.md # Original project blueprint (historical)
├── langgraph.json # LangGraph Platform config
└── requirements.txt用例
| 用例 | 发生地点 |
|---|---|
| 影子AI发现 | 图1，第1阶段——扫描仪发现未注册的代理 |
| 代理注册和入职 | 图1，第2阶段——唯一ID分配 |
| 所有权分配 | 图1，第3阶段——富集专家 |
| 访问认证 | 图2——根据政策进行完全访问审查 |
使用的LangGraph模式
| 模式 | 如何使用 |
|---|---|
| StateGraph+TypedDict | 具有显式字段行为的类型化状态模式 |
| Reducers（operator.add） | raw_findings、richment_updates、review_findings和approved_updates从并行代理中累积 |
| 发送API | 扫描程序、丰富和审查扇出-编排器决定生成多少个代理 |
| 动态编排 | LLM生成具有名称、角色、system_prompt、工具和代理批处理的AgentSpecs |
| 确定性+LLM | Python差距分析保证覆盖率，LLM增加了智能层 |
| 结构化工具调用 | report_discovered_agent、classify_agent、flag_policy_violation等。 |
| 每个字段评论循环 | 逐个字段验证丰富字段，只重试被拒绝的字段 |
| MCP客户端集成 | 扫描程序连接到外部MCP服务器以提取实时代理数据 |
| 条件边 | 如果所有代理都已完成，则跳过富集 |
| RetryPolicy | 所有LLM节点处理API瞬时故障，最多3次尝试 |
| 检查点 | 本地开发的MemorySaver，平台管理部署 |
设置
pip install -r requirements.txt# .env
ANTHROPIC_API_KEY=your-api-key
LANGCHAIN_API_KEY=your-langsmith-key
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=agent-identity-security
LLM_MODEL=claude-sonnet-4-20250514场景
来自MCP连接器的8个静态代理+2个活动代理+1个假阳性：
| 代理 | 来源 | 场景 |
|---|---|---|
| 人力资源福利助理 | AWS Bedrock | 干净、有管理、所有字段都已填写 |
| 客户流失预测器 | AWS Lambda | Shadow AI，带有OpenAI密钥的流氓函数，大部分为空 |
| 数据管道处理器 | AWS SageMaker | 有分类，缺少所有权 |
| pdf缩略图生成器 | AWS Lambda | 误报--实用程序脚本，而不是代理 |
| it帮助台副驾驶 | Azure副驾驶工作室 | 拥有所有权，缺少分类 |
| 合同摘要 | Azure Logic应用程序 | 影子AI，稀疏，无所有者 |
| 客户支持机器人 | GCP Vertex AI | 完全受控的间接访问风险 |
| 案例路由机器人 | Salesforce Einstein | 具有分类+所有权，缺少shadow_ai |
| 发票处理程序 | MCP连接器 | 字段最多，缺少risk_score和shadow_ai |
| 候选筛选器 | MCP连接器 | Shadow AI候选--8个字段中有7个为空 |
MCP服务器
该系统还作为MCP服务器通过Claude Desktop进行对话交互。
工具：
| 工具 | 它做什么 |
|---|---|
run_discovery | 运行完整的发现和富集管道 |
run_access_review | 对注册表运行访问权限审查 |
get_agent | 按ID查找特定代理 |
get_governance_status | 检查哪些代理存在治理差距 |
update_agent | 更新代理上的字段（所有者、risk_score等） |
资源： 代理注册表、安全策略、团队目录、治理清单
提示： 代理风险分析、执行摘要、补救计划
看 docs/mcp_guide.md 查看完整的截图演练。
MCP在行动
外部平台连接器
该系统还可以从外部MCP服务器中提取代理，模拟真实的平台连接器：
全面治理——10/10个代理
LangGraph工作室
看 docs/mcp_guide.md 完整的演练。