Agent-to-Agent Protocol

from a2a.types import (
    AgentCard, AgentSkill, AgentCapabilities, AgentInterface,
)

AGENT_CARD = AgentCard(
    name="translator-agent",
    displayName="翻译助手",
    description="支持中英文互译的翻译 Agent",
    version="1.0.0",
    endpoints=[
        AgentInterface(
            url="http://localhost:9000/a2a",
            protocol="jsonrpc",
        )
    ],
    skills=[
        AgentSkill(
            id="translate",
            name="文本翻译",
            description="将中文翻译为英文，或将英文翻译为中文",
        )
    ],
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=False,
        multiTurn=False,
    ),
)

from a2a.server.agent_execution import AgentExecutor
from a2a.server.events import EventQueue
from a2a.types import TaskState, TaskStatus, Artifact, TextPart
import uuid

class TranslatorAgentExecutor(AgentExecutor):

    async def execute(self, context, event_queue: EventQueue):
        # 1. 提取用户消息
        user_message = context.get_user_message()
        input_text = user_message.parts[0].text

        # 2. 更新状态 → working
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                status=TaskStatus(state=TaskState.working),
            )
        )

        # 3. 执行翻译（实际中调用 LLM / API）
        translated = await self._translate(input_text)

        # 4. 生成 Artifact
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                taskId=context.task_id,
                artifact=Artifact(
                    id=str(uuid.uuid4()),
                    parts=[TextPart(text=translated)],
                ),
            )
        )

        # 5. 完成
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                status=TaskStatus(state=TaskState.completed),
            )
        )

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from agent_card import AGENT_CARD
from agent_executor import TranslatorAgentExecutor
import uvicorn

def main():
    agent_executor = TranslatorAgentExecutor()
    request_handler = DefaultRequestHandler(
        agent_executor=agent_executor,
        agent_card=AGENT_CARD,
    )
    app = A2AStarletteApplication(
        agent_card=AGENT_CARD,
        http_handler=request_handler,
    )
    uvicorn.run(app.build(), host="0.0.0.0", port=9000)

if __name__ == "__main__":
    main()

from a2a.client import A2AClient

async def main():
    # 自动获取 Agent Card
    client = await A2AClient.create(
        base_url="http://localhost:9000"
    )

    # 发送消息
    response = await client.send_message(
        message={
            "role": "user",
            "parts": [{"text": "Hello, how are you?"}]
        }
    )
    print(response)