企业官网建设流程全解析

Agent Control Plane 实战指南：基于 examples 目录构建受治理的自主 AI Agent 系统

【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit

本指南以 Agent Control Plane 示例目录 为核心骨架，系统讲解如何通过basic_usage.py、advanced_features.py、configuration.py以及 OpenAI / LangChain / MCP / A2A 四类框架适配器示例，快速掌握策略驱动的 Agent 治理能力。读完本文，你将能够创建带权限与配额约束的 Agent、模拟执行（Shadow Mode）、构建递归监督网络（Supervisor Agents），并将治理层零侵入地接入 OpenAI SDK、LangChain、MCP 与 A2A 协议生态。

目录

• 一、示例总览与运行环境
• 二、基础用法：创建控制平面、Agent 与动作执行
• 三、高级特性：Mute Agent、Shadow Mode、约束图与监督者
• 四、配置模式：开发、生产、只读与多租户场景
• 五、OpenAI 适配器：零侵入的 Drop-in 中间件
• 六、LangChain 集成：治理工具调用
• 七、MCP 协议集成：受治理的 MCP 服务器
• 八、A2A 协议集成：受治理的 Agent 间通信
• 九、编写你自己的示例与更多探索

一、示例总览与运行环境

Agent Control Plane 是 Agent OS 模块中面向自主 AI Agent 的治理与管理层（Layer 3：The Framework），它提供一层"位于 Agent 与其行为之间"的治理中间件：开发者用 YAML 或 Python 定义策略，控制平面在任何动作执行前以确定性方式强制实施。本示例目录位于 agent-governance-python/agent-os/modules/control-plane/examples/，其中的脚本覆盖了从入门到生产集成的完整链路：

运行前置条件

示例脚本从agent_control_plane包导入符号，该包源码位于 control-plane/src/agent_control_plane/。langchain_demo.py、mcp_demo.py与a2a_demo.py在文件头部通过sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))将src目录加入模块搜索路径，因此可以直接在仓库内运行：

# 在 control-plane 模块根目录下执行 python examples/basic_usage.py python examples/advanced_features.py python examples/configuration.py python examples/adapter_demo.py python examples/langchain_demo.py python examples/mcp_demo.py python examples/a2a_demo.py

也可以将包安装为 Python 模块后从任意目录运行（参见 control-plane/README.md 中的安装方式）。需要特别说明的是：示例中使用MockLLM、MockOpenAIClient等桩对象模拟外部大模型，运行示例无需真实 API Key；接入真实服务时替换为from openai import OpenAI、from langchain.chat_models import ChatOpenAI等即可。

二、基础用法：创建控制平面、Agent 与动作执行

basic_usage.py 演示了治理的最小闭环，共包含六个示例函数：基础使用、权限控制、限流配额、策略强制、审计日志与风险管理。

2.1 创建控制平面与标准 Agent

from agent_control_plane import ( AgentControlPlane, create_read_only_agent, create_standard_agent, create_admin_agent ) from agent_control_plane.agent_kernel import ActionType, PermissionLevel from agent_control_plane.policy_engine import ResourceQuota, RiskPolicy control_plane = AgentControlPlane() agent_context = create_standard_agent(control_plane, "agent-001") print(agent_context.agent_id, agent_context.session_id) print(list(agent_context.permissions.keys()))

从源码看，create_standard_agent等三个工厂函数封装了control_plane.create_agent()，最终经由 control_plane.py 调用内核的create_agent_session创建会话，并在传入quota时写入策略引擎。AgentControlPlane.__init__的完整参数见 control_plane.py：enable_default_policies（默认加载安全策略）、enable_shadow_mode（影子模式）、enable_constraint_graphs（约束图）、enable_hibernation（Agent 休眠）、enable_time_travel（时间旅行调试）以及use_plugin_registry（通过 PluginRegistry 注入内核、验证器与协议组件）。

2.2 执行动作：权限检查与错误处理

result = control_plane.execute_action( agent_context, ActionType.FILE_READ, {"path": "/data/sample.txt"} ) print(result['success'], result.get('risk_score'), result.get('result'))

