Template improvements (#5)

* Messenger: add timeout parameter

* Improve executor logic; add echo agent as example

* Clarify publishing steps in README

Author: staverm

README.md

@@ -56,4 +56,8 @@ ghcr.io/<your-username>/<your-repo-name>:latest
 ```
 ghcr.io/<your-username>/<your-repo-name>:1.0.0
 ghcr.io/<your-username>/<your-repo-name>:1
-```
+```
+
+Once the workflow completes, find your Docker image in the Packages section (right sidebar of your repository). Configure the package visibility in package settings.
+
+> **Note:** Organization repositories may need package write permissions enabled manually (Settings → Actions → General). Version tags must follow [semantic versioning](https://semver.org/) (e.g., `v1.0.0`).

src/agent.py

@@ -1,19 +1,32 @@
 from a2a.server.tasks import TaskUpdater
+from a2a.types import Message, TaskState, Part, TextPart
+from a2a.utils import get_message_text, new_agent_text_message
+
 from messenger import Messenger
 
 
 class Agent:
     def __init__(self):
         self.messenger = Messenger()
-        # initialize other state here
+        # Initialize other state here
 
-    async def run(self, input_text: str, updater: TaskUpdater) -> None:
+    async def run(self, message: Message, updater: TaskUpdater) -> None:
         """Implement your agent logic here.
 
         Args:
-            input_text: The incoming message text
+            message: The incoming message
             updater: Report progress (update_status) and results (add_artifact)
 
         Use self.messenger.talk_to_agent(message, url) to call other agents.
         """
-        raise NotImplementedError("Agent not implemented.")
+        input_text = get_message_text(message)
+
+        # Replace this example code with your agent logic
+
+        await updater.update_status(
+            TaskState.working, new_agent_text_message("Thinking...")
+        )
+        await updater.add_artifact(
+            parts=[Part(root=TextPart(text=input_text))],
+            name="Echo",
+        )

src/executor.py

@@ -3,6 +3,7 @@
 from a2a.server.tasks import TaskUpdater
 from a2a.types import (
     Task,
+    TaskState,
     UnsupportedOperationError,
     InvalidRequestError,
 )
@@ -15,44 +16,47 @@
 from agent import Agent
 
 
-TERMINAL_STATES = ["completed", "canceled", "rejected", "failed"]
+TERMINAL_STATES = {
+    TaskState.completed,
+    TaskState.canceled,
+    TaskState.failed,
+    TaskState.rejected
+}
 
 
 class Executor(AgentExecutor):
     def __init__(self):
-        self.agent_store: dict[str, Agent] = {}
+        self.agents: dict[str, Agent] = {} # context_id to agent instance
 
-    async def execute(
-        self,
-        context: RequestContext,
-        event_queue: EventQueue,
-    ) -> None:
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
         msg = context.message
         if not msg:
             raise ServerError(error=InvalidRequestError(message="Missing message in request"))
 
         task = context.current_task
-        if not task or task.status.state in TERMINAL_STATES:
+        if task and task.status.state in TERMINAL_STATES:
+            raise ServerError(error=InvalidRequestError(message=f"Task {task.id} already processed (state: {task.status.state})"))
+
+        if not task:
             task = new_task(msg)
+            await event_queue.enqueue_event(task)
 
         context_id = task.context_id
-        agent = self.agent_store.get(context_id)
+        agent = self.agents.get(context_id)
         if not agent:
             agent = Agent()
-            self.agent_store[context_id] = agent
+            self.agents[context_id] = agent
 
-        await event_queue.enqueue_event(task)
         updater = TaskUpdater(event_queue, task.id, context_id)
-        await updater.start_work(new_agent_text_message(text="Thinking...", context_id=context_id, task_id=task.id))
 
+        await updater.start_work()
         try:
-            await agent.run(context.get_user_input(), updater)
-            await updater.complete()
+            await agent.run(msg, updater)
+            if not updater._terminal_state_reached:
+                await updater.complete()
         except Exception as e:
             print(f"Task failed with agent error: {e}")
-            await updater.failed(new_agent_text_message(f"Agent error: {e}", context_id=context_id))
+            await updater.failed(new_agent_text_message(f"Agent error: {e}", context_id=context_id, task_id=task.id))
 
-    async def cancel(
-        self, request: RequestContext, event_queue: EventQueue
-    ) -> Task | None:
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
         raise ServerError(error=UnsupportedOperationError())

src/messenger.py

@@ -20,15 +20,18 @@
 DEFAULT_TIMEOUT = 300
 
 
-def create_message(*, role: Role = Role.user, text: str, context_id: str | None = None) -> Message:
+def create_message(
+    *, role: Role = Role.user, text: str, context_id: str | None = None
+) -> Message:
     return Message(
         kind="message",
         role=role,
         parts=[Part(TextPart(kind="text", text=text))],
         message_id=uuid4().hex,
-        context_id=context_id
+        context_id=context_id,
     )
 
+
 def merge_parts(parts: list[Part]) -> str:
     chunks = []
     for part in parts:
@@ -38,9 +41,17 @@ def merge_parts(parts: list[Part]) -> str:
             chunks.append(json.dumps(part.root.data, indent=2))
     return "\n".join(chunks)
 
-async def send_message(message: str, base_url: str, context_id: str | None = None, streaming=False, consumer: Consumer | None = None):
+
+async def send_message(
+    message: str,
+    base_url: str,
+    context_id: str | None = None,
+    streaming: bool = False,
+    timeout: int = DEFAULT_TIMEOUT,
+    consumer: Consumer | None = None,
+):
     """Returns dict with context_id, response and status (if exists)"""
-    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as httpx_client:
+    async with httpx.AsyncClient(timeout=timeout) as httpx_client:
         resolver = A2ACardResolver(httpx_client=httpx_client, base_url=base_url)
         agent_card = await resolver.get_agent_card()
         config = ClientConfig(
@@ -54,10 +65,7 @@ async def send_message(message: str, base_url: str, context_id: str | None = Non
 
         outbound_msg = create_message(text=message, context_id=context_id)
         last_event = None
-        outputs = {
-            "response": "",
-            "context_id": None
-        }
+        outputs = {"response": "", "context_id": None}
 
         # if streaming == False, only one event is generated
         async for event in client.send_message(outbound_msg):
@@ -88,19 +96,31 @@ class Messenger:
     def __init__(self):
         self._context_ids = {}
 
-    async def talk_to_agent(self, message: str, url: str, new_conversation: bool = False):
+    async def talk_to_agent(
+        self,
+        message: str,
+        url: str,
+        new_conversation: bool = False,
+        timeout: int = DEFAULT_TIMEOUT,
+    ):
         """
         Communicate with another agent by sending a message and receiving their response.
 
         Args:
             message: The message to send to the agent
             url: The agent's URL endpoint
             new_conversation: If True, start fresh conversation; if False, continue existing conversation
+            timeout: Timeout in seconds for the request (default: 300)
 
         Returns:
             str: The agent's response message
         """
-        outputs = await send_message(message=message, base_url=url, context_id=None if new_conversation else self._context_ids.get(url, None))
+        outputs = await send_message(
+            message=message,
+            base_url=url,
+            context_id=None if new_conversation else self._context_ids.get(url, None),
+            timeout=timeout,
+        )
         if outputs.get("status", "completed") != "completed":
             raise RuntimeError(f"{url} responded with: {outputs}")
         self._context_ids[url] = outputs.get("context_id", None)

src/server.py

@@ -27,8 +27,8 @@ def main():
         id="",
         name="",
         description="",
-        tags=[""],
-        examples=[""]
+        tags=[],
+        examples=[]
     )
 
     agent_card = AgentCard(