Refactor code base

Change-Id: Ib8754fe89591cde3a2e15a2927eac2dc62088e31

Author: kthota-g

examples/helloworld/README.md

@@ -0,0 +1,13 @@
+Hello World example agent that only returns Message events
+
+## Getting started
+
+1. Start the server
+    ```bash
+    uv run .
+    ```
+
+4. Run the test client
+    ```bash
+    uv run test_client.py
+    ```

examples/helloworld/__init__.py

@@ -1,6 +1,7 @@
 from agent_executor import HelloWorldAgentExecutor
 
-from a2a.server import A2AServer, DefaultA2ARequestHandler
+from a2a.server import A2AServer
+from a2a.server.request_handlers import DefaultA2ARequestHandler
 from a2a.types import (
     AgentAuthentication,
     AgentCapabilities,

examples/helloworld/__main__.py

@@ -4,24 +4,18 @@
 from typing import Any
 from uuid import uuid4
 
-from a2a.server import AgentExecutor
+from typing_extensions import override
+
+from a2a.server.agent_execution import BaseAgentExecutor
+from a2a.server.events import EventQueue
 from a2a.types import (
-    CancelTaskRequest,
-    CancelTaskResponse,
-    JSONRPCErrorResponse,
     Message,
     Part,
     Role,
     SendMessageRequest,
-    SendMessageResponse,
-    SendMessageStreamingRequest,
-    SendMessageStreamingResponse,
-    SendMessageStreamingSuccessResponse,
-    SendMessageSuccessResponse,
+    SendStreamingMessageRequest,
     Task,
-    TaskResubscriptionRequest,
     TextPart,
-    UnsupportedOperationError,
 )
 
 
@@ -37,57 +31,40 @@ async def stream(self) -> AsyncGenerator[dict[str, Any], None]:
         yield {'content': 'World', 'done': True}
 
 
-class HelloWorldAgentExecutor(AgentExecutor):
+class HelloWorldAgentExecutor(BaseAgentExecutor):
     """Test AgentProxy Implementation."""
 
     def __init__(self):
         self.agent = HelloWorldAgent()
 
+    @override
     async def on_message_send(
-        self, request: SendMessageRequest, task: Task | None
-    ) -> SendMessageResponse:
+        self,
+        request: SendMessageRequest,
+        event_queue: EventQueue,
+        task: Task | None,
+    ) -> None:
         result = await self.agent.invoke()
 
         message: Message = Message(
             role=Role.agent,
             parts=[Part(root=TextPart(text=result))],
             messageId=str(uuid4()),
         )
+        event_queue.enqueue_event(message)
 
-        return SendMessageResponse(
-            root=SendMessageSuccessResponse(id=request.id, result=message)
-        )
-
-    async def on_message_stream(  # type: ignore
-        self, request: SendMessageStreamingRequest, task: Task | None
-    ) -> AsyncGenerator[SendMessageStreamingResponse, None]:
+    @override
+    async def on_message_stream(
+        self,
+        request: SendStreamingMessageRequest,
+        event_queue: EventQueue,
+        task: Task | None,
+    ) -> None:
         async for chunk in self.agent.stream():
             message: Message = Message(
                 role=Role.agent,
                 parts=[Part(root=TextPart(text=chunk['content']))],
                 messageId=str(uuid4()),
                 final=chunk['done'],
             )
-            yield SendMessageStreamingResponse(
-                root=SendMessageStreamingSuccessResponse(
-                    id=request.id, result=message
-                )
-            )
-
-    async def on_cancel(
-        self, request: CancelTaskRequest, task: Task
-    ) -> CancelTaskResponse:
-        return CancelTaskResponse(
-            root=JSONRPCErrorResponse(
-                id=request.id, error=UnsupportedOperationError()
-            )
-        )
-
-    async def on_resubscribe(  # type: ignore
-        self, request: TaskResubscriptionRequest, task: Task
-    ) -> AsyncGenerator[SendMessageStreamingResponse, None]:
-        yield SendMessageStreamingResponse(
-            root=JSONRPCErrorResponse(
-                id=request.id, error=UnsupportedOperationError()
-            )
-        )
+            event_queue.enqueue_event(message)

