一、不同协议底层逻辑关联分析

1. OPENAPI协议

OPENAPI 协议核心在于定义 API 的规范结构，它使用 YAML 或 JSON 格式来描述 API 的端点、请求参数、响应格式等信息。其底层逻辑是构建一个清晰、标准化的 API 描述文档，方便不同的客户端和服务端进行对接。它不直接参与通信过程，而是为通信双方提供了一个契约，规定了如何进行交互。

2. Selenium、Appium相关协议

Selenium 用于 Web 应用测试，Appium 用于移动应用测试。它们基于 WebDriver 协议，该协议定义了客户端（测试脚本）与服务端（浏览器驱动或移动设备驱动）之间的通信规则。客户端通过发送 HTTP 请求到服务端，服务端根据请求执行相应的操作，如打开页面、点击元素等，并将结果返回给客户端。其底层逻辑是基于 HTTP 协议的请求 - 响应模式。

3. Playwright协议

Playwright 同样采用客户端 - 服务端架构。它的底层逻辑与 Selenium 类似，也是通过 HTTP 协议进行通信。不过，Playwright 对不同浏览器的驱动进行了封装，提供了统一的 API，使得测试脚本可以更方便地在不同浏览器上运行。它的优势在于对浏览器的控制更加精细，能够处理更多的现代 Web 技术。

4. Pipeline上Selenium Grid协议

Selenium Grid 是 Selenium 的扩展，用于分布式测试。它的底层逻辑基于 Selenium WebDriver 协议，同时引入了中心节点（Hub）和多个节点（Node）的概念。Hub 作为服务端接收客户端的测试请求，然后根据节点的状态和配置将请求分配到合适的 Node 上执行。节点之间通过网络进行通信，实现测试任务的并行执行。

5. A2A协议

A2A 协议基于客户端 - 服务端架构，用于智能体之间的通信和协作。其底层逻辑可以基于多种网络协议，如 HTTP、WebSocket 等。智能体之间通过发送和接收消息来交换信息和协调任务。A2A 协议强调智能体的自主性和协作性，能够根据不同的场景和需求动态调整通信策略。

6. MCP协议

假设 MCP 协议通信底层逻辑基于 OPENAPI 协议。这意味着 MCP 协议在 OPENAPI 定义的 API 规范基础上，构建了自己的通信机制。它可能使用 OPENAPI 来描述服务的接口和数据格式，然后通过 HTTP 或其他协议进行实际的通信。MCP 协议可以在客户端和服务端之间传递测试任务、结果等信息，实现自动化测试的远程控制和管理。

底层逻辑关联总结

• Selenium、Appium 和 Playwright 都基于 WebDriver 协议，通过 HTTP 进行通信，主要用于 UI 自动化测试。
• Selenium Grid 是 Selenium 的扩展，在 WebDriver 协议基础上实现了分布式测试的任务分配和管理。
• A2A 协议为智能体之间的协作提供了通信机制，底层可基于多种网络协议。
• MCP 协议基于 OPENAPI 协议构建，利用 OPENAPI 的规范来定义通信接口，实现自动化测试的远程控制。

二、A2A协议和MCP协议对传统自动化测试流程的影响及未来流程设想

目前Pipeline流程

目前的自动化测试 Pipeline 流程通常包括以下几个步骤：

1. 代码提交：开发人员将代码提交到版本控制系统。
2. 触发构建：Pipeline 监测到代码变更后，触发构建任务。
3. 环境准备：创建测试环境，安装必要的依赖和软件。
4. 测试执行：使用 Selenium、Appium 等工具执行测试用例。
5. 结果报告：生成测试报告，展示测试结果。

未来流程设想

引入 A2A 协议和 MCP 协议后，未来的自动化测试 Pipeline 流程可能如下：

1. 智能体通信目的

智能体之间通信的目的主要是为了高效协作完成自动化测试任务。例如：

• 任务分配与协调：Master Agent 节点根据不同子 Agent 的能力、负载情况等信息，将测试任务合理分配给子 Agent。子 Agent 收到任务后，若发现自身资源不足或遇到无法处理的情况，可与 Master Agent 或其他子 Agent 通信，请求协助或重新分配任务。
• 信息共享：智能体之间可以共享测试数据、测试结果、环境信息等。比如，一个子 Agent 在测试过程中发现某个特定的异常情况，它可以将该信息及时共享给其他相关的子 Agent，避免重复错误，提高整体测试效率。
• 状态同步：确保各个智能体的状态一致，避免因状态不一致导致的测试冲突。例如，在分布式测试环境中，所有子 Agent 需要知道当前测试的整体进度和状态。

2. 通过 MCP 进行测试的方式

MCP 协议基于 OPENAPI 规范，在自动化测试中可以这样使用：