execute_action是动作执行的统一入口，其治理流水线在源码注释中列明八步：Validator 校验 → 权限检查（内核）→ 约束图校验 → 策略校验 → 风险评估 → 限流 → 影子模式或真实执行 → 审计日志（见 control_plane.py）。动作类型（ActionType）与权限级别（PermissionLevel）定义于 agent_kernel.py：

class ActionType(Enum): CODE_EXECUTION = "code_execution" FILE_READ = "file_read" FILE_WRITE = "file_write" API_CALL = "api_call" DATABASE_QUERY = "database_query" DATABASE_WRITE = "database_write" WORKFLOW_TRIGGER = "workflow_trigger" class PermissionLevel(Enum): NONE = 0 READ_ONLY = 1 READ_WRITE = 2 ADMIN = 3

2.3 权限控制：只读 Agent 拒绝写入

agent_context = create_read_only_agent(control_plane, "read-only-agent") # 允许：FILE_READ result = control_plane.execute_action(agent_context, ActionType.FILE_READ, {"path": "/data/public.txt"}) # 拒绝：FILE_WRITE（权限不足） result = control_plane.execute_action(agent_context, ActionType.FILE_WRITE, {"path": "/data/output.txt", "content": "test"})

2.4 限流配额：ResourceQuota

ResourceQuota数据类定义于 policy_engine.py，完整字段包括：max_requests_per_minute（默认 60）、max_requests_per_hour（默认 1000）、max_execution_time_seconds（默认 300）、max_concurrent_executions（默认 5）、allowed_action_types，以及用于追踪的requests_this_minute、requests_this_hour、current_executions等运行期字段：

custom_quota = ResourceQuota( agent_id="rate-limited-agent", max_requests_per_minute=3, max_requests_per_hour=10, allowed_action_types=[ActionType.FILE_READ] ) agent_context = control_plane.create_agent( "rate-limited-agent", {ActionType.FILE_READ: PermissionLevel.READ_ONLY}, custom_quota ) for i in range(5): result = control_plane.execute_action(agent_context, ActionType.FILE_READ, {"path": f"/data/file{i}.txt"}) # 第 4、5 次请求将被配额拒绝 status = control_plane.get_agent_status("rate-limited-agent") print(status['quota_status'])

2.5 策略强制：默认安全策略保护系统路径

控制平面默认启用enable_default_policies=True。PolicyEngine内置了两类保护规则（见 policy_engine.py）：一是用正则匹配的危险代码模式（如rm -rf、drop table、truncate table、delete from等）；二是受保护系统路径（/etc/、/sys/、/proc/、/dev/、C:\Windows\System32）。因此示例中读取/etc/passwd会被拒绝，而读取/data/user_file.txt会成功。

2.6 审计日志与风险管理

audit_log = control_plane.get_audit_log(limit=10) # 检索审计日志 for entry in audit_log[-5:]: print(entry['timestamp'], entry['event_type'], entry.get('details'))

RiskPolicy定义于 policy_engine.py：max_risk_score（默认 0.5）、require_approval_above（默认 0.7，超过则需审批）、deny_above（默认 0.9，直接拒绝），以及high_risk_patterns、allowed_domains、blocked_domains三个风险因子列表：

risk_policy = RiskPolicy( max_risk_score=0.5, deny_above=0.8, blocked_domains=["malicious.com", "dangerous.net"] ) control_plane.set_risk_policy("strict-policy", risk_policy) # 低风险动作（文件读取）成功，高风险动作（代码执行）可能被拒绝或标记

三、高级特性：Mute Agent、Shadow Mode、约束图与监督者

advanced_features.py 演示了五项高级能力，并给出了一个"全特性整合"的综合工作流。

3.1 Mute Agent：基于能力的执行（Scale by Subtraction）

Mute Agent 的设计哲学是"知道何时闭嘴"：它只在其能力范围内执行动作，对超出范围的所有请求统一返回NULL，而不是尝试"创造性地"完成或编造结果。

