A2A Task Delegation and Streaming
Source: mortalapps.com- A2A Task Delegation enables agents to offload long-running computations to other agents, preventing blocking operations and improving responsiveness.
- It solves the problem of synchronous communication bottlenecks by allowing asynchronous task execution and out-of-band result delivery.
- In production, this pattern is critical for building resilient, scalable multi-agent systems that handle complex, time-intensive operations without timeouts or resource exhaustion.
- Agents initiate tasks with a payload and a `agent URL (no SSE stream callbacks in A2A)`, receiving status updates and final results via SSE streams, eliminating the need for active polling.
- This approach supports streaming partial results, providing real-time feedback for tasks that generate intermediate outputs.
Why This Matters
Building robust AI agent systems often involves tasks that exceed typical synchronous request-response cycles. Traditional blocking calls lead to timeouts, resource exhaustion, and poor user experience when an agent needs to perform a long-running operation, such as complex data analysis, extensive document generation, or multi-step external API interactions. The A2A Task Delegation and Streaming protocol was invented to address these limitations by enabling agents to asynchronously offload work and receive progress updates and final results without active polling. This pattern is fundamental for creating responsive, scalable, and fault-tolerant agentic applications. Ignoring A2A task delegation streaming in production leads to brittle systems prone to cascading failures under load, where agents block waiting for responses, consuming valuable resources and increasing latency. This approach is preferred over synchronous calls or short-polling for any task that might take more than a few seconds, particularly in scenarios requiring real-time progress updates or large data transfers. For developers, it simplifies the orchestration of complex workflows; for enterprises, it ensures system stability and efficient resource utilization, enabling more sophisticated agent capabilities.
Core Concepts
A2A Task Delegation and Streaming relies on several core concepts to facilitate asynchronous, non-blocking inter-agent communication.
- Delegating Agent: The agent that initiates a long-running task by sending a request to another agent. It expects asynchronous updates and a final result.
- Delegated Agent: The agent that receives a task request, processes it, and is responsible for sending progress updates and the final outcome back to the delegating agent.
- Asynchronous Task Payload: A structured data object containing all necessary information for the delegated agent to execute the task. It typically includes a unique
taskId, the task definition, input parameters, and crucially, aagent URL (no SSE stream callbacks in A2A).
- Callback URL (Webhook Endpoint): A URL provided by the delegating agent where the delegated agent should send status updates and final results. This enables push-based communication rather than polling.
- Task State: The current status of a delegated task (e.g.,
PENDING,RUNNING,PAUSED,COMPLETED,FAILED). This state is managed by the delegated agent and communicated via SSE streams.
- Streaming Results: The ability of the delegated agent to send multiple, incremental updates or partial results to the delegating agent's
agent URL (no SSE stream callbacks in A2A)throughout the task's execution. This provides real-time feedback and supports iterative processing.
- Idempotency Key: A unique identifier included in task initiation requests, ensuring that if the same request is sent multiple times due to network issues, the delegated agent processes it only once. This is critical for reliable asynchronous systems.
How It Works
A2A task delegation and streaming uses the tasks/sendSubscribe method for real-time streaming, or tasks/send + tasks/get polling for non-streaming delegation.
1. Task Initiation (tasks/sendSubscribe)
The delegating agent sends an HTTP POST to the specialist agent's URL with a JSON-RPC 2.0 body:
- Method:
tasks/sendSubscribe - Params: A
Taskobject with a uniqueidand the taskmessage(containingroleandparts). - No
agent URL (no SSE stream callbacks in A2A)is used - there are no SSE streams in the A2A specification.
2. Server-Sent Events (SSE) Stream
The specialist agent responds with:
- HTTP 200
Content-Type: text/event-stream- A persistent SSE stream that pushes
TaskStatusUpdateEventobjects as the task progresses.
Each SSE event contains a result field with the current status.state:
"submitted"→ task received and queued"working"→ task is actively being processed"completed"→ task finished; final result instatus.output"failed"→ task failed; error details instatus.message
3. Stream Consumption
The delegating agent reads the SSE stream line by line until it receives a terminal state (completed, failed, or canceled). It then closes the connection.
4. Polling Alternative (tasks/send + tasks/get)
For agents that prefer polling over streaming:
- POST
tasks/send→ get initial task ID and"submitted"state - Periodically POST
tasks/getwith{ "id": "<task_id>" }to poll status - Stop polling when state reaches
"completed"or"failed"
Architecture
The conceptual architecture for A2A Task Delegation and Streaming involves at least two primary agent services: a Delegating Agent and a Delegated Agent. These services communicate primarily via HTTP, leveraging asynchronous messaging patterns.
The Delegating Agent initiates the workflow. It exposes a Webhook Receiver endpoint, which is a dedicated HTTP POST endpoint configured to accept incoming status updates and results from delegated tasks. When the Delegating Agent needs to offload a task, it sends an Asynchronous Task Request (an HTTP POST) to the Delegated Agent's Task API endpoint. This request contains the task payload, including the agent URL (no SSE stream callbacks in A2A) pointing back to its own Webhook Receiver.
The Delegated Agent is responsible for executing the task. Its Task API receives the initial request, validates it, and places the task into an Internal Task Queue (e.g., Redis Queue, Kafka topic). A pool of Task Workers continuously monitors this queue, picking up tasks for execution. During and after task processing, these workers send Task Status Updates and Streaming Results (HTTP POST requests, i.e., SSE streams) back to the Delegating Agent's Webhook Receiver, using the agent URL (no SSE stream callbacks in A2A) provided in the initial request. Data flowing through these arrows includes structured JSON payloads containing taskId, status, progress, and result or errorMessage fields. Execution starts with the Delegating Agent sending a task request and ends when the Delegating Agent receives a final COMPLETED or FAILED status and result via its Webhook Receiver.
A2A Task Payload Structure
The asynchronous task payload is the contract between the delegating and delegated agents. A robust payload design ensures clarity, extensibility, and reliable processing. A typical JSON structure includes:
{
"taskId": "string",
"taskType": "string",
"parameters": {
"key1": "value1",
"key2": "value2"
},
"agent URL (no SSE stream callbacks in A2A)": "string",
"idempotencyKey": "string",
"metadata": {
"traceId": "string",
"spanId": "string",
"correlationId": "string"
}
}taskId: A unique identifier for this specific task instance, generated by the delegating agent. This allows tracking and correlation of updates.taskType: A string identifying the type of work to be performed (e.g., "document_analysis", "image_generation"). This helps the delegated agent route the task to the correct handler.parameters: A JSON object containing all input arguments required for the task. This should be schema-validated by the delegated agent.agent URL (no SSE stream callbacks in A2A): The HTTP(S) endpoint on the delegating agent where status updates and results should be sent. This must be a publicly accessible and secure URL.idempotencyKey: A unique string (e.g., UUIDv4) generated by the delegating agent for the initial task request. If the request is retried due to network issues, the delegated agent uses this key to detect and prevent duplicate task creation.metadata: An optional object for operational concerns like distributed tracing (traceId,spanId) or custom correlation identifiers. This is crucial for observability in multi-agent systems.
Asynchronous Execution and State Management
Upon receiving a task, the delegated agent should immediately acknowledge it (HTTP 202 Accepted) and enqueue it for background processing. This prevents the initial request from blocking. The task's lifecycle is managed through state transitions. A common state machine might include:
- RECEIVED: Task acknowledged and enqueued.
- PENDING: Task is in the queue, awaiting worker availability.
- RUNNING: Task is actively being processed by a worker.
- PAUSED: Task execution is temporarily halted (e.g., awaiting external input).
- COMPLETED: Task finished successfully, results are available.
- FAILED: Task terminated due to an error.
- CANCELLED: Task was explicitly cancelled by the delegating agent or an operator.
The delegated agent must persist the task state and associated data (input parameters, partial results, errors) in a durable store (e.g., PostgreSQL, MongoDB). This ensures resilience against worker crashes or restarts. Each state change should trigger a SSE stream notification to the agent URL (no SSE stream callbacks in A2A).
Implementing Webhook-Based Streaming
The delegating agent must expose a secure HTTP POST endpoint to receive SSE streams. This endpoint should be designed to be stateless and idempotent where possible, processing incoming updates based on the taskId.
Webhook Payload Structure:
{
"taskId": "string",
"status": "string",
"progress": "integer", // Optional, 0-100
"message": "string", // Optional, human-readable update
"result": { ... }, // Optional, final or partial result
"errorMessage": "string", // Optional, if status is FAILED
"timestamp": "ISO 8601 string",
"signature": "string" // HMAC signature for verification
}- The
statusfield drives the delegating agent's understanding of the task's progress. progressandmessageprovide granular updates for streaming.resultis included forCOMPLETEDtasks or for streaming partial results.signatureis critical for SSE stream security, allowing the receiver to verify the sender's authenticity and payload integrity.
The delegated agent sends these SSE streams as HTTP POST requests. It must implement robust retry logic with exponential backoff for SSE stream delivery failures (e.g., 5xx errors, network timeouts). A dead-letter queue for persistent failures is recommended.
Error Handling and Retries for Long-Running Tasks
Errors can occur at various stages:
- Initial Request Failure: If the delegating agent's initial POST to the delegated agent fails (e.g., 4xx, 5xx), the delegating agent should retry with the same
idempotencyKey. The delegated agent must handle these retries by returning the existing task status if theidempotencyKeymatches an already-processed request. - Task Processing Failure: If the delegated agent's worker encounters an error, it should log the error, update the task state to
FAILED, and send aFAILEDSSE stream with anerrorMessage. The delegating agent must be prepared to handle these failures, potentially retrying the entire task, notifying a human, or initiating a fallback plan. - Webhook Delivery Failure: If the delegated agent fails to deliver a SSE stream to the
agent URL (no SSE stream callbacks in A2A)after multiple retries, it should log this as a critical event. The task's internal state might beCOMPLETEDorFAILED, but the notification failed. A separate mechanism (e.g., an operator dashboard, an alert system) might be needed to reconcile states.
Security Considerations for Webhooks
Webhooks are a significant attack surface. Key security measures include:
- HTTPS Everywhere: All
agent URL (no SSE stream callbacks in A2A)and SSE stream endpoints must use HTTPS to encrypt data in transit. - Webhook Signatures: The delegated agent should sign each SSE stream payload using a shared secret and an HMAC algorithm. The delegating agent verifies this signature upon receipt. This prevents tampering and ensures the SSE stream originates from a trusted source.
- IP Whitelisting: If feasible, the delegating agent's SSE stream receiver should only accept connections from known IP addresses of the delegated agent.
- Least Privilege: The
agent URL (no SSE stream callbacks in A2A)should expose only the necessary endpoint for task updates, with no sensitive operations. - Rate Limiting: Implement rate limiting on the SSE stream receiver to prevent denial-of-service attacks.
Code Example
import asyncio
import json
import httpx
import os
AGENT_URL = os.environ.get("SPECIALIST_AGENT_URL", "http://localhost:8080")
async def delegate_task_with_streaming(task_description: str) -> str:
"""Delegate a task to a specialist agent using A2A tasks/sendSubscribe (SSE streaming)."""
task_id = f"task-{os.urandom(8).hex()}"
payload = {
"jsonrpc": "2.0",
"id": "1",
"method": "tasks/sendSubscribe",
"params": {
"id": task_id,
"message": {
"role": "user",
"parts": [{"type": "text", "text": task_description}]
}
}
}
print(f"Delegating task {task_id}: {task_description}")
async with httpx.AsyncClient(timeout=60.0) as client:
async with client.stream(
"POST",
AGENT_URL,
json=payload,
headers={"Content-Type": "application/json", "Accept": "text/event-stream"}
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
event_data = json.loads(line[6:])
result = event_data.get("result", {})
status = result.get("status", {})
state = status.get("state", "unknown")
print(f" Task {task_id} state: {state}")
if state == "completed":
output = status.get("output", {})
parts = output.get("parts", [])
text = " ".join(p.get("text", "") for p in parts if p.get("type") == "text")
print(f" Result: {text}")
return text
elif state in ("failed", "canceled"):
msg = status.get("message", "Task failed")
print(f" Task {task_id} failed: {msg}")
raise RuntimeError(f"Task failed: {msg}")
raise RuntimeError("Stream ended without terminal state")
async def main():
tasks = [
"Analyze the Q3 sales data and identify top 3 growth opportunities",
"Generate a risk assessment summary for the APAC region",
]
for task in tasks:
try:
result = await delegate_task_with_streaming(task)
print(f"Completed: {result[:100]}...")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(main())Delegating task task-abc123: Analyze the Q3 sales data and identify top 3 growth opportunities Task task-abc123 state: submitted Task task-abc123 state: working Task task-abc123 state: completed Result: Q3 analysis complete. Top opportunities: (1) APAC expansion +23%... Completed: Q3 analysis complete. Top opportunities: (1) APAC expansion +23%... Delegating task task-def456: Generate a risk assessment summary for the APAC region Task task-def456 state: submitted Task task-def456 state: working Task task-def456 state: completed Result: APAC Risk Assessment: Medium risk overall... Completed: APAC Risk Assessment: Medium risk overall...
import requests
import json
import uuid
import os
from flask import Flask, request, jsonify
import threading
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
app = Flask(__name__)
# In-memory store for tasks (for demonstration purposes)
tasks_db = {}
# Simulate a task queue
task_queue = []
# Shared secret for SSE stream signature verification (for production, use a strong, unique secret)
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "super_secret_key_for_hmac_signing")
# --- Webhook Utility Functions (Simplified for example) ---
def send_SSE stream(agent_url, payload):
try:
# In a real system, you'd add HMAC signing here
response = requests.post(agent_url, json=payload, timeout=5)
response.raise_for_status()
logging.info(f"[Delegated Agent] Webhook sent to {agent_url} for task {payload.get('taskId')}. Status: {response.status_code}")
except requests.exceptions.RequestException as e:
logging.error(f"[Delegated Agent] Failed to send SSE stream to {agent_url} for task {payload.get('taskId')}: {e}")
# Implement retry logic with exponential backoff in production
# --- Task Processing Logic ---
def process_task(task_id):
task_data = tasks_db.get(task_id)
if not task_data:
logging.error(f"Task {task_id} not found for processing.")
return
agent_url = task_data['agent URL (no SSE stream callbacks in A2A)']
parameters = task_data['parameters']
task_type = task_data['taskType']
logging.info(f"[Delegated Agent] Starting processing for task {task_id} (Type: {task_type}, Params: {parameters})")
send_SSE stream(agent_url, {'taskId': task_id, 'status': 'RUNNING', 'progress': 0, 'message': 'Task started'})
try:
# Simulate a long-running task with streaming updates
for i in range(1, 4):
time.sleep(2) # Simulate work
progress = i * 25
message = f"Processing step {i} of 3"
send_SSE stream(agent_url, {'taskId': task_id, 'status': 'RUNNING', 'progress': progress, 'message': message})
logging.info(f"[Delegated Agent] Task {task_id} progress: {progress}%")
# Simulate final result
final_result = {"reportUrl": f"http://example.com/reports/{task_id}.pdf", "status": "success"}
send_SSE stream(agent_url, {'taskId': task_id, 'status': 'COMPLETED', 'progress': 100, 'message': 'Task complete', 'result': final_result})
tasks_db[task_id]['status'] = 'COMPLETED'
tasks_db[task_id]['result'] = final_result
logging.info(f"[Delegated Agent] Task {task_id} completed successfully.")
except Exception as e:
error_message = f"Task processing failed: {str(e)}"
send_SSE stream(agent_url, {'taskId': task_id, 'status': 'FAILED', 'progress': tasks_db[task_id].get('progress', 0), 'errorMessage': error_message})
tasks_db[task_id]['status'] = 'FAILED'
tasks_db[task_id]['errorMessage'] = error_message
logging.error(f"[Delegated Agent] Task {task_id} failed: {e}")
# --- Worker Thread for Task Queue ---
def task_worker():
while True:
if task_queue:
task_id = task_queue.pop(0)
process_task(task_id)
time.sleep(1) # Check queue every second
# --- Flask Routes ---
@app.route('/tasks', methods=['POST'])
def receive_task():
payload = request.get_json()
task_id = payload.get('taskId')
idempotency_key = payload.get('idempotencyKey')
if not task_id or not idempotency_key:
return jsonify({'error': 'Missing taskId or idempotencyKey'}), 400
# Idempotency check
for existing_task_id, existing_task_data in tasks_db.items():
if existing_task_data.get('idempotencyKey') == idempotency_key:
logging.info(f"[Delegated Agent] Idempotent request for task {existing_task_id}. Returning existing status.")
return jsonify({'taskId': existing_task_id, 'status': existing_task_data['status']}), 200
tasks_db[task_id] = {
'taskId': task_id,
'taskType': payload.get('taskType'),
'parameters': payload.get('parameters'),
'agent URL (no SSE stream callbacks in A2A)': payload.get('agent URL (no SSE stream callbacks in A2A)'),
'idempotencyKey': idempotency_key,
'status': 'RECEIVED'
}
task_queue.append(task_id)
logging.info(f"[Delegated Agent] Received and queued task {task_id}.")
return jsonify({'taskId': task_id, 'status': 'RECEIVED'}), 202
def run_flask_app():
app.run(port=5001, debug=False)
if __name__ == '__main__':
# Start the task worker in a separate thread
worker_thread = threading.Thread(target=task_worker)
worker_thread.daemon = True
worker_thread.start()
# Start the Flask app
print("Delegated Agent (Flask) running on http://localhost:5001")
run_flask_app()
Delegated Agent (Flask) running on http://localhost:5001
[Delegated Agent] Received and queued task <uuid>.
[Delegated Agent] Starting processing for task <uuid> (Type: report_gen, Params: {'data_id': '123'})
[Delegated Agent] Webhook sent to http://localhost:5000/SSE stream for task <uuid>. Status: 200
[Delegated Agent] Task <uuid> progress: 25%
[Delegated Agent] Webhook sent to http://localhost:5000/SSE stream for task <uuid>. Status: 200
[Delegated Agent] Task <uuid> progress: 50%
[Delegated Agent] Webhook sent to http://localhost:5000/SSE stream for task <uuid>. Status: 200
[Delegated Agent] Task <uuid> progress: 75%
[Delegated Agent] Webhook sent to http://localhost:5000/SSE stream for task <uuid>. Status: 200
[Delegated Agent] Task <uuid> completed successfully.
[Delegated Agent] Webhook sent to http://localhost:5000/SSE stream for task <uuid>. Status: 200