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 3The Framework它提供一层位于 Agent 与其行为之间的治理中间件开发者用 YAML 或 Python 定义策略控制平面在任何动作执行前以确定性方式强制实施。本示例目录位于 agent-governance-python/agent-os/modules/control-plane/examples/其中的脚本覆盖了从入门到生产集成的完整链路示例脚本核心主题说明basic_usage.py基础用法创建控制平面、按权限级别创建 Agent、执行动作、权限与错误处理advanced_features.py高级特性Mute Agent、Shadow Mode、约束图、监督者、推理遥测configuration.py配置模式开发/测试、生产、只读、多租户等 Agent 配置画像adapter_demo.pyOpenAI 适配器Drop-in 中间件、工具调用拦截、自定义工具映射langchain_demo.pyLangChain 集成基础适配、自定义工具映射、拦截回调、审计mcp_demo.pyMCP 协议集成受治理的 MCP 服务器、JSON-RPC 消息处理、资源注册a2a_demo.pyA2A 协议集成Agent 卡片、任务请求与委派、多 Agent 协调运行前置条件示例脚本从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 创建控制平面与标准 Agentfrom 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.pyenable_default_policies默认加载安全策略、enable_shadow_mode影子模式、enable_constraint_graphs约束图、enable_hibernationAgent 休眠、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.pyclass 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 32.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 限流配额ResourceQuotaResourceQuota数据类定义于 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_idrate-limited-agent, max_requests_per_minute3, max_requests_per_hour10, 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_policiesTrue。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(limit10) # 检索审计日志 for entry in audit_log[-5:]: print(entry[timestamp], entry[event_type], entry.get(details))RiskPolicy定义于 policy_engine.pymax_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_score0.5, deny_above0.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 SubtractionMute 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返回一个MuteAgentConfigstrict_modeTrue、null_response_messageNULL、enable_explanationFalse其能力仅限 SELECT 查询。另有create_mute_data_analyst可创建仅读取/data目录并执行分析查询的数据分析师 Agent。在控制平面内部Mute Agent 通过ValidatorInterface注册为验证器参与执行流水线源码中标记为 legacy 兼容路径返回capability_mismatch状态。3.2 Shadow Mode模拟执行The Matrix for AgentsShadow Mode 让 Agent 以为自己执行了实际只是模拟与记录意图适合生产上线前的安全演练control_plane AgentControlPlane(enable_shadow_modeTrue) 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_chainreasoning_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_graphsTrue) # 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_levelread) # Policy Graph声明规则 control_plane.add_policy_constraint(pii_protection, No PII in output, applies_to[table:users], rule_typedeny) control_plane.add_policy_constraint(finance_approval, Requires CFO approval, applies_to[table:financial_data], rule_typerequire_approval) # Temporal Graph声明时间窗口 from datetime import time control_plane.add_maintenance_window(nightly_maintenance, start_timetime(2, 0), end_timetime(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.pyagent1 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})SupervisorConfigsupervisor_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_policiesTrue, enable_shadow_modeTrue, # 先以影子模式启动 enable_constraint_graphsTrue ) # 配置约束图 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_iddev-agent, max_requests_per_minute120, max_requests_per_hour2000, max_concurrent_executions10, allowed_action_types[ActionType.FILE_READ, ActionType.FILE_WRITE, ActionType.CODE_EXECUTION, ActionType.API_CALL])生产 Agent严格仅文件读取、API 调用、数据库查询与工作流触发配额收紧为 30 次/分钟、500 次/小时、3 并发。数据分析 Agent文件读取 数据库查询 代码执行用于分析配额 60 次/分钟、1000 次/小时、5 并发。集成 AgentAPI 调用为主 文件读写配额 90 次/分钟、1500 次/小时、8 并发。4.2 三级风险策略策略max_risk_scorerequire_approval_abovedeny_aboveblocked_domains适用场景严格0.30.50.7malicious.com、untrusted.net、suspicious.org敏感环境均衡0.60.80.9malicious.com、dangerous.net通用环境宽松0.80.90.95空开发环境严格策略还可配置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_idstr(uuid.uuid4()), nameno_pii_access, descriptionPrevent access to PII data, action_types[ActionType.FILE_READ, ActionType.FILE_WRITE], validatorno_pii_access, priority10), PolicyRule(rule_idstr(uuid.uuid4()), nameno_production_db_writes, descriptionPrevent writes to production databases, action_types[ActionType.DATABASE_WRITE], validatorno_production_db_writes, priority10), ] 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_planecontrol_plane, agent_contextagent_context, original_clientoriginal_client) # 生产环境为 OpenAI(...) # 方式二一行创建推荐 governed create_governed_client( control_planecontrol_plane, agent_iddemo-agent-2, openai_clientoriginal_client, permissions{ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY, ActionType.FILE_READ: PermissionLevel.READ_ONLY} ) # 未列入 permissions 的动作如 FILE_WRITE被隐式拒绝 # 调用方式与原生 OpenAI SDK 完全一致 response governed.chat.completions.create( modelgpt-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.pyread_file/file_read→FILE_READwrite_file/file_write→FILE_WRITEexecute_code/run_code/python/bash/code_interpreter→CODE_EXECUTIONdatabase_query/sql_query/db_query→DATABASE_QUERYapi_call/http_request/make_request→API_CALLtrigger_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_planecontrol_plane, agent_iddemo-agent-3, openai_clientoriginal_client, permissions{ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY}, tool_mappingcustom_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_planecontrol_plane, agent_contextagent_context, original_clientoriginal_client, on_blockon_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_planecontrol_plane, agent_idlangchain-demo-agent, langchain_clientllm, # 生产环境为 ChatOpenAI(...) permissionspermissions)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_planecontrol_plane, agent_contextagent_context, langchain_clientllm, tool_mappingcustom_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(temperature0, modelgpt-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, agentzero-shot-react-description) agent.run(Search Wikipedia and save results to file) # 所有工具调用均受治理最后通过governed_llm.get_statistics()获取 Agent ID、Session ID、审计条目与执行历史实现完整审计追踪。七、MCP 协议集成受治理的 MCP 服务器MCPModel 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_planecontrol_plane, agent_idmcp-file-server, server_namefile-server, permissionspermissions, transportstdio) def handle_read_file(args): return {content: fMock 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_planecontrol_plane, agent_contextagent_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 校验、ResourcesURI 化数据访问、Prompts模板化提示词、JSON-RPC 2.0标准消息格式、Transportsstdio / SSE / HTTP。所有 MCP 操作均经过治理兼容任意 MCP 客户端Claude、IDE 等。八、A2A 协议集成受治理的 Agent 间通信A2AAgent-to-Agent是 Google / Linux Foundation 推动的开放标准用于不同框架 Agent 之间的安全互操作。a2a_demo.py 共 7 个演示。8.1 创建受治理的 A2A Agent 与 Agent Cardfrom 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_planecontrol_plane, agent_iddata-processor-agent, agent_card{name: Data Processor, description: Processes and analyzes data, version: 1.0.0, capabilities: []}, permissionspermissions) 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 发现与协议特性示例创建了三个专业 AgentData Processor、File Handler、API Connector各自持有不同权限与 Agent Card通过卡片实现动态任务路由。A2A 协议特性来自示例Agent Discovery卡片描述能力、Task Coordination任务请求、委派、多步工作流、Secure Communication不共享内部记忆、私有逻辑不泄露、受治理的消息传递、Interoperability跨框架、厂商中立、Message Typestask_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 支持的框架与协议一览框架/协议适配器入口特点OpenAI SDKControlPlaneAdapter/create_governed_clientDrop-in 中间件拦截工具调用LangChainLangChainAdapter/create_governed_langchain_client治理 Agent 与工具调用MCPMCPAdapter/MCPServer/create_governed_mcp_server工具/资源/提示词的协议级治理A2AA2AAdapter/A2AAgent/create_governed_a2a_agentAgent 间通信与任务协调治理所有适配器提供一致的治理方式统一的安全能力权限、策略、配额、风险与审计能力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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考