1  """
  2  Helpers for creating and consuming A2A Task objects.
  3  """
  4  from datetime import datetime, timezone
  5  from typing import Any, Dict, List, Optional
  6  
  7  from a2a.types import (
  8      Artifact,
  9      Message,
 10      Task,
 11      TaskState,
 12      TaskStatus,
 13  )
 14  
 15  
 16  # --- Creation Helpers ---
 17  
 18  
 19  def create_initial_task(
 20      task_id: str,
 21      context_id: str,
 22      agent_name: str,
 23  ) -> Task:
 24      """
 25      Creates an initial Task object, typically for returning to a client
 26      after a task has been submitted.
 27  
 28      Args:
 29          task_id: The unique ID for the task.
 30          context_id: The context/session ID for the task.
 31          agent_name: The name of the agent handling the task.
 32  
 33      Returns:
 34          A new `Task` object with 'submitted' status.
 35      """
 36      initial_status = TaskStatus(state=TaskState.submitted)
 37      return Task(
 38          id=task_id,
 39          context_id=context_id,
 40          status=initial_status,
 41          kind="task",
 42          metadata={"agent_name": agent_name},
 43      )
 44  
 45  
 46  def create_task_status(
 47      state: TaskState,
 48      message: Optional[Message] = None,
 49  ) -> TaskStatus:
 50      """
 51      Creates a TaskStatus object.
 52  
 53      Args:
 54          state: The state of the task.
 55          message: An optional message providing more details.
 56  
 57      Returns:
 58          A new `TaskStatus` object with a current timestamp.
 59      """
 60      return TaskStatus(
 61          state=state,
 62          message=message,
 63          timestamp=datetime.now(timezone.utc).isoformat(),
 64      )
 65  
 66  
 67  def create_final_task(
 68      task_id: str,
 69      context_id: str,
 70      final_status: TaskStatus,
 71      artifacts: Optional[List[Artifact]] = None,
 72      metadata: Optional[Dict[str, Any]] = None,
 73  ) -> Task:
 74      """
 75      Creates a final Task object, typically for sending as a final response.
 76  
 77      Args:
 78          task_id: The unique ID for the task.
 79          context_id: The context/session ID for the task.
 80          final_status: The final status of the task (e.g., completed, failed).
 81          artifacts: A list of artifacts produced by the task.
 82          metadata: Optional metadata to include in the task.
 83  
 84      Returns:
 85          A new `Task` object representing the final state.
 86      """
 87      return Task(
 88          id=task_id,
 89          context_id=context_id,
 90          status=final_status,
 91          artifacts=artifacts,
 92          metadata=metadata,
 93          kind="task",
 94      )
 95  
 96  
 97  # --- Consumption Helpers ---
 98  
 99  
100  def get_task_id(task: Task) -> str:
101      """Safely retrieves the ID from a Task object."""
102      return task.id
103  
104  
105  def get_task_context_id(task: Task) -> str:
106      """Safely retrieves the context ID from a Task object."""
107      return task.context_id
108  
109  
110  def get_task_status(task: Task) -> TaskState:
111      """Safely retrieves the state from a Task's status."""
112      return task.status.state
113  
114  
115  def get_task_history(task: Task) -> Optional[List[Message]]:
116      """Safely retrieves the history from a Task object."""
117      return task.history
118  
119  
120  def get_task_artifacts(task: Task) -> Optional[List[Artifact]]:
121      """Safely retrieves the artifacts from a Task object."""
122      return task.artifacts
123  
124  
125  def get_task_metadata(task: Task) -> Optional[Dict[str, Any]]:
126      """Safely retrieves the metadata from a Task object."""
127      return task.metadata