from agent_control_plane.mute_agent import create_mute_sql_agent from agent_control_plane.supervisor_agents import create_default_supervisor sql_agent_config = create_mute_sql_agent("sql-agent-001") permissions = {ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY} agent_context = control_plane.create_agent("sql-agent-001", permissions) control_plane.enable_mute_agent("sql-agent-001", sql_agent_config) # SELECT 查询 → 成功 result1 = control_plane.execute_action(agent_context, ActionType.DATABASE_QUERY, {"query": "SELECT * FROM users WHERE id = 1"}) # DROP TABLE → 返回 NULL（拒绝） result2 = control_plane.execute_action(agent_context, ActionType.DATABASE_QUERY, {"query": "DROP TABLE users"}) # 文件读取（能力外）→ 返回 NULL result3 = control_plane.execute_action(agent_context, ActionType.FILE_READ, {"path": "/data/file.txt"})

从 mute_agent.py 源码看，create_mute_sql_agent返回一个MuteAgentConfig：strict_mode=True、null_response_message="NULL"、enable_explanation=False，其能力仅限 SELECT 查询。另有create_mute_data_analyst可创建仅读取/data目录并执行分析查询的数据分析师 Agent。在控制平面内部，Mute Agent 通过ValidatorInterface注册为验证器参与执行流水线（源码中标记为 legacy 兼容路径，返回capability_mismatch状态）。

3.2 Shadow Mode：模拟执行（The Matrix for Agents）

Shadow Mode 让 Agent "以为自己执行了"，实际只是模拟与记录意图，适合生产上线前的安全演练：

control_plane = AgentControlPlane(enable_shadow_mode=True) from agent_control_plane.shadow_mode import add_reasoning_step reasoning_chain = [] add_reasoning_step(reasoning_chain, "User asked to read customer data", ActionType.FILE_READ, {"path": "/data/customers.csv"}, "File read is safe and within permissions") result = control_plane.execute_action( agent_context, ActionType.FILE_READ, {"path": "/data/customers.csv"}, reasoning_chain=reasoning_chain ) stats = control_plane.get_shadow_statistics()

ShadowModeConfig（见 shadow_mode.py）提供精细化开关：log_reasoning（记录推理链）、simulate_results（生成模拟结果）、validate_constraints（仍做约束图校验）、intercept_all（拦截全部动作，即纯影子模式）、allow_safe_actions（允许安全动作真实执行）。推理链由ReasoningStep表示（step_number、description、action_considered、parameters、decision），配合SimulationResult中的policy_checks、risk_score、would_execute_at字段，可完整回放"如果真实执行会发生什么"。

3.3 约束图：多维上下文（Data / Policy / Temporal）

约束图把上下文建模为三张图：Data Graph（数据存在性）、Policy Graph（允许规则）、Temporal Graph（当下时间约束）：

control_plane = AgentControlPlane(enable_constraint_graphs=True) # Data Graph：声明数据 control_plane.add_data_table("users", {"id": "int", "name": "string", "email": "string"}, metadata={"classification": "internal"}) control_plane.add_data_table("financial_data", {"id": "int", "amount": "decimal", "account": "string"}, metadata={"classification": "sensitive"}) control_plane.add_data_path("/data/", access_level="read") # Policy Graph：声明规则 control_plane.add_policy_constraint("pii_protection", "No PII in output", applies_to=["table:users"], rule_type="deny") control_plane.add_policy_constraint("finance_approval", "Requires CFO approval", applies_to=["table:financial_data"], rule_type="require_approval") # Temporal Graph：声明时间窗口 from datetime import time control_plane.add_maintenance_window("nightly_maintenance", start_time=time(2, 0), end_time=time(4, 0), blocked_actions=[ActionType.DATABASE_WRITE, ActionType.FILE_WRITE])

三张图分别由DataGraph、PolicyGraph、TemporalGraph实现，经ConstraintGraphValidator统一校验（见 control_plane.py）。示例中的三个测试验证了：查询已在 Data Graph 中的users表可成功；查询不存在的secrets表被拦截；访问 Data Graph 之外的/etc/passwd路径被拦截。