• 任务传递：Master Agent 节点将测试任务按照 MCP 协议定义的接口和数据格式，通过 HTTP 或其他协议发送给子 Agent。测试任务可以包含测试用例的详细信息，如测试目标、测试步骤、预期结果等。
• 执行控制：子 Agent 接收到测试任务后，根据 MCP 协议与测试目标（如浏览器、移动设备等）进行通信。例如，子 Agent 可以通过 MCP 协议发送请求，控制浏览器打开指定的页面、点击元素、输入数据等操作。
• 结果反馈：子 Agent 在执行测试任务过程中，将测试结果按照 MCP 协议的要求反馈给 Master Agent 节点。测试结果可以包括测试是否通过、错误信息、执行时间等。

3. 使用自然语言的可能性

目前像 Playwright 等工具已经开始支持一定程度的自然语言处理，未来 A2A 和 MCP 结合也有很大的使用自然语言的可能性。

• 任务下达：测试人员可以使用自然语言向 Master Agent 节点描述测试任务，如“测试登录页面的用户名和密码输入框是否正常工作”。Master Agent 节点通过自然语言处理技术解析任务，并将其转化为具体的测试用例，然后分配给子 Agent 执行。
• 结果解读：子 Agent 在反馈测试结果时，可以使用自然语言进行描述。例如，“登录页面输入错误密码后，提示信息显示不正确”，这样更便于测试人员理解和分析。

对比分析

• 智能化：引入 A2A 协议后，测试任务的分配和协调更加智能，能够根据不同的测试场景和资源情况进行动态调整。
• 远程控制：MCP 协议使得测试任务可以在远程环境中执行，提高了测试的灵活性和可扩展性。
• 协作性：A2A 协议促进了不同智能体之间的协作，使得测试过程更加高效和准确。

三、客户端和服务端A2A代码交互示例

通过以下示例，帮助我们更好的理解协议

client端

import asyncio
from uuid import uuid4
from common.client import A2AClient
from common.types import AgentCard
from common.resolver import A2ACardResolver

async def main():
    try:
        # 解析智能体卡片
        card_resolver = A2ACardResolver("http://localhost:10000")
        card = card_resolver.get_agent_card()

        print("======= Agent Card ========")
        print(card.model_dump_json(exclude_none=True))

        # 创建 A2A 客户端
        client = A2AClient(agent_card=card)

        # 构建请求负载
        payload = {
            "id": uuid4().hex,
            "sessionId": uuid4().hex,
            "acceptedOutputModes": ["text"],
            "message": {
                "role": "user",
                "parts": [
                    {
                        "type": "text",
                        "text": "执行登录测试用例",
                    }
                ],
            },
        }

        # 发送任务并获取结果
        ret = await client.send_task(payload=payload)
        print(ret.model_dump_json())
    except Exception as e:
        print(f"Error occurred: {str(e)}")


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

服务端

from common.server import A2AServer
from common.types import AgentCard, AgentCapabilities, AgentSkill, MissingAPIKeyError
from common.utils.push_notification_auth import PushNotificationSenderAuth
from agents.langgraph.task_manager import AgentTaskManager
from agents.langgraph.agent import CurrencyAgent
import click
import os
import logging
from dotenv import load_dotenv

# 日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 加载环境变量
load_dotenv()

@click.command()
@click.option("--host", "host", default="localhost")
@click.option("--port", "port", default=10000)
def main(host, port):
    """Starts the A2A Agent server."""
    try:
        # 定义智能体能力
        capabilities = AgentCapabilities(streaming=True, pushNotifications=True)

        # 定义技能
        skill1 = AgentSkill(
            id="skill1",
            name="测试用例执行技能",
            description="执行指定的测试用例",
            tags=["测试用例执行"],
            examples=["执行登录测试用例"]
        )

        # 创建智能体卡片
        agent_card = AgentCard(
            name="自动化测试智能体",
            description="执行自动化测试任务",
            url=f"http://{host}:{port}/",
            version="1.0.0",
            defaultInputModes=["text"],
            defaultOutputModes=["text"],
            capabilities=capabilities,
            skills=[skill1],
        )

        # 推送通知认证
        notification_sender_auth = PushNotificationSenderAuth()
        notification_sender_auth.generate_jwk()

        # 创建任务管理器
        task_manager = AgentTaskManager(agent=CurrencyAgent(), notification_sender_auth=notification_sender_auth)

        # 创建 A2A 服务器
        server = A2AServer(
            agent_card=agent_card,
            task_manager=task_manager,
            host=host,
            port=port,
        )

        # 添加 JWKS 端点
        server.app.add_route(
            "/.well-known/jwks.json", notification_sender_auth.handle_jwks_endpoint, methods=["GET"]
        )

        logger.info(f"Starting server on {host}:{port}")
        server.start()
    except MissingAPIKeyError as e:
        logger.error(f"Error: {e}")
        exit(1)
    except Exception as e:
        logger.error(f"An error occurred during server startup: {e}")
        exit(1)


if __name__ == "__main__":
    main()
    

代码说明

• 服务端代码（server.py）：创建了一个 A2A 服务器，定义了智能体的能力和技能，设置了推送通知认证，添加了 JWKS 端点，并启动服务器。
• 客户端代码（client.py）：解析智能体卡片，创建 A2A 客户端，构建请求负载并发送测试任务，最后打印服务端的响应结果。