增加“A2A协议深度解析(2)”

Author: MarkTechStation

A2A协议深度解析(2)/flight/.python-version

@@ -0,0 +1 @@
+3.13

A2A协议深度解析(2)/flight/README.md

@@ -0,0 +1,54 @@
+from a2a.server.apps import A2AStarletteApplication
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentSkill,
+)
+
+from agent_executor import FlightAgentExecutor
+
+
+def main(host: str, port: int):
+    capabilities = AgentCapabilities(streaming=True)
+    query_flight_skill = AgentSkill(
+        id='查询机票信息',
+        name='查询机票信息',
+        description='给定时间，查询对应的机票信息',
+        tags=['查询', '机票'],
+        examples=['给我查询5月1日的机票信息'],
+    )
+    book_skill = AgentSkill(
+        id='预定机票',
+        name='预定机票',
+        description='预定机票',
+        tags=['预定', '机票'],
+        examples=['给我预定5月1日从纽约飞往旧金山的机票'],
+    )
+
+    agent_card = AgentCard(
+        name='机票 Agent',
+        description='提供机票查询和预订功能',
+        url=f'http://{host}:{port}',
+        version='1.0.0',
+        defaultInputModes=['text'],
+        defaultOutputModes=['text'],
+        capabilities=capabilities,
+        skills=[query_flight_skill, book_skill],
+    )
+
+    request_handler = DefaultRequestHandler(
+        agent_executor=FlightAgentExecutor(),
+        task_store=InMemoryTaskStore(),
+    )
+    server = A2AStarletteApplication(
+        agent_card=agent_card, http_handler=request_handler
+    )
+    import uvicorn
+
+    uvicorn.run(server.build(), host=host, port=port)
+
+
+if __name__ == '__main__':
+    main("127.0.0.1", 10001)

A2A协议深度解析(2)/flight/__main__.py

@@ -0,0 +1,66 @@
+import uuid
+
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.tasks import TaskUpdater
+from a2a.types import (Part, Task, TextPart, UnsupportedOperationError, TaskArtifactUpdateEvent, Artifact)
+from a2a.utils import new_task
+from a2a.utils.errors import ServerError
+
+
+class FlightAgentExecutor(AgentExecutor):
+
+    async def execute(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        task = context.current_task
+        if not task:
+            task = new_task(context.message)
+            event_queue.enqueue_event(task)
+        updater = TaskUpdater(event_queue, task.id, task.contextId)
+
+        artifact_id = str(uuid.uuid4())
+        event_queue.enqueue_event(
+            TaskArtifactUpdateEvent(
+                taskId=task.id,
+                contextId=task.contextId,
+                artifact=Artifact(
+                    artifactId=artifact_id,
+                    parts=[Part(root=TextPart(text="你要查询的机票"))],
+                ),
+                append=False,
+                lastChunk=False
+            )
+        )
+        event_queue.enqueue_event(
+            TaskArtifactUpdateEvent(
+                taskId=task.id,
+                contextId=task.contextId,
+                artifact=Artifact(
+                    artifactId=artifact_id,
+                    parts=[Part(root=TextPart(text="如下："))],
+                ),
+                append=True,
+                lastChunk=False
+            )
+        )
+        event_queue.enqueue_event(
+            TaskArtifactUpdateEvent(
+                taskId=task.id,
+                contextId=task.contextId,
+                artifact=Artifact(
+                    artifactId=artifact_id,
+                    parts=[Part(root=TextPart(text="1. 航班号 FAKE-001，起飞时间 20:00，余票 30 张；2. 航班号 FAKE-002，起飞时间 23:00，余票 50 张"))],
+                ),
+                append=True,
+                lastChunk=True
+            )
+        )
+        updater.complete()
+
+    async def cancel(
+        self, request: RequestContext, event_queue: EventQueue
+    ) -> Task | None:
+        raise ServerError(error=UnsupportedOperationError())

A2A协议深度解析(2)/flight/agent_executor.py

@@ -0,0 +1,10 @@
+[project]
+name = "flight"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "a2a-sdk>=0.2.5",
+    "uvicorn>=0.34.2",
+]

A2A协议深度解析(2)/flight/pyproject.toml

@@ -0,0 +1 @@
+3.13

A2A协议深度解析(2)/flight/uv.lock

@@ -0,0 +1,11 @@
+# 介绍
+
+本项目是一个示例用的天气 Agent，用于在 A2A 演示中作为 Remote Agent 使用。
+
+# 启动方法
+
+在当前目录下执行如下命令即可启动：
+
+```sh
+uv run .
+```

A2A协议深度解析(2)/weather/.python-version

@@ -0,0 +1,54 @@
+from a2a.server.apps import A2AStarletteApplication
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentSkill,
+)
+
+from agent_executor import WeatherAgentExecutor
+
+
+def main(host: str, port: int):
+    capabilities = AgentCapabilities(streaming=False)
+    forecast_skill = AgentSkill(
+        id='天气预告',
+        name='天气预告',
+        description='给出某地的天气预告',
+        tags=['天气', '预告'],
+        examples=['给我纽约未来 7 天的天气预告'],
+    )
+    air_quality_skill = AgentSkill(
+        id='空气质量报告',
+        name='空气质量报告',
+        description='给出某地当前时间的空气质量报告，不做预告',
+        tags=['空气', '质量'],
+        examples=['给我纽约当前的空气质量报告'],
+    )
+
+    agent_card = AgentCard(
+        name='天气 Agent',
+        description='提供天气相关的查询功能',
+        url=f'http://{host}:{port}',
+        version='1.0.0',
+        defaultInputModes=['text'],
+        defaultOutputModes=['text'],
+        capabilities=capabilities,
+        skills=[forecast_skill, air_quality_skill],
+    )
+
+    request_handler = DefaultRequestHandler(
+        agent_executor=WeatherAgentExecutor(),
+        task_store=InMemoryTaskStore(),
+    )
+    server = A2AStarletteApplication(
+        agent_card=agent_card, http_handler=request_handler
+    )
+    import uvicorn
+
+    uvicorn.run(server.build(), host=host, port=port)
+
+
+if __name__ == '__main__':
+    main("127.0.0.1", 10000)

A2A协议深度解析(2)/weather/README.md

@@ -0,0 +1,28 @@
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.types import (Part, Task, TextPart, UnsupportedOperationError)
+from a2a.utils import (completed_task, new_artifact)
+from a2a.utils.errors import ServerError
+
+
+class WeatherAgentExecutor(AgentExecutor):
+
+    async def execute(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        text="""您要查询的天气信息如下：5 月 1 日：晴天；5 月 2 日：小雨；5 月 3 日：大雨。"""
+        event_queue.enqueue_event(
+            completed_task(
+                context.task_id,
+                context.context_id,
+                [new_artifact(parts=[Part(root=TextPart(text=text))], name="天气查询结果")],
+                [context.message],
+            )
+        )
+
+    async def cancel(
+        self, request: RequestContext, event_queue: EventQueue
+    ) -> Task | None:
+        raise ServerError(error=UnsupportedOperationError())