UFO 项目 AIP 消息协议完全参考Pydantic 消息模型、关联机制与最佳实践【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOAIPAgent Interaction Protocol是 UFO 仓库中连接 Constellation 编排器、设备 Agent 服务与设备客户端的统一通信协议其全部消息基于 Pydantic 模型定义天然具备自动校验、序列化与类型安全能力。本文以 documents/docs/aip/messages.md 为骨架结合 aip/messages.py 源码与协议实现系统讲解 AIP 的消息类型、核心数据结构、双向通信流程、标识符关联链以及消息验证方法帮助读者掌握在 UFO²/Galaxy 体系下构造、发送、接收与关联 AIP 消息的完整能力。协议背景为什么需要统一的消息模型AIP 作为 UFO 项目的神经系统将 ConstellationClient、设备 Agent 服务与设备客户端接入同一个事件驱动的控制平面。与短生命周期、无状态的 HTTP 协调方式如 A2A、ACP不同AIP 依赖持久化双向 WebSocket 会话因此消息在设计之初就需要满足强类型约束防止畸形消息进入协议管道架构 L1消息 Schema 层显式的请求/响应关联保证会话内命令的确定性顺序目标 G4结构化的状态机表达支持任务多轮执行与心跳健康检查易于扩展的元数据通道便于异构设备的能力发现目标 G2。这些诉求最终落地为 aip/messages.py 中基于 PydanticBaseModel定义的一组消息类所有消息通过 WebSocket 以 JSON 形式传输。双向通信与消息类型总览双向消息流AIP 是双向协议客户端与服务端都可以主动发起消息单向箭头表示请求-响应模式双向箭头--表示任一方均可主动发起。注意HEARTBEAT与TASK_END在不同场景下都可能双向流动客户端发送心跳保活服务端以OK状态心跳应答客户端上报任务完成服务端也向编排端下发任务结束通知。消息类型速查方向消息类型用途关键字段Client → ServerREGISTER初始能力通告client_id、metadataCOMMAND_RESULTS返回命令执行结果action_results、prev_response_idTASK_END通知任务完成status、session_idHEARTBEAT保活信号client_idServer → ClientTASK任务分配user_request、task_nameCOMMAND命令执行请求actions、response_idHEARTBEAT保活确认response_idTASK_END任务完成通知status、result双向DEVICE_INFO_REQUEST请求设备遥测request_idDEVICE_INFO_RESPONSE设备信息设备规格ERROR错误条件error在 aip/messages.py 中这些类型分别由ClientMessageType与ServerMessageType两个枚举承载。需要注意ClientMessageType与ServerMessageType都包含TASK、TASK_END、HEARTBEAT、ERROR、DEVICE_INFO_REQUEST、DEVICE_INFO_RESPONSE等成员但REGISTER与COMMAND_RESULTS仅存在于客户端枚举COMMAND仅存在于服务端枚举——这正反映了谁是发起方的语义约束。核心数据结构以下 Pydantic 模型是构建所有 AIP 消息的基础构件源码见 aip/messages.py。类型速查类型用途关键字段使用场景RectUI 元素坐标x、y、width、heightUI 自动化ControlInfoUI 控件元数据annotation_id、name、rectangle控件发现WindowInfo窗口元数据继承 ControlInfoprocess_id、is_active窗口管理MCPToolInfo工具定义tool_key、namespace、input_schema能力通告Command执行请求tool_name、parameters、call_id动作分发Result执行结果status、result、error结果上报RectUI 元素包围盒rect Rect(x100, y200, width300, height150)字段类型描述xint左上角 X 坐标yint左上角 Y 坐标widthint像素宽度heightint像素高度ControlInfoUI 控件元数据control ControlInfo( annotation_idctrl_001, nameSubmit Button, class_nameButton, rectangleRect(x100, y200, width80, height30), is_enabledTrue, is_visibleTrue )完整字段列表源码中全部为Optional默认None字段类型描述annotation_idstr?唯一标注标识符namestr?控件名称titlestr?控件标题handleint?Windows 句柄HWNDclass_namestr?UI 类名rectangleRect?包围矩形control_typestr?类型Button、TextBox 等automation_idstr?UI Automation IDis_enabledbool?启用状态is_visiblebool?可见状态sourcestr?数据来源标识符text_contentstr?文本内容此外aip/messages.py 还定义了AppWindowControlInfo它组合一个WindowInfo与可选的controls: List[ControlInfo]用于一次性描述窗口 其下所有控件的完整 UI 快照。WindowInfo窗口元数据WindowInfo继承自ControlInfo额外补充进程与窗口状态字段字段类型描述process_idint?进程 IDPIDprocess_namestr?进程名如 notepad.exeis_minimizedbool?最小化状态is_maximizedbool?最大化状态is_activebool?是否拥有焦点MCPToolInfoMCP 工具能力定义设备 Agent 在注册阶段使用MCPToolInfo通告自己的能力将 MCP 服务器上的工具暴露给编排端实现异构设备能力统一发现设计目标 G2tool_info MCPToolInfo( tool_keyui_automation.click_button, tool_nameclick_button, namespaceui_automation, tool_typeaction, descriptionClick a button by its ID, input_schema{ type: object, properties: { button_id: {type: string} } } )字段类型描述tool_keystr唯一键namespace.tool_nametool_namestr工具名称namespacestrMCP 命名空间tool_typestraction或data_collectiondescriptionstr?工具描述input_schemadict?输入 JSON Schemaoutput_schemadict?输出 JSON Schemametadict?元数据annotationsdict?附加注解在源码中还有一个与MCPToolInfo结构几乎对应的MCPToolCallaip/messages.py它额外携带parameters与mcp_server: BaseMCPServer实例引用通过ConfigDict(arbitrary_types_allowedTrue)允许非 Pydantic 类型并提供tool_info属性将自身转换为MCPToolInfo。从源码结构看MCPToolCall用于客户端内部的工具调用上下文而MCPToolInfo用于跨网络的能力通告。MCP 工具的详细接入方式见 MCP 集成指南。Command 与 Result执行请求与结果上报Command发送给设备 Agent 的执行请求cmd Command( tool_nameclick_element, parameters{control_id: btn_submit}, tool_typeaction, call_idcmd_12345 )字段类型必填描述tool_namestr✅要执行的工具名parametersdict工具参数tool_typestr✅data_collection或actioncall_idstr用于关联的唯一标识符在 aip/messages.py 中tool_type被建模为Literal[data_collection, action]即 Pydantic 会在反序列化阶段直接拒绝非法的tool_type值这是Schema 层早期错误检测的直接体现。用call_id将命令与其在Result对象中的结果配对是命令级关联的核心手法。ResultStatus执行结果枚举状态含义使用时机SUCCESS✅ 成功完成命令无错误执行FAILURE❌ 带错误失败执行遇到错误SKIPPED⏭️ 跳过执行条件执行、未运行NONE⚪ 无状态初始/未知状态Result命令执行结果⚠️ 先查 Status 再取结果访问result前必须检查status。若为FAILURE应使用error字段进行诊断。# 成功结果 result Result( statusResultStatus.SUCCESS, result{element_found: True, clicked: True}, namespaceui_automation, call_idcmd_12345 ) # 失败结果 result Result( statusResultStatus.FAILURE, errorElement not found: btn_submit, namespaceui_automation, call_idcmd_12345 )字段类型描述statusResultStatus执行状态errorstr?错误消息FAILURE 时resultAny结果负载类型随工具而异namespacestr?所执行工具的命名空间call_idstr?与Command.call_id匹配状态枚举与客户端类型TaskStatus任务生命周期状态CONTINUE → CONTINUE自环代表多轮执行——任务在完成前可以持续请求更多命令COMPLETED与FAILED是终态。状态含义用途CONTINUE 任务进行中多轮执行仍需更多步骤COMPLETED✅ 任务完成成功完成FAILED❌ 任务失败遇到错误OK✓ 确认心跳、健康检查通过ERROR⚠️ 协议错误协议级错误源码中TaskStatus为str, Enum双继承aip/messages.py枚举值分别是continue、completed、failed、ok、error这使得消息的 JSON 表示同时可读且可机器校验。ClientType客户端身份类型角色特征DEVICE设备 Agent 执行器本地执行任务、上报遥测、单设备聚焦CONSTELLATION多设备编排器管理多设备、协调任务、需要target_id# 设备客户端 device_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.DEVICE, client_iddevice_001 ) # Constellation 客户端 constellation_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.CONSTELLATION, client_idorchestrator_001, target_iddevice_001 # 目标设备 )ClientMessage客户端 → 服务端设备与 Constellation 客户端都通过ClientMessage与服务端通信。完整的构造逻辑可参考 RegistrationProtocol设备注册与 TaskExecutionProtocol任务请求中自动生成timestamp、request_id并填充statusTaskStatus.CONTINUE的方式。消息类型类型用途必填字段REGISTER初始注册client_id、client_typeHEARTBEAT保活client_id、statusOKTASK请求任务执行request、client_idTASK_END通知完成session_id、statusCOMMAND_RESULTS返回结果action_results、prev_response_idDEVICE_INFO_REQUEST请求遥测request_idDEVICE_INFO_RESPONSE提供遥测设备数据ERROR报告错误error公共字段字段类型描述typeClientMessageType消息类型statusTaskStatus当前任务状态client_typeClientTypeDEVICE 或 CONSTELLATIONsession_idstr?会话标识符task_namestr?可读任务名client_idstr?唯一客户端标识符target_idstr?目标设备constellation 使用requeststr?请求文本TASK 用action_resultsList[Result]?命令结果timestampstr?ISO 8601 时间戳request_idstr?唯一请求标识符prev_response_idstr?前一条响应 IDerrorstr?错误消息metadatadict?附加元数据示例REGISTERregister_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.DEVICE, client_idwindows_agent_001, statusTaskStatus.OK, timestamp2024-11-04T10:30:00Z, metadata{ platform: windows, os_version: Windows 11, capabilities: [ui_automation, file_operations] } )在注册流程中metadata是能力通告的核心载体。RegistrationProtocol.register_as_device()aip/protocol/registration.py会自动补充platform与registration_time字段register_as_constellation()则会额外写入type: constellation_client与targeted_device_id体现 Constellation 对目标设备的绑定关系。示例COMMAND_RESULTSresults_msg ClientMessage( typeClientMessageType.COMMAND_RESULTS, client_idwindows_agent_001, session_idsession_123, prev_response_idresp_456, # 关联服务端的 COMMAND 消息 statusTaskStatus.CONTINUE, action_results[ Result(statusResultStatus.SUCCESS, result{clicked: True}), Result(statusResultStatus.SUCCESS, result{text_entered: True}) ], timestamp2024-11-04T10:31:00Z, request_idreq_789 )服务端对COMMAND_RESULTS的处理可在 ufo/server/ws/handler.py 中看到它依据prev_response_id定位response_id并通过session_manager将结果绑定到当前连接注册的设备身份只查找、绝不从该路径创建会话从而防止攻击者用伪造的session_id向他人会话注入结果会话注入与幽灵会话拒绝服务防护。ServerMessage服务端 → 客户端设备服务通过ServerMessage向客户端分配任务、下发命令。服务端构造命令、任务分配与结束通知的便捷方法见 TaskExecutionProtocolsend_task_assignment、send_commands、send_task_end等其中每个方法都自动补齐timestamp与response_id。消息类型类型用途必填字段TASK分配任务user_request、task_name、session_idCOMMAND执行命令actions、response_id、session_idTASK_END通知完成status、session_idHEARTBEAT保活确认response_idDEVICE_INFO_REQUEST请求遥测request_idDEVICE_INFO_RESPONSE遥测数据设备信息ERROR错误通知error公共字段字段类型描述typeServerMessageType消息类型statusTaskStatus当前任务状态user_requeststr?原始用户请求agent_namestr?处理任务的 Agentprocess_namestr?执行上下文进程root_namestr?根应用名actionsList[Command]?要执行的命令messagesList[str]?日志消息errorstr?错误描述session_idstr?会话标识符task_namestr?任务名timestampstr?ISO 8601 时间戳response_idstr?响应标识符resultAny?结果负载示例TASK 任务分配task_msg ServerMessage( typeServerMessageType.TASK, statusTaskStatus.CONTINUE, user_requestOpen Notepad and create a new file, task_namecreate_notepad_file, session_idsession_123, response_idresp_001, agent_nameAppAgent, process_namenotepad.exe, timestamp2024-11-04T10:30:00Z )示例COMMAND 命令执行command_msg ServerMessage( typeServerMessageType.COMMAND, statusTaskStatus.CONTINUE, session_idsession_123, response_idresp_456, actions[ Command( tool_namelaunch_application, parameters{app_name: notepad}, tool_typeaction, call_idcmd_001 ), Command( tool_nametype_text, parameters{text: Hello World}, tool_typeaction, call_idcmd_002 ) ], timestamp2024-11-04T10:30:30Z )一个COMMAND消息可携带多条actions服务端按会话内顺序依次执行命令批处理减少网络往返对应确定性排序目标 G4。示例TASK_ENDtask_end_msg ServerMessage( typeServerMessageType.TASK_END, statusTaskStatus.COMPLETED, session_idsession_123, response_idresp_999, result{ file_created: True, path: C:\\Users\\user\\document.txt }, timestamp2024-11-04T10:35:00Z )消息验证MessageValidator⚠️ 内置验证AIP 提供MessageValidator类保证消息完整性。处理任何消息前都应先验证避免协议错误。验证方法方法用途要求validate_registration()校验注册typeREGISTER、client_id存在validate_task_request()校验任务请求typeTASK、request与client_id存在validate_command_results()校验结果typeCOMMAND_RESULTS、prev_response_id存在validate_server_message()校验服务端消息type与status存在from aip.messages import MessageValidator # 校验注册 if MessageValidator.validate_registration(client_message): await process_registration(client_message) # 校验任务请求 if MessageValidator.validate_task_request(client_message): await dispatch_task(client_message) # 校验命令结果 if MessageValidator.validate_command_results(client_message): await process_results(client_message)从源码看aip/messages.pyvalidate_registration要求type REGISTER且client_id非空validate_command_results要求prev_response_id与action_results均存在validate_server_message对COMMAND类型额外要求actions与response_id非空。这一层校验是 Pydantic Schema 校验之上的协议语义校验两者互补。对应测试见 tests/aip/test_messages.py例如缺少client_id的注册消息应返回False缺少request的 TASK 消息应返回False而结构完整的消息应返回True。这些用例同时验证了消息的 JSON 序列化model_dump_json与反序列化model_validate_json往返一致性见 tests/aip/test_messages.py。消息关联标识符链与会话跟踪AIP 通过一组标识符链在多轮消息交换中维持会话上下文每条新请求通过prev_response_id指向前一条服务端响应形成一条可追踪的会话链。这为多轮对话提供了审计轨迹、调试依据与请求-响应关联能力。关联字段字段用途示例request_id唯一请求标识符req_abc123response_id唯一响应标识符resp_def456prev_response_id指向前一条响应resp_def456session_id分组相关消息session_xyzcall_id关联命令/结果cmd_001会话分组同一任务执行期间的所有消息共享同一个session_id便于全程追溯# 所有消息使用相同 session_id SESSION_ID session_abc123 task_msg.session_id SESSION_ID command_msg.session_id SESSION_ID results_msg.session_id SESSION_ID task_end_msg.session_id SESSION_ID服务端实际使用session_id时还会做身份绑定如 ufo/server/ws/handler.py 所示COMMAND_RESULTS必须由会话所派发命令的设备本人返回否则会被拒绝防止无关认证对等端伪造结果注入他人会话。最佳实践消息构造时间戳始终使用 ISO 8601 格式from datetime import datetime, timezone timestamp datetime.now(timezone.utc).isoformat()唯一 ID为关联生成 UUIDimport uuid request_id str(uuid.uuid4())协议实现也遵循这一约定TaskExecutionProtocol与RegistrationProtocol在构造消息时统一使用datetime.datetime.now(datetime.timezone.utc).isoformat()与str(uuid4())见 aip/protocol/task_execution.py。错误处理访问结果数据前先检查Result.status始终提供有意义的错误消息使用ResultStatus.FAILURE并填写描述性error字段。可扩展性使用metadata字段携带自定义数据而不破坏协议利用 Pydantic 校验获得类型安全始终用prev_response_id关联消息。序列化与协议管道AIPProtocol消息在 WebSocket 上的收发由 AIPProtocol 统一处理发送时依次经过出站中间件 →model_dump_json()序列化 →transport.send()接收时transport.receive()→model_validate_json()反序列化 → 逆序执行入站中间件。因此只要消息是 Pydantic 模型协议层即可自动完成序列化开发者无需手写 JSON 编解码。from aip.protocol.base import ProtocolMiddleware class AuditMiddleware(ProtocolMiddleware): async def process_outgoing(self, msg): log_to_audit_trail(msg) return msg async def process_incoming(self, msg): log_to_audit_trail(msg) return msg除标准文本消息外aip/protocol/base.py 还支持二进制消息与分块文件传输send_binary_message采用JSON 元数据文本帧 二进制数据帧两帧结构send_file以默认 1MB 分块发送大文件并在完成帧中附带 MD5 校验和用于完整性验证。对应元数据结构BinaryMetadata、FileTransferStart、FileTransferComplete、ChunkMetadata同样定义在 aip/messages.py。快速参考导入全部消息类型from aip.messages import ( ClientMessage, ServerMessage, ClientMessageType, ServerMessageType, ClientType, TaskStatus, Command, Result, ResultStatus, MessageValidator, )相关文档协议指南 —— 各专用协议如何构造与使用消息端点文档 —— 端点如何处理消息AIP 概览 —— 系统架构中的高层消息流传输层 —— 消息传输的 WebSocket 载体韧性机制 —— 消息重试与超时处理MCP 集成指南 —— MCP 工具如何与 AIP 消息集成总结AIP 消息体系以 Pydantic 强类型模型为基石通过ClientMessage/ServerMessage承载双向通信以request_id、response_id、prev_response_id、session_id、call_id五类标识符构成完整的关联链配合MessageValidator的协议语义校验与AIPProtocol的中间件管道在 UFO² 的 DAG 编排、设备 Agent 执行与 Constellation 调度之间建立了正确、可追溯、可扩展的通信契约。无论是接入新设备、实现自定义消息类型还是调试多轮任务本文给出的字段说明、构造示例与源码路径都可作为直接参考。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 UFO 项目 AIP 消息协议完全参考：Pydantic 消息模型、关联机制与最佳实践
UFO 项目 AIP 消息协议完全参考Pydantic 消息模型、关联机制与最佳实践【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOAIPAgent Interaction Protocol是 UFO 仓库中连接 Constellation 编排器、设备 Agent 服务与设备客户端的统一通信协议其全部消息基于 Pydantic 模型定义天然具备自动校验、序列化与类型安全能力。本文以 documents/docs/aip/messages.md 为骨架结合 aip/messages.py 源码与协议实现系统讲解 AIP 的消息类型、核心数据结构、双向通信流程、标识符关联链以及消息验证方法帮助读者掌握在 UFO²/Galaxy 体系下构造、发送、接收与关联 AIP 消息的完整能力。协议背景为什么需要统一的消息模型AIP 作为 UFO 项目的神经系统将 ConstellationClient、设备 Agent 服务与设备客户端接入同一个事件驱动的控制平面。与短生命周期、无状态的 HTTP 协调方式如 A2A、ACP不同AIP 依赖持久化双向 WebSocket 会话因此消息在设计之初就需要满足强类型约束防止畸形消息进入协议管道架构 L1消息 Schema 层显式的请求/响应关联保证会话内命令的确定性顺序目标 G4结构化的状态机表达支持任务多轮执行与心跳健康检查易于扩展的元数据通道便于异构设备的能力发现目标 G2。这些诉求最终落地为 aip/messages.py 中基于 PydanticBaseModel定义的一组消息类所有消息通过 WebSocket 以 JSON 形式传输。双向通信与消息类型总览双向消息流AIP 是双向协议客户端与服务端都可以主动发起消息单向箭头表示请求-响应模式双向箭头--表示任一方均可主动发起。注意HEARTBEAT与TASK_END在不同场景下都可能双向流动客户端发送心跳保活服务端以OK状态心跳应答客户端上报任务完成服务端也向编排端下发任务结束通知。消息类型速查方向消息类型用途关键字段Client → ServerREGISTER初始能力通告client_id、metadataCOMMAND_RESULTS返回命令执行结果action_results、prev_response_idTASK_END通知任务完成status、session_idHEARTBEAT保活信号client_idServer → ClientTASK任务分配user_request、task_nameCOMMAND命令执行请求actions、response_idHEARTBEAT保活确认response_idTASK_END任务完成通知status、result双向DEVICE_INFO_REQUEST请求设备遥测request_idDEVICE_INFO_RESPONSE设备信息设备规格ERROR错误条件error在 aip/messages.py 中这些类型分别由ClientMessageType与ServerMessageType两个枚举承载。需要注意ClientMessageType与ServerMessageType都包含TASK、TASK_END、HEARTBEAT、ERROR、DEVICE_INFO_REQUEST、DEVICE_INFO_RESPONSE等成员但REGISTER与COMMAND_RESULTS仅存在于客户端枚举COMMAND仅存在于服务端枚举——这正反映了谁是发起方的语义约束。核心数据结构以下 Pydantic 模型是构建所有 AIP 消息的基础构件源码见 aip/messages.py。类型速查类型用途关键字段使用场景RectUI 元素坐标x、y、width、heightUI 自动化ControlInfoUI 控件元数据annotation_id、name、rectangle控件发现WindowInfo窗口元数据继承 ControlInfoprocess_id、is_active窗口管理MCPToolInfo工具定义tool_key、namespace、input_schema能力通告Command执行请求tool_name、parameters、call_id动作分发Result执行结果status、result、error结果上报RectUI 元素包围盒rect Rect(x100, y200, width300, height150)字段类型描述xint左上角 X 坐标yint左上角 Y 坐标widthint像素宽度heightint像素高度ControlInfoUI 控件元数据control ControlInfo( annotation_idctrl_001, nameSubmit Button, class_nameButton, rectangleRect(x100, y200, width80, height30), is_enabledTrue, is_visibleTrue )完整字段列表源码中全部为Optional默认None字段类型描述annotation_idstr?唯一标注标识符namestr?控件名称titlestr?控件标题handleint?Windows 句柄HWNDclass_namestr?UI 类名rectangleRect?包围矩形control_typestr?类型Button、TextBox 等automation_idstr?UI Automation IDis_enabledbool?启用状态is_visiblebool?可见状态sourcestr?数据来源标识符text_contentstr?文本内容此外aip/messages.py 还定义了AppWindowControlInfo它组合一个WindowInfo与可选的controls: List[ControlInfo]用于一次性描述窗口 其下所有控件的完整 UI 快照。WindowInfo窗口元数据WindowInfo继承自ControlInfo额外补充进程与窗口状态字段字段类型描述process_idint?进程 IDPIDprocess_namestr?进程名如 notepad.exeis_minimizedbool?最小化状态is_maximizedbool?最大化状态is_activebool?是否拥有焦点MCPToolInfoMCP 工具能力定义设备 Agent 在注册阶段使用MCPToolInfo通告自己的能力将 MCP 服务器上的工具暴露给编排端实现异构设备能力统一发现设计目标 G2tool_info MCPToolInfo( tool_keyui_automation.click_button, tool_nameclick_button, namespaceui_automation, tool_typeaction, descriptionClick a button by its ID, input_schema{ type: object, properties: { button_id: {type: string} } } )字段类型描述tool_keystr唯一键namespace.tool_nametool_namestr工具名称namespacestrMCP 命名空间tool_typestraction或data_collectiondescriptionstr?工具描述input_schemadict?输入 JSON Schemaoutput_schemadict?输出 JSON Schemametadict?元数据annotationsdict?附加注解在源码中还有一个与MCPToolInfo结构几乎对应的MCPToolCallaip/messages.py它额外携带parameters与mcp_server: BaseMCPServer实例引用通过ConfigDict(arbitrary_types_allowedTrue)允许非 Pydantic 类型并提供tool_info属性将自身转换为MCPToolInfo。从源码结构看MCPToolCall用于客户端内部的工具调用上下文而MCPToolInfo用于跨网络的能力通告。MCP 工具的详细接入方式见 MCP 集成指南。Command 与 Result执行请求与结果上报Command发送给设备 Agent 的执行请求cmd Command( tool_nameclick_element, parameters{control_id: btn_submit}, tool_typeaction, call_idcmd_12345 )字段类型必填描述tool_namestr✅要执行的工具名parametersdict工具参数tool_typestr✅data_collection或actioncall_idstr用于关联的唯一标识符在 aip/messages.py 中tool_type被建模为Literal[data_collection, action]即 Pydantic 会在反序列化阶段直接拒绝非法的tool_type值这是Schema 层早期错误检测的直接体现。用call_id将命令与其在Result对象中的结果配对是命令级关联的核心手法。ResultStatus执行结果枚举状态含义使用时机SUCCESS✅ 成功完成命令无错误执行FAILURE❌ 带错误失败执行遇到错误SKIPPED⏭️ 跳过执行条件执行、未运行NONE⚪ 无状态初始/未知状态Result命令执行结果⚠️ 先查 Status 再取结果访问result前必须检查status。若为FAILURE应使用error字段进行诊断。# 成功结果 result Result( statusResultStatus.SUCCESS, result{element_found: True, clicked: True}, namespaceui_automation, call_idcmd_12345 ) # 失败结果 result Result( statusResultStatus.FAILURE, errorElement not found: btn_submit, namespaceui_automation, call_idcmd_12345 )字段类型描述statusResultStatus执行状态errorstr?错误消息FAILURE 时resultAny结果负载类型随工具而异namespacestr?所执行工具的命名空间call_idstr?与Command.call_id匹配状态枚举与客户端类型TaskStatus任务生命周期状态CONTINUE → CONTINUE自环代表多轮执行——任务在完成前可以持续请求更多命令COMPLETED与FAILED是终态。状态含义用途CONTINUE 任务进行中多轮执行仍需更多步骤COMPLETED✅ 任务完成成功完成FAILED❌ 任务失败遇到错误OK✓ 确认心跳、健康检查通过ERROR⚠️ 协议错误协议级错误源码中TaskStatus为str, Enum双继承aip/messages.py枚举值分别是continue、completed、failed、ok、error这使得消息的 JSON 表示同时可读且可机器校验。ClientType客户端身份类型角色特征DEVICE设备 Agent 执行器本地执行任务、上报遥测、单设备聚焦CONSTELLATION多设备编排器管理多设备、协调任务、需要target_id# 设备客户端 device_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.DEVICE, client_iddevice_001 ) # Constellation 客户端 constellation_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.CONSTELLATION, client_idorchestrator_001, target_iddevice_001 # 目标设备 )ClientMessage客户端 → 服务端设备与 Constellation 客户端都通过ClientMessage与服务端通信。完整的构造逻辑可参考 RegistrationProtocol设备注册与 TaskExecutionProtocol任务请求中自动生成timestamp、request_id并填充statusTaskStatus.CONTINUE的方式。消息类型类型用途必填字段REGISTER初始注册client_id、client_typeHEARTBEAT保活client_id、statusOKTASK请求任务执行request、client_idTASK_END通知完成session_id、statusCOMMAND_RESULTS返回结果action_results、prev_response_idDEVICE_INFO_REQUEST请求遥测request_idDEVICE_INFO_RESPONSE提供遥测设备数据ERROR报告错误error公共字段字段类型描述typeClientMessageType消息类型statusTaskStatus当前任务状态client_typeClientTypeDEVICE 或 CONSTELLATIONsession_idstr?会话标识符task_namestr?可读任务名client_idstr?唯一客户端标识符target_idstr?目标设备constellation 使用requeststr?请求文本TASK 用action_resultsList[Result]?命令结果timestampstr?ISO 8601 时间戳request_idstr?唯一请求标识符prev_response_idstr?前一条响应 IDerrorstr?错误消息metadatadict?附加元数据示例REGISTERregister_msg ClientMessage( typeClientMessageType.REGISTER, client_typeClientType.DEVICE, client_idwindows_agent_001, statusTaskStatus.OK, timestamp2024-11-04T10:30:00Z, metadata{ platform: windows, os_version: Windows 11, capabilities: [ui_automation, file_operations] } )在注册流程中metadata是能力通告的核心载体。RegistrationProtocol.register_as_device()aip/protocol/registration.py会自动补充platform与registration_time字段register_as_constellation()则会额外写入type: constellation_client与targeted_device_id体现 Constellation 对目标设备的绑定关系。示例COMMAND_RESULTSresults_msg ClientMessage( typeClientMessageType.COMMAND_RESULTS, client_idwindows_agent_001, session_idsession_123, prev_response_idresp_456, # 关联服务端的 COMMAND 消息 statusTaskStatus.CONTINUE, action_results[ Result(statusResultStatus.SUCCESS, result{clicked: True}), Result(statusResultStatus.SUCCESS, result{text_entered: True}) ], timestamp2024-11-04T10:31:00Z, request_idreq_789 )服务端对COMMAND_RESULTS的处理可在 ufo/server/ws/handler.py 中看到它依据prev_response_id定位response_id并通过session_manager将结果绑定到当前连接注册的设备身份只查找、绝不从该路径创建会话从而防止攻击者用伪造的session_id向他人会话注入结果会话注入与幽灵会话拒绝服务防护。ServerMessage服务端 → 客户端设备服务通过ServerMessage向客户端分配任务、下发命令。服务端构造命令、任务分配与结束通知的便捷方法见 TaskExecutionProtocolsend_task_assignment、send_commands、send_task_end等其中每个方法都自动补齐timestamp与response_id。消息类型类型用途必填字段TASK分配任务user_request、task_name、session_idCOMMAND执行命令actions、response_id、session_idTASK_END通知完成status、session_idHEARTBEAT保活确认response_idDEVICE_INFO_REQUEST请求遥测request_idDEVICE_INFO_RESPONSE遥测数据设备信息ERROR错误通知error公共字段字段类型描述typeServerMessageType消息类型statusTaskStatus当前任务状态user_requeststr?原始用户请求agent_namestr?处理任务的 Agentprocess_namestr?执行上下文进程root_namestr?根应用名actionsList[Command]?要执行的命令messagesList[str]?日志消息errorstr?错误描述session_idstr?会话标识符task_namestr?任务名timestampstr?ISO 8601 时间戳response_idstr?响应标识符resultAny?结果负载示例TASK 任务分配task_msg ServerMessage( typeServerMessageType.TASK, statusTaskStatus.CONTINUE, user_requestOpen Notepad and create a new file, task_namecreate_notepad_file, session_idsession_123, response_idresp_001, agent_nameAppAgent, process_namenotepad.exe, timestamp2024-11-04T10:30:00Z )示例COMMAND 命令执行command_msg ServerMessage( typeServerMessageType.COMMAND, statusTaskStatus.CONTINUE, session_idsession_123, response_idresp_456, actions[ Command( tool_namelaunch_application, parameters{app_name: notepad}, tool_typeaction, call_idcmd_001 ), Command( tool_nametype_text, parameters{text: Hello World}, tool_typeaction, call_idcmd_002 ) ], timestamp2024-11-04T10:30:30Z )一个COMMAND消息可携带多条actions服务端按会话内顺序依次执行命令批处理减少网络往返对应确定性排序目标 G4。示例TASK_ENDtask_end_msg ServerMessage( typeServerMessageType.TASK_END, statusTaskStatus.COMPLETED, session_idsession_123, response_idresp_999, result{ file_created: True, path: C:\\Users\\user\\document.txt }, timestamp2024-11-04T10:35:00Z )消息验证MessageValidator⚠️ 内置验证AIP 提供MessageValidator类保证消息完整性。处理任何消息前都应先验证避免协议错误。验证方法方法用途要求validate_registration()校验注册typeREGISTER、client_id存在validate_task_request()校验任务请求typeTASK、request与client_id存在validate_command_results()校验结果typeCOMMAND_RESULTS、prev_response_id存在validate_server_message()校验服务端消息type与status存在from aip.messages import MessageValidator # 校验注册 if MessageValidator.validate_registration(client_message): await process_registration(client_message) # 校验任务请求 if MessageValidator.validate_task_request(client_message): await dispatch_task(client_message) # 校验命令结果 if MessageValidator.validate_command_results(client_message): await process_results(client_message)从源码看aip/messages.pyvalidate_registration要求type REGISTER且client_id非空validate_command_results要求prev_response_id与action_results均存在validate_server_message对COMMAND类型额外要求actions与response_id非空。这一层校验是 Pydantic Schema 校验之上的协议语义校验两者互补。对应测试见 tests/aip/test_messages.py例如缺少client_id的注册消息应返回False缺少request的 TASK 消息应返回False而结构完整的消息应返回True。这些用例同时验证了消息的 JSON 序列化model_dump_json与反序列化model_validate_json往返一致性见 tests/aip/test_messages.py。消息关联标识符链与会话跟踪AIP 通过一组标识符链在多轮消息交换中维持会话上下文每条新请求通过prev_response_id指向前一条服务端响应形成一条可追踪的会话链。这为多轮对话提供了审计轨迹、调试依据与请求-响应关联能力。关联字段字段用途示例request_id唯一请求标识符req_abc123response_id唯一响应标识符resp_def456prev_response_id指向前一条响应resp_def456session_id分组相关消息session_xyzcall_id关联命令/结果cmd_001会话分组同一任务执行期间的所有消息共享同一个session_id便于全程追溯# 所有消息使用相同 session_id SESSION_ID session_abc123 task_msg.session_id SESSION_ID command_msg.session_id SESSION_ID results_msg.session_id SESSION_ID task_end_msg.session_id SESSION_ID服务端实际使用session_id时还会做身份绑定如 ufo/server/ws/handler.py 所示COMMAND_RESULTS必须由会话所派发命令的设备本人返回否则会被拒绝防止无关认证对等端伪造结果注入他人会话。最佳实践消息构造时间戳始终使用 ISO 8601 格式from datetime import datetime, timezone timestamp datetime.now(timezone.utc).isoformat()唯一 ID为关联生成 UUIDimport uuid request_id str(uuid.uuid4())协议实现也遵循这一约定TaskExecutionProtocol与RegistrationProtocol在构造消息时统一使用datetime.datetime.now(datetime.timezone.utc).isoformat()与str(uuid4())见 aip/protocol/task_execution.py。错误处理访问结果数据前先检查Result.status始终提供有意义的错误消息使用ResultStatus.FAILURE并填写描述性error字段。可扩展性使用metadata字段携带自定义数据而不破坏协议利用 Pydantic 校验获得类型安全始终用prev_response_id关联消息。序列化与协议管道AIPProtocol消息在 WebSocket 上的收发由 AIPProtocol 统一处理发送时依次经过出站中间件 →model_dump_json()序列化 →transport.send()接收时transport.receive()→model_validate_json()反序列化 → 逆序执行入站中间件。因此只要消息是 Pydantic 模型协议层即可自动完成序列化开发者无需手写 JSON 编解码。from aip.protocol.base import ProtocolMiddleware class AuditMiddleware(ProtocolMiddleware): async def process_outgoing(self, msg): log_to_audit_trail(msg) return msg async def process_incoming(self, msg): log_to_audit_trail(msg) return msg除标准文本消息外aip/protocol/base.py 还支持二进制消息与分块文件传输send_binary_message采用JSON 元数据文本帧 二进制数据帧两帧结构send_file以默认 1MB 分块发送大文件并在完成帧中附带 MD5 校验和用于完整性验证。对应元数据结构BinaryMetadata、FileTransferStart、FileTransferComplete、ChunkMetadata同样定义在 aip/messages.py。快速参考导入全部消息类型from aip.messages import ( ClientMessage, ServerMessage, ClientMessageType, ServerMessageType, ClientType, TaskStatus, Command, Result, ResultStatus, MessageValidator, )相关文档协议指南 —— 各专用协议如何构造与使用消息端点文档 —— 端点如何处理消息AIP 概览 —— 系统架构中的高层消息流传输层 —— 消息传输的 WebSocket 载体韧性机制 —— 消息重试与超时处理MCP 集成指南 —— MCP 工具如何与 AIP 消息集成总结AIP 消息体系以 Pydantic 强类型模型为基石通过ClientMessage/ServerMessage承载双向通信以request_id、response_id、prev_response_id、session_id、call_id五类标识符构成完整的关联链配合MessageValidator的协议语义校验与AIPProtocol的中间件管道在 UFO² 的 DAG 编排、设备 Agent 执行与 Constellation 调度之间建立了正确、可追溯、可扩展的通信契约。无论是接入新设备、实现自定义消息类型还是调试多轮任务本文给出的字段说明、构造示例与源码路径都可作为直接参考。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考