a2at-engine

独立工作流执行 SDK，支持宿主 Agent 执行编排中心工作流（PSOP），同时保留对 A2A 通信、A2A-T 扩展与路由决策的完全控制权。SDK 自包含，不依赖编排中心任何代码。

完整设计参见 DESIGN.md。本文档面向集成者，快速上手与接口说明。

设计原则

架构

共享传输层 + 两个门面，职责单一：

A2ATransport（共享通信层：httpx + 认证 + AgentCard 映射 + SSE 消费）
  ├── WorkflowEngineClient（工作流发送门面：Task-T 生成、Negotiation-T 自动循环、事件回调、ControlPoint/ExtensionCallback 装配）
  └── ExtensionSender（一次性预置门面：Authorization-T / Notification-T 发送）

决策层拆分为两个接口：

• ControlPoint — 流程决策（on_task / on_self_task / on_route / on_negotiation）
• ExtensionCallback — 被动响应 Agent 推送的 A2A-T 数据（on_authorization / on_notification）

flowchart TB
    subgraph User["用户（宿主 Agent）"]
        AC["AgentCards<br/>（注册中心或自定义来源）"]
        CP["ControlPoint<br/>流程决策"]
        ECB["ExtensionCallback<br/>授权/通知"]
    end
    subgraph SDK["SDK（自包含）"]
        TR["A2ATransport<br/>共享通信层"]
        WEC["WorkflowEngineClient<br/>工作流发送"]
        ES["ExtensionSender<br/>一次性预置"]
        WE["WorkflowExecutor<br/>DAG 遍历"]
    end
    subgraph Agents["远端 Agents"]
        A1["Agent A"]
        A2["Agent B"]
    end
    AC --> TR
    TR --> WEC
    TR --> ES
    WEC -->|send_message| A1
    WEC -->|send_message| A2
    ES -->|预置发送| A1
    WE -->|on_task/on_route| CP
    WEC -->|on_negotiation| CP
    WEC -->|on_authorization/on_notification| ECB

快速开始

import asyncio
from a2at_engine import (
    execute_psop, ControlPoint, ExtensionCallback, RouteDecision,
    RegistryClient, load_psop,
)


class MyControlPoint(ControlPoint):
    async def on_task(self, request, engine_client):
        # SDK 已组装完整消息（上下文 + 任务 + 语言提示），直接发送
        result = await engine_client.send_message(
            request.agent_name, request.message
        )
        return TaskResponse(success=bool(result.text), output=result.text)

    async def on_route(self, step_name, results, conditions):
        # conditions: List[JumpCondition]，每个含 .step 与 .condition
        # 用你的 LLM 或业务逻辑选一个分支
        return RouteDecision(next_step=conditions[0].step)


class MyExtCallback(ExtensionCallback):
    async def on_authorization(self, agent_name, auth_request):
        return True  # 批准

    async def on_notification(self, agent_name, notification):
        print(f"通知来自 {agent_name}: {notification}")


async def main():
    # 1. 获取 AgentCards（注册中心或自定义来源）
    registry = RegistryClient(url="https://127.0.0.1:5000", ssl_verify=False)
    agent_cards = await registry.fetch_agent_cards()

    # 2. 加载 PSOP 工作流
    workflow = await load_psop(
        base_url="https://127.0.0.1:5001",
        psop_id="your-psop-id",
        ssl_verify=False,
    )

    # 3. 执行：execute_psop 内部构建 A2ATransport + WorkflowEngineClient
    async for event in execute_psop(
        psop=workflow,
        agent_cards=agent_cards,
        control_point=MyControlPoint(),
        extension_callback=MyExtCallback(),
        a2at_env_path=".env",
        credentials_config="agent_credentials.json",
        runtime_intent="诊断 SPN 跨市故障",
        ssl_verify=False,
    ):
        print(f"[{event['type']}] {event['data']}")


if __name__ == "__main__":
    asyncio.run(main())

分层入口

大多数集成使用第 2 层。需要手动控制时使用第 1 层。仅做一次性预置发送时直接使用 ExtensionSender。

用户实现的接口

ControlPoint（流程决策）

ExtensionCallback（被动响应）

订阅结果（如 Agent 后续推送的恢复结论）通过 send_notification 的响应流返回，不经过 on_notification。后者仅在 Agent 在 send_message 响应中主动附带 Notification-T 时触发。

A2ATransport + 门面（第 0 层）

from a2at_engine import A2ATransport, WorkflowEngineClient, ExtensionSender

transport = A2ATransport(
    agent_cards=agent_cards,
    a2at_env_path=".env",
    credentials_config="agent_credentials.json",
    ssl_verify=False,
)

# 工作流发送门面
engine_client = WorkflowEngineClient(transport)
engine_client.set_extension_callback(MyExtCallback())

# 一次性预置门面（工作流开始前）
sender = ExtensionSender(transport)
await sender.send_authorization("agent_a", "授权诊断操作", "诊断 SPN 跨市故障")
await sender.send_notification("agent_a", "订阅恢复结果通知", "诊断 SPN 跨市故障")

两个门面共享同一个 transport，不重复 wire 代码。

A2A-T 扩展

ExtensionRegistry 自动注册 Task-T 与 Negotiation-T（工作流内处理器）；Authorization-T / Notification-T 是预置操作，不自动注册，其 handler 类保留供手动注册处理 Agent 内联推送的数据。

智能体认证配置

当 AgentCard 声明 securitySchemes 与 securityRequirements 时，SDK 自动通过登录接口获取 token 并将认证头附加到出站请求。创建 JSON 文件：

{
  "agent_a": {
    "bearerAuth": {
      "login_url": "https://127.0.0.1:8080/auth/login",
      "method": "POST",
      "content_type": "application/json",
      "request_fields": { "username": "user", "password": "pass" },
      "token_field": "access_token",
      "token_ttl": 3600,
      "auth_header": "Authorization",
      "auth_header_prefix": "Bearer "
    }
  }
}

智能体名称须与 AgentCard 的 name 一致；认证方案名须与 securitySchemes 键一致。也可直接传 dict：credentials_config=dict。参见 examples/agent_credentials.example.json。

文件结构

workflow-exec-engine/
├── README.md              # 本文档
├── README_en.md           # English
├── DESIGN.md              # 设计文档
├── DEVELOPER_GUIDE.md     # 开发者指南
├── pyproject.toml
├── examples/
│   ├── quickstart.py
│   └── execute_psop_demo.py
└── a2at_engine/
    ├── __init__.py         # 公共 API 导出
    ├── runner.py           # execute_psop 高层运行器
    ├── core/               # 核心执行
    │   ├── models.py       # 数据模型
    │   ├── context_builder.py
    │   └── executor.py     # WorkflowExecutor DAG 遍历
    ├── client/             # 通信层
    │   ├── a2a_transport.py     # A2ATransport 共享通信层
    │   ├── engine_client.py     # WorkflowEngineClient 工作流门面
    │   ├── extension_sender.py  # ExtensionSender 一次性门面
    │   ├── extension_handlers.py
    │   ├── extensions.py        # A2ATExtension 枚举
    │   ├── auth_manager.py
    │   ├── credential_service.py
    │   ├── ssl_context.py
    │   └── sse_normalization.py
    ├── control/            # 决策接口
    │   └── control_points.py    # ControlPoint + ExtensionCallback + EventType
    └── registry/           # 注册中心集成（可选）
        └── registry_client.py

许可证

Apache License 2.0