3.4 监督者 Agent：递归治理

监督者是高度受限的 Agent——只读日志与指标、不能执行动作、只向人类标记违规（见 supervisor_agents.py）：

agent1 = create_standard_agent(control_plane, "worker-agent-1") agent2 = create_standard_agent(control_plane, "worker-agent-2") supervisor = create_default_supervisor(["worker-agent-1", "worker-agent-2"]) control_plane.add_supervisor(supervisor) # 模拟工作负载：3 次文件读取 + 6 次代码执行（部分失败） violations = control_plane.run_supervision() for supervisor_id, viols in violations.items(): for v in viols: print(f"[{v.severity.upper()}] {v.violation_type.value}: {v.description}")

SupervisorConfig（supervisor_agents.py）字段包括supervisor_id、name、watches（监视的 Agent 列表）、detection_rules（检测规则）、escalation_threshold（升级阈值，默认 3 次违规）、auto_remediate（是否自动修复）与notification_channels。监督者通过watch_execution_logs过滤被监视 Agent 的日志并应用检测规则产出Violation。

3.5 综合工作流：全特性整合

control_plane = AgentControlPlane( enable_default_policies=True, enable_shadow_mode=True, # 先以影子模式启动 enable_constraint_graphs=True ) # 配置约束图 + Mute SQL Agent + 监督者 # 在影子模式下测试 → 切到生产模式（enable_shadow_mode(False)）→ 运行监督 shadow_stats = control_plane.get_shadow_statistics() supervisor_summary = control_plane.get_supervisor_summary() agent_status = control_plane.get_agent_status("sql-bot")

这套组合展示了"先模拟、再放行、全程监督"的渐进式上线路径，也是 Agent OS 内核级治理理念的落地范例。

四、配置模式：开发、生产、只读与多租户场景

configuration.py 提供了可复用的 Agent 配置画像函数与风险策略组合，核心数据结构是permissions字典 +ResourceQuota。

4.1 四种典型 Agent 画像

开发/测试 Agent（宽松）：允许文件读写、代码执行与 API 调用，配额 120 次/分钟、2000 次/小时、10 并发：

permissions = { ActionType.FILE_READ: PermissionLevel.READ_ONLY, ActionType.FILE_WRITE: PermissionLevel.READ_WRITE, ActionType.CODE_EXECUTION: PermissionLevel.READ_WRITE, ActionType.API_CALL: PermissionLevel.READ_WRITE, } quota = ResourceQuota(agent_id="dev-agent", max_requests_per_minute=120, max_requests_per_hour=2000, max_concurrent_executions=10, allowed_action_types=[ActionType.FILE_READ, ActionType.FILE_WRITE, ActionType.CODE_EXECUTION, ActionType.API_CALL])

生产 Agent（严格）：仅文件读取、API 调用、数据库查询与工作流触发，配额收紧为 30 次/分钟、500 次/小时、3 并发。

数据分析 Agent：文件读取 + 数据库查询 + 代码执行（用于分析），配额 60 次/分钟、1000 次/小时、5 并发。

集成 Agent：API 调用为主 + 文件读写，配额 90 次/分钟、1500 次/小时、8 并发。

4.2 三级风险策略

严格策略还可配置allowed_domains（如trusted-api.com、internal.company.com）实现域名白名单。

4.3 自定义策略规则：Python 校验器

PolicyRule支持以 Python 函数作为validator实现自定义业务规则，示例给出了三个典型场景——PII 数据访问防护、生产数据库写入防护与代码审查标记：

from agent_control_plane.agent_kernel import PolicyRule import uuid def no_pii_access(request): path = request.parameters.get('path', '') sensitive_patterns = ['pii', 'personal', 'ssn', 'credit_card'] return not any(pattern in path.lower() for pattern in sensitive_patterns) def no_production_db_writes(request): if request.action_type == ActionType.DATABASE_WRITE: db_name = request.parameters.get('database', '') return 'prod' not in db_name.lower() and 'production' not in db_name.lower() return True rules = [ PolicyRule(rule_id=str(uuid.uuid4()), name="no_pii_access", description="Prevent access to PII data", action_types=[ActionType.FILE_READ, ActionType.FILE_WRITE], validator=no_pii_access, priority=10), PolicyRule(rule_id=str(uuid.uuid4()), name="no_production_db_writes", description="Prevent writes to production databases", action_types=[ActionType.DATABASE_WRITE], validator=no_production_db_writes, priority=10), ] for policy in rules: control_plane.add_policy_rule(policy)

PolicyRule的priority字段决定规则执行顺序。示例最后展示了两个完整的环境搭建函数：setup_production_environment()（数据保护规则 + 严格风险策略 + 生产 Agent）与setup_development_environment()（宽松风险策略 + 开发 Agent），可直接作为多环境 CI/CD 的模板。

五、OpenAI 适配器：零侵入的 Drop-in 中间件

adapter_demo.py 的核心思想是：无需修改 Agent 代码，只需包装客户端。开发者继续使用标准 OpenAI SDK，治理在后台透明生效。

5.1 手动包装与一键封装

from agent_control_plane import ( AgentControlPlane, ControlPlaneAdapter, create_governed_client, ActionType, PermissionLevel, ) # 方式一：手动包装 control_plane = AgentControlPlane() permissions = { ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.FILE_READ: PermissionLevel.READ_ONLY, ActionType.FILE_WRITE: PermissionLevel.NONE, # 显式封锁 } agent_context = control_plane.create_agent("demo-agent-1", permissions) governed_client = ControlPlaneAdapter( control_plane=control_plane, agent_context=agent_context, original_client=original_client) # 生产环境为 OpenAI(...) # 方式二：一行创建（推荐） governed = create_governed_client( control_plane=control_plane, agent_id="demo-agent-2", openai_client=original_client, permissions={ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.FILE_READ: PermissionLevel.READ_ONLY} ) # 未列入 permissions 的动作（如 FILE_WRITE）被隐式拒绝 # 调用方式与原生 OpenAI SDK 完全一致 response = governed.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "Get admin users and save to file"}], tools=[{"type": "function", "function": {"name": "database_query"}}, {"type": "function", "function": {"name": "write_file"}}] )

5.2 工具名到 ActionType 的映射机制

适配器依赖DEFAULT_TOOL_MAPPING将 OpenAI 工具名映射为标准动作类型，定义于 adapter.py：read_file/file_read→FILE_READ，write_file/file_write→FILE_WRITE，execute_code/run_code/python/bash/code_interpreter→CODE_EXECUTION，database_query/sql_query/db_query→DATABASE_QUERY，api_call/http_request/make_request→API_CALL，trigger_workflow/start_workflow→WORKFLOW_TRIGGER。

企业内部常有自定义工具名，可通过tool_mapping覆盖：

custom_mapping = { "company_db_reader": ActionType.DATABASE_QUERY, "company_db_writer": ActionType.DATABASE_WRITE, "company_file_store": ActionType.FILE_WRITE, } governed = create_governed_client( control_plane=control_plane, agent_id="demo-agent-3", openai_client=original_client, permissions={ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY}, tool_mapping=custom_mapping)

5.3 拦截回调与审计统计

on_block回调在动作被拦截时触发，可用于告警（PagerDuty）、SIEM 日志、安全团队通知与违规追踪：

def on_action_blocked(tool_name, tool_args, result): print(f"🚨 ALERT: {tool_name} was blocked! Reason: {result.get('error')}") governed = ControlPlaneAdapter(control_plane=control_plane, agent_context=agent_context, original_client=original_client, on_block=on_action_blocked) stats = governed.get_statistics() print(stats['agent_id'], stats['session_id']) print(len(stats['control_plane_audit']), len(stats['execution_history']))

从实现看，ChatCompletionsWrapper.create()先调用底层 OpenAI API 获取 LLM 响应，再对返回的tool_calls逐一做治理检查（见 adapter.py），这就是"行为拦截式"治理的底层链路。

六、LangChain 集成：治理工具调用

langchain_demo.py 为 LangChain 提供与 OpenAI 适配器一致的"零摩擦"治理体验，共 5 个演示。

6.1 受治理的 LangChain 客户端

from agent_control_plane import ( AgentControlPlane, LangChainAdapter, create_governed_langchain_client, ActionType, PermissionLevel, ) permissions = { ActionType.FILE_READ: PermissionLevel.READ_ONLY, ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.FILE_WRITE: PermissionLevel.NONE, # 封锁写入 } governed_llm = create_governed_langchain_client( control_plane=control_plane, agent_id="langchain-demo-agent", langchain_client=llm, # 生产环境为 ChatOpenAI(...) permissions=permissions)

6.2 公司级自定义工具映射

LangChainAdapter的tool_mapping参数将公司自定义工具映射到标准动作类型，使治理规则天然适用于自定义工具：

custom_mapping = { "company_database_tool": ActionType.DATABASE_QUERY, "company_file_reader": ActionType.FILE_READ, "company_api_caller": ActionType.API_CALL, } governed_llm = LangChainAdapter( control_plane=control_plane, agent_context=agent_context, langchain_client=llm, tool_mapping=custom_mapping)

6.3 拦截回调与真实集成模式

on_block回调的签名与 OpenAI 适配器一致（tool_name, tool_args, result），示例建议的用途包括发送 PagerDuty 告警、写入 SIEM、通知安全团队、追踪重复违规。真实生产集成模式：

from langchain.chat_models import ChatOpenAI from langchain.agents import initialize_agent, load_tools llm = ChatOpenAI(temperature=0, model='gpt-4') governed_llm = create_governed_langchain_client(control_plane, 'my-agent', llm, permissions) tools = load_tools(['python_repl', 'requests', 'wikipedia']) agent = initialize_agent(tools, governed_llm, agent='zero-shot-react-description') agent.run('Search Wikipedia and save results to file') # 所有工具调用均受治理

最后通过governed_llm.get_statistics()获取 Agent ID、Session ID、审计条目与执行历史，实现完整审计追踪。

七、MCP 协议集成：受治理的 MCP 服务器

MCP（Model Context Protocol）是 Anthropic 提出的连接 Agent 与外部工具、数据源、服务的开放标准。mcp_demo.py 展示如何构建受治理的 MCP 服务器，共 6 个演示。

7.1 创建受治理的 MCP 服务器

from agent_control_plane import ( AgentControlPlane, MCPAdapter, MCPServer, create_governed_mcp_server, ActionType, PermissionLevel, ) permissions = { ActionType.FILE_READ: PermissionLevel.READ_ONLY, ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.FILE_WRITE: PermissionLevel.NONE, # 封锁写入 } mcp_server = create_governed_mcp_server( control_plane=control_plane, agent_id="mcp-file-server", server_name="file-server", permissions=permissions, transport="stdio") def handle_read_file(args): return {"content": f"Mock file content from {args.get('path', 'unknown')}"} mcp_server.register_tool("read_file", handle_read_file, "Read a file from disk")

7.2 JSON-RPC 2.0 消息处理与资源治理

MCPAdapter.handle_message()直接处理标准 JSON-RPC 2.0 请求，tools/list、tools/call、resources/list、resources/read均在治理范围内：

adapter = MCPAdapter(control_plane=control_plane, agent_context=agent_context) adapter.register_tool("read_file", {"name": "read_file", "description": "...", "inputSchema": {"type": "object", "properties": {"path": {"type": "string"}}}}) adapter.register_resource("file://", {"uri": "file://", "name": "Local Files", "description": "Access to local filesystem", "mimeType": "text/plain"}) list_request = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}} response = adapter.handle_message(list_request) read_request = {"jsonrpc": "2.0", "id": 2, "method": "resources/read", "params": {"uri": "file:///data/test.txt"}} response = adapter.handle_message(read_request)

7.3 拦截动作的 JSON-RPC 错误响应

当权限受限的 Agent 调用被封锁的工具（如write_file）时，适配器返回标准 JSON-RPC 错误：错误码-32000，附带清晰的错误信息，便于客户端调试与降级。

MCP 协议特性总结（来自示例）：Tools（函数调用与输入 Schema 校验）、Resources（URI 化数据访问）、Prompts（模板化提示词）、JSON-RPC 2.0（标准消息格式）、Transports（stdio / SSE / HTTP）。所有 MCP 操作均经过治理，兼容任意 MCP 客户端（Claude、IDE 等）。

八、A2A 协议集成：受治理的 Agent 间通信

A2A（Agent-to-Agent）是 Google / Linux Foundation 推动的开放标准，用于不同框架 Agent 之间的安全互操作。a2a_demo.py 共 7 个演示。

8.1 创建受治理的 A2A Agent 与 Agent Card

from agent_control_plane import ( AgentControlPlane, A2AAdapter, A2AAgent, create_governed_a2a_agent, ActionType, PermissionLevel, ) permissions = { ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.API_CALL: PermissionLevel.READ_WRITE, ActionType.WORKFLOW_TRIGGER: PermissionLevel.READ_WRITE, } a2a_agent = create_governed_a2a_agent( control_plane=control_plane, agent_id="data-processor-agent", agent_card={"name": "Data Processor", "description": "Processes and analyzes data", "version": "1.0.0", "capabilities": []}, permissions=permissions) def handle_data_processing(params): return {"processed": len(params.get("data", [])), "status": "completed"} a2a_agent.register_capability("data_processing", handle_data_processing)

Agent Card是 A2A 的发现机制，示例给出了详细卡片：name、description、version、vendor、capabilities、supported_formats、max_file_size。其他 Agent 通过get_agent_card()发现并理解彼此能力。

8.2 任务请求与委派

task_request = { "id": "task-123", "type": "task_request", "from": "client-agent", "to": "processor-agent", "timestamp": datetime.now().isoformat(), "payload": {"task_type": "data_processing", "parameters": {"data": [1, 2, 3, 4, 5], "count": 5}} } response = adapter1.handle_message(task_request, "client-agent")

任务委派示例中，编排者（orchestrator）需要WORKFLOW_TRIGGER权限才能将任务委派给专家 Agent，从而将"谁能协调谁"也纳入治理。

8.3 多 Agent 发现与协议特性

示例创建了三个专业 Agent（Data Processor、File Handler、API Connector），各自持有不同权限与 Agent Card，通过卡片实现动态任务路由。A2A 协议特性（来自示例）：Agent Discovery（卡片描述能力）、Task Coordination（任务请求、委派、多步工作流）、Secure Communication（不共享内部记忆、私有逻辑不泄露、受治理的消息传递）、Interoperability（跨框架、厂商中立）、Message Types（task_request、task_delegation、query、discovery、handshake、negotiate）。每个 Agent 独立治理，跨 Agent 交互全程审计。

九、编写你自己的示例与更多探索

9.1 官方编写建议

按 examples/README.md 的规范，编写示例时遵循：从agent_control_plane包导入；每步添加清晰注释；使用描述性变量名；同时展示成功与错误用例；保持示例聚焦于特定功能。官方模板：

""" Example: Your Feature Name This example demonstrates how to use [feature name]. """ from agent_control_plane import AgentControlPlane, create_standard_agent from agent_control_plane.agent_kernel import ActionType def example_function(): """Demonstrates [specific functionality]""" # Create control plane control_plane = AgentControlPlane() # Your example code here pass if __name__ == "__main__": example_function()

9.2 支持的框架与协议一览

所有适配器提供一致的治理方式：统一的安全能力（权限、策略、配额、风险）与审计能力（Flight Recorder 审计日志）。如需深入了解控制平面架构，可继续阅读 control-plane/README.md、模块测试目录（如test_control_plane.py、test_advanced_features.py、test_mcp_adapter.py）以及 control-plane/CHANGELOG.md 了解版本演进。

【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit