[Feat]: Add helper methods to TaskUpdater for creating messages with auto-populated taskId and contextId #516
Description
Is your feature request related to a problem? Please describe.
We have to explictly pass task_id and context_id to new_agent_text_message() every single time I create a message.
TaskUpdater already has access to task_id and context_id (passed in its constructor), but we still have to remember to manually pass these to new_agent_text_message() when creating status messages. This is verbose and error-prone:
from a2a.server.tasks import TaskUpdater
from a2a.utils import new_agent_text_message
from a2a.types import TaskState
# Current verbose pattern - easy to forget the IDs
async def some_agent_function(task, event_queue):
updater = TaskUpdater(event_queue, task.id, task.context_id)
# 😞 Repetitive - must manually pass IDs every time
await updater.update_status(
TaskState.input_required,
new_agent_text_message(
"Please provide more information",
task_id=updater.task_id, # Have to remember this
context_id=updater.context_id # Have to remember this
),
final=True
)Problems with current approach:
- ❌ Verbose and repetitive -
task_idandcontext_idmust be passed every time - ❌ Error-prone - Easy to forget these parameters (and code still runs!)
- ❌ Non-obvious - Not clear from API that these should be included
- ❌ Inconsistent results - Developers who forget these params create incomplete messages
What happens when I forget the IDs
The TaskUpdater class already has access to task_id and context_id (passed in its constructor), but developers must remember to manually pass these to new_agent_text_message() when creating status messages. This common pattern is verbose and easy to get wrong:
from a2a.server.tasks import TaskUpdater
from a2a.utils import new_agent_text_message
from a2a.types import TaskState
# Current verbose pattern - easy to forget the IDs
async def some_agent_function(task, event_queue):
updater = TaskUpdater(event_queue, task.id, task.context_id)
# 😞 Repetitive - must manually pass IDs every time
await updater.update_status(
TaskState.input_required,
new_agent_text_message(
"Please provide more information",
task_id=updater.task_id, # Have to remember this
context_id=updater.context_id # Have to remember this
),
final=True
)Problems with current approach:
- ❌ Verbose and repetitive -
task_idandcontext_idmust be passed every time - ❌ Error-prone - Easy to forget these parameters (and code still runs!)
- ❌ Non-obvious - Not clear from API that these should be included
- ❌ Inconsistent results - Developers who forget these params create incomplete messages
What developers want (simple and correct)
{
"status": {
"message": {
"kind": "message",
"messageId": "abc-123",
"taskId": "9d08ff19-931a-4c76-a82d-b3bce2019da9", // ✅ Included
"contextId": "3ce145ed-1373-4f23-9634-efcc59d95f19", // ✅ Included
"role": "agent",
"parts": [{"kind": "text", "text": "Please provide more information"}]
},
"state": "input-required",
"timestamp": "2025-10-21T03:20:10.769279+00:00"
}
}What developers often get (when they forget the IDs)
{
"status": {
"message": {
"kind": "message",
"messageId": "abc-123",
// ❌ Missing taskId
// ❌ Missing contextId
"role": "agent",
"parts": [{"kind": "text", "text": "Please provide more information"}]
},
"state": "input-required",
"timestamp": "2025-10-21T03:20:10.769279+00:00"
}
}Note: While taskId and contextId are optional per the spec (Section 6.4), all official A2A specification examples include them (Section 9.4), indicating they should be considered best practice.
Describe the solution you'd like
Add convenience methods to TaskUpdater that automatically create messages with taskId and contextId pre-filled, so I don't have to pass them manually every time.
Implementation
class TaskUpdater:
"""Helper class for agents to publish updates to a task's event queue."""
def __init__(self, event_queue: EventQueue, task_id: str, context_id: str):
self.event_queue = event_queue
self.task_id = task_id
self.context_id = context_id
# ... existing code ...
def create_message(self, text: str) -> Message:
"""Create an agent message with taskId and contextId pre-filled.
This is a convenience method that creates a message following A2A
best practices by automatically including task and context identifiers.
Args:
text: The text content of the message.
Returns:
A Message object with role='agent', messageId (auto-generated),
taskId, and contextId populated.
Example:
>>> updater = TaskUpdater(queue, task.id, task.context_id)
>>> await updater.update_status(
... TaskState.input_required,
... updater.create_message("Please provide more details")
... )
"""
return new_agent_text_message(
text,
task_id=self.task_id,
context_id=self.context_id,
)
def create_message_with_parts(self, parts: list[Part]) -> Message:
"""Create an agent message with multiple parts and IDs pre-filled.
Similar to create_message() but accepts a list of Part objects
instead of plain text.
Args:
parts: List of Part objects for the message content.
Returns:
A Message object with taskId and contextId populated.
"""
return new_agent_parts_message(
parts,
task_id=self.task_id,
context_id=self.context_id,
)Usage Example
Before (verbose and error-prone):
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
task = context.current_task or new_task(context.message)
updater = TaskUpdater(event_queue, task.id, task.context_id)
# 😞 Must remember to pass IDs every single time
await updater.update_status(
TaskState.working,
new_agent_text_message(
"Processing your request...",
task_id=updater.task_id,
context_id=updater.context_id
)
)
await updater.update_status(
TaskState.input_required,
new_agent_text_message(
"Please confirm to proceed",
task_id=updater.task_id,
context_id=updater.context_id
),
final=True
)After (clean and correct):
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
task = context.current_task or new_task(context.message)
updater = TaskUpdater(event_queue, task.id, task.context_id)
# 😊 Simple and automatically includes the IDs
await updater.update_status(
TaskState.working,
updater.create_message("Processing your request...")
)
await updater.update_status(
TaskState.input_required,
updater.create_message("Please confirm to proceed"),
final=True
)Benefits
✅ Less boilerplate - No need to pass task_id and context_id repeatedly
✅ Cleaner code - updater.create_message("text") is more concise
✅ Fewer errors - Impossible to forget the IDs
✅ Better defaults - Makes best practice the easy path
✅ No breaking changes - Adds new methods, doesn't change existing behavior
✅ Spec compliant - Automatically follows A2A spec examples pattern
Method Naming
Suggested name: create_message() (or create_agent_message())
Rationale:
- Short and clear
- Indicates it's creating something (not just returning existing data)
- Fits TaskUpdater's role as a helper for task operations
- Parallel to existing methods like
update_status(),add_artifact(),complete()
Alternative names to consider:
message()- Very concise but less clear about what it doesnew_message()- Clear but slightly redundant withcreate_make_message()- Less conventional in Python
Additional Context
A2A Spec References:
- Section 6.4 (Message): Defines
taskIdandcontextIdas optional fields - Section 9.4 (Multi-Turn Task Flow): All example messages include both fields
Example from spec Section 9.4:
{
"status": {
"state": "input-required",
"message": {
"role": "agent",
"parts": [...],
"messageId": "c2e1b2dd-f200-4b04-bc22-1b0c65a1aad2",
"taskId": "3f36680c-7f37-4a5f-945e-d78981fafd36", // Present in spec example
"contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4" // Present in spec example
}
}
}Describe alternatives you've considered
No response
Additional context
Issue analysed and created using GitHub Copilot
Code of Conduct
- I agree to follow this project's Code of Conduct