examples/helloworld/agent_executor.py

@@ -0,0 +1,26 @@
+[project]
+name = "helloworld"
+version = "0.1.0"
+description = "HelloWorld agent example that only returns Messages"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "a2a",
+    "click>=8.1.8",
+    "dotenv>=0.9.9",
+    "httpx>=0.28.1",
+    "langchain-google-genai>=2.1.4",
+    "langgraph>=0.4.1",
+    "pydantic>=2.11.4",
+    "python-dotenv>=1.1.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["."]
+
+[tool.uv.sources]
+a2a = { workspace = true }
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

examples/helloworld/pyproject.toml

@@ -7,7 +7,9 @@
 from agent_executor import CurrencyAgentExecutor
 from dotenv import load_dotenv
 
-from a2a.server import A2AServer, DefaultA2ARequestHandler, InMemoryTaskStore
+
+from a2a.server.request_handlers import DefaultA2ARequestHandler
+from a2a.server import A2AServer
 from a2a.types import (
     AgentAuthentication,
     AgentCapabilities,
@@ -27,11 +29,8 @@ def main(host: str, port: int):
         print('GOOGLE_API_KEY environment variable not set.')
         sys.exit(1)
 
-    task_store = InMemoryTaskStore()
-
     request_handler = DefaultA2ARequestHandler(
-        agent_executor=CurrencyAgentExecutor(task_store=task_store),
-        task_store=task_store,
+        agent_executor=CurrencyAgentExecutor()
     )
 
     server = A2AServer(

examples/helloworld/uv.lock

@@ -1,71 +1,66 @@
-from collections.abc import AsyncGenerator
 from typing import Any
 
 from agent import CurrencyAgent
 from helpers import (
-    create_task_obj,
     process_streaming_agent_response,
     update_task_with_agent_response,
 )
+from typing_extensions import override
 
-from a2a.server import AgentExecutor, TaskStore
+from a2a.server.agent_execution import BaseAgentExecutor
+from a2a.server.events.event_queue import EventQueue
 from a2a.types import (
-    CancelTaskRequest,
-    CancelTaskResponse,
-    JSONRPCErrorResponse,
     MessageSendParams,
     SendMessageRequest,
-    SendMessageResponse,
-    SendMessageStreamingRequest,
-    SendMessageStreamingResponse,
-    SendMessageStreamingSuccessResponse,
-    SendMessageSuccessResponse,
+    SendStreamingMessageRequest,
     Task,
-    TaskNotCancelableError,
-    TaskResubscriptionRequest,
     TextPart,
-    UnsupportedOperationError,
 )
+from a2a.utils import create_task_obj
 
 
-class CurrencyAgentExecutor(AgentExecutor):
+class CurrencyAgentExecutor(BaseAgentExecutor):
     """Currency AgentExecutor Example."""
 
-    def __init__(self, task_store: TaskStore):
+    def __init__(self):
         self.agent = CurrencyAgent()
-        self.task_store = task_store
 
+    @override
     async def on_message_send(
-        self, request: SendMessageRequest, task: Task | None
-    ) -> SendMessageResponse:
+        self,
+        request: SendMessageRequest,
+        event_queue: EventQueue,
+        task: Task | None,
+    ) -> None:
         """Handler for 'message/send' requests."""
         params: MessageSendParams = request.params
         query = self._get_user_query(params)
 
         if not task:
             task = create_task_obj(params)
-            await self.task_store.save(task)
 
         # invoke the underlying agent
         agent_response: dict[str, Any] = self.agent.invoke(
             query, task.contextId
         )
-
         update_task_with_agent_response(task, agent_response)
-        return SendMessageResponse(
-            root=SendMessageSuccessResponse(id=request.id, result=task)
-        )
-
-    async def on_message_stream(  # type: ignore
-        self, request: SendMessageStreamingRequest, task: Task | None
-    ) -> AsyncGenerator[SendMessageStreamingResponse, None]:
+        event_queue.enqueue_event(task)
+
+    @override
+    async def on_message_stream(
+        self,
+        request: SendStreamingMessageRequest,
+        event_queue: EventQueue,
+        task: Task | None,
+    ) -> None:
         """Handler for 'message/sendStream' requests."""
         params: MessageSendParams = request.params
         query = self._get_user_query(params)
 
         if not task:
             task = create_task_obj(params)
-            await self.task_store.save(task)
+            # emit the initial task so it is persisted to TaskStore
+            event_queue.enqueue_event(task)
 
         # kickoff the streaming agent and process responses
         async for item in self.agent.stream(query, task.contextId):
@@ -74,37 +69,9 @@ async def on_message_stream(  # type: ignore
             )
 
             if task_artifact_update_event:
-                yield SendMessageStreamingResponse(
-                    root=SendMessageStreamingSuccessResponse(
-                        id=request.id, result=task_artifact_update_event
-                    )
-                )
-
-            yield SendMessageStreamingResponse(
-                root=SendMessageStreamingSuccessResponse(
-                    id=request.id, result=task_status_event
-                )
-            )
+                event_queue.enqueue_event(task_artifact_update_event)
 
-    async def on_cancel(
-        self, request: CancelTaskRequest, task: Task
-    ) -> CancelTaskResponse:
-        """Handler for 'tasks/cancel' requests."""
-        return CancelTaskResponse(
-            root=JSONRPCErrorResponse(
-                id=request.id, error=TaskNotCancelableError()
-            )
-        )
-
-    async def on_resubscribe(  # type: ignore
-        self, request: TaskResubscriptionRequest, task: Task
-    ) -> AsyncGenerator[SendMessageStreamingResponse, None]:
-        """Handler for 'tasks/resubscribe' requests."""
-        yield SendMessageStreamingResponse(
-            root=JSONRPCErrorResponse(
-                id=request.id, error=UnsupportedOperationError()
-            )
-        )
+            event_queue.enqueue_event(task_status_event)
 
     def _get_user_query(self, task_send_params: MessageSendParams) -> str:
         """Helper to get user query from task send params."""

examples/langgraph/__main__.py

@@ -5,7 +5,6 @@
 from a2a.types import (
     Artifact,
     Message,
-    MessageSendParams,
     Part,
     Role,
     Task,
@@ -17,19 +16,6 @@
 )
 
 
-def create_task_obj(message_send_params: MessageSendParams) -> Task:
-    """Create a new task object."""
-    if not message_send_params.message.contextId:
-        message_send_params.message.contextId = str(uuid4())
-
-    return Task(
-        id=str(uuid4()),
-        contextId=message_send_params.message.contextId,
-        status=TaskStatus(state=TaskState.submitted),
-        history=[message_send_params.message],
-    )
-
-
 def update_task_with_agent_response(
     task: Task, agent_response: dict[str, Any]
 ) -> None:
@@ -38,13 +24,19 @@ def update_task_with_agent_response(
     parts: list[Part] = [Part(root=TextPart(text=agent_response['content']))]
     if agent_response['require_user_input']:
         task.status.state = TaskState.input_required
-        task.status.message = Message(
+        message = Message(
             messageId=str(uuid4()),
             role=Role.agent,
             parts=parts,
         )
+        task.status.message = message
+        if not task.history:
+            task.history = []
+
+        task.history.append(message)
     else:
         task.status.state = TaskState.completed
+        task.status.message = None
         if not task.artifacts:
             task.artifacts = []
 
@@ -83,13 +75,15 @@ def process_streaming_agent_response(
     if artifact:
         task_artifact_update_event = TaskArtifactUpdateEvent(
             taskId=task.id,
+            contextId=task.contextId,
             artifact=artifact,
             append=False,
             lastChunk=True,
         )
 
     task_status_event = TaskStatusUpdateEvent(
         taskId=task.id,
+        contextId=task.contextId,
         status=TaskStatus(
             state=task_state,
             message=message,