A2A Task Management

Manage A2A task lifecycle, store, and JSON-RPC methods for listing, getting, and cancelling tasks

python

from praisonaiagents import Agent

agent = Agent(name="task-manager", instructions="Manage A2A tasks via JSON-RPC.")
agent.start("List all active A2A tasks and cancel any that are overdue.")

Tasks are the core unit of work in the A2A protocol, managing the complete lifecycle from submission to completion with built-in state tracking and storage.

The user asks for task status; the agent calls A2A JSON-RPC to list or cancel remote tasks.

Diagram

How It Works

Diagram

Quick Start

Send Message (Creates Task)

python

import aiohttp
import json

async def send_message():
    payload = {
        "jsonrpc": "2.0",
        "method": "message/send",
        "id": "1",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"text": "Analyze this data"}]
            }
        }
    }

    async with aiohttp.ClientSession() as session:
        async with session.post("http://localhost:8000/a2a", json=payload) as response:
            result = await response.json()
            task_id = result["result"]["id"]
            print(f"Task created: {task_id}")

Get Task Status

python

async def get_task_status(task_id):
    payload = {
        "jsonrpc": "2.0",
        "method": "tasks/get", 
        "id": "2",
        "params": {"id": task_id}
    }

    async with aiohttp.ClientSession() as session:
        async with session.post("http://localhost:8000/a2a", json=payload) as response:
            result = await response.json()
            state = result["result"]["status"]["state"]
            print(f"Task state: {state}")

Cancel Task

python

async def cancel_task(task_id):
    payload = {
        "jsonrpc": "2.0",
        "method": "tasks/cancel",
        "id": "3", 
        "params": {"id": task_id}
    }

    async with aiohttp.ClientSession() as session:
        async with session.post("http://localhost:8000/a2a", json=payload) as response:
            result = await response.json()
            print(f"Cancelled: {result['result']['status']['state']}")

Task Lifecycle

Every A2A interaction creates a Task that progresses through defined states:

Diagram

Task States

State	Description
`submitted`	Task received, not yet processing
`working`	Agent is actively processing
`completed`	Task finished successfully
`failed`	Task failed with an error
`cancelled`	Task was cancelled by client
`input_required`	Agent needs more information
`auth_required`	Authentication required

Python Task States

python

from praisonaiagents.ui.a2a import TaskState

print(TaskState.SUBMITTED.value)      # "submitted"
print(TaskState.WORKING.value)        # "working"
print(TaskState.COMPLETED.value)      # "completed"
print(TaskState.FAILED.value)         # "failed"
print(TaskState.CANCELLED.value)      # "cancelled"
print(TaskState.INPUT_REQUIRED.value) # "input_required"
print(TaskState.AUTH_REQUIRED.value)  # "auth_required"

Task Structure

A task contains all information about an A2A interaction:

python

from praisonaiagents.ui.a2a import Task, TaskStatus, TaskState, Message, Artifact

task = Task(
    id="task-uuid-123",
    context_id="ctx-uuid-456",
    status=TaskStatus(state=TaskState.COMPLETED),
    history=[
        Message(role="user", parts=[TextPart(text="Hello")]),
        Message(role="agent", parts=[TextPart(text="Hi there!")]),
    ],
    artifacts=[
        Artifact(
            artifact_id="art-001",
            parts=[TextPart(text="Response content")],
        )
    ],
)

Task Fields

Field	Type	Description
`id`	`str`	Unique task ID (auto-generated UUID)
`context_id`	`str`	Conversation context ID
`status`	`TaskStatus`	Current state + optional message
`history`	`List[Message]`	Message exchange history
`artifacts`	`List[Artifact]`	Output artifacts from agent

JSON-RPC Methods

tasks/get — Retrieve a Task

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/get",
    "id": "1",
    "params": {
      "id": "task-uuid-123"
    }
  }'

Response:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "id": "task-uuid-123",
    "contextId": "ctx-uuid-456",
    "status": {"state": "completed"},
    "history": [
      {"role": "user", "parts": [{"text": "Hello"}]},
      {"role": "agent", "parts": [{"text": "Hi there!"}]}
    ],
    "artifacts": [
      {"artifactId": "art-001", "parts": [{"text": "Response"}]}
    ]
  }
}

Task not found:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "error": {"code": -32000, "message": "Task not found: task-uuid-123"}
}

tasks/list — List Tasks

The task store supports listing tasks with optional context filtering:

python

# From the task store (server-side only)
from praisonaiagents.ui.a2a.task_store import TaskStore

store = TaskStore()

# List all tasks
all_tasks = store.list_tasks()

# Filter by context
context_tasks = store.list_tasks(context_id="ctx-uuid-456")

Planned JSON-RPC Interface:

bash

# Future implementation
curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0", 
    "method": "tasks/list",
    "id": "1",
    "params": {
      "contextId": "ctx-uuid-456"
    }
  }'

tasks/cancel — Cancel a Task

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/cancel",
    "id": "1",
    "params": {
      "id": "task-uuid-123"
    }
  }'

Response:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "id": "task-uuid-123",
    "status": {"state": "cancelled"}
  }
}

tasks/cancel also works on in-flight message/stream tasks — the stream stops and ends with a final cancelled status event.

Response when the task is already terminal (completed, failed, or cancelled) — the state is left unchanged:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "error": {
    "code": -32002,
    "message": "Task not cancelable: already completed"
  }
}

Python Client Implementation

python

import aiohttp
import json

class SimpleA2AClient:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/') + '/a2a'

    async def send_message(self, text: str, context_id: str = None):
        payload = {
            "jsonrpc": "2.0",
            "method": "message/send", 
            "id": "1",
            "params": {
                "message": {
                    "role": "user",
                    "parts": [{"text": text}],
                    "contextId": context_id
                }
            }
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(self.base_url, json=payload) as response:
                return await response.json()

    async def get_task(self, task_id: str):
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/get",
            "id": "2", 
            "params": {"id": task_id}
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(self.base_url, json=payload) as response:
                return await response.json()

    async def cancel_task(self, task_id: str):
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/cancel",
            "id": "3",
            "params": {"id": task_id}
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(self.base_url, json=payload) as response:
                return await response.json()

# Usage
async def main():
    client = SimpleA2AClient("http://localhost:8000")

    # Send a message (creates a task)
    result = await client.send_message("Analyze this data")
    task_id = result["result"]["id"]
    context_id = result["result"]["contextId"]

    # Get task status
    task = await client.get_task(task_id)
    print(f"State: {task['result']['status']['state']}")

    # Cancel a task
    cancelled = await client.cancel_task(task_id)
    print(f"Cancelled: {cancelled['result']['status']['state']}")

Task Store

The A2A server maintains an in-memory task store that tracks all tasks:

Diagram

Tasks are automatically created when message/send or message/stream is called
Streamed tasks follow the same lifecycle as message/send: they move working → completed | failed | cancelled; on success the final response is appended to history and the artifact is stored on the task. The task stays in the store while the stream is running.
Each task gets a unique UUID
Tasks within the same conversation share a contextId
Task history includes both user messages and agent responses
Artifacts contain the agent’s output
Store is per-server-instance and does not persist across restarts
The store holds up to 10,000 tasks; once that cap is reached, the oldest terminal tasks are evicted first. Tasks whose response is still being produced are pinned and never evicted.

Context-Based Conversations

Tasks can be grouped by contextId for multi-turn conversations:

python

client = SimpleA2AClient("http://localhost:8000")

# First message — new context created  
r1 = await client.send_message("What is Python?")
ctx = r1["result"]["contextId"]

# Follow-up — same context
r2 = await client.send_message("What about its GIL?", context_id=ctx)

# Both tasks now share the same contextId for conversation tracking
print(f"Context ID: {ctx}")
print(f"First task: {r1['result']['id']}")
print(f"Second task: {r2['result']['id']}")

Error Handling

Common error codes returned by task management methods:

Error Code	Message	When
`-32602`	Invalid params: ‘id’ required	Missing task ID
`-32000`	Task not found	Task ID doesn’t exist
`-32002`	Task not cancelable: already `<state>`	`tasks/cancel` on a task whose state is `completed`, `failed`, or `cancelled`
`-32601`	Method not found	Unknown JSON-RPC method

Error Response Format:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "error": {
    "code": -32000,
    "message": "Task not found: task-uuid-123"
  }
}

Best Practices

Task State Monitoring

Always check task state before assuming completion. Tasks can fail or require input.

python

client = SimpleA2AClient("http://localhost:8000")
task = await client.get_task(task_id)
state = task["result"]["status"]["state"]

if state == "completed":
    # Process results
elif state == "failed":
    # Handle error
elif state == "input_required":
    # Provide additional input

Context Management

Use context IDs to group related tasks and maintain conversation history.

python

# Keep context for follow-up questions
result = await client.send_message("Initial question")
context_id = result["result"]["contextId"]
follow_up = await client.send_message("Follow-up question", context_id=context_id)

Cleanup

Cancel unnecessary tasks to free resources and avoid confusion.

python

# Cancel long-running tasks when no longer needed
await client.cancel_task(task_id)

Error Handling

Handle JSON-RPC errors and HTTP errors gracefully.

python

async def safe_send_message(client, text):
    try:
        response = await client.send_message(text)
        if "error" in response:
            print(f"A2A Error: {response['error']['message']}")
        return response
    except aiohttp.ClientError as e:
        print(f"HTTP Error: {e}")
        return None

Terminal states are final

Once a task reaches completed, failed, or cancelled, its status cannot change. tasks/cancel on a terminal task returns -32002 and the state is unchanged. Treat these three states as sinks in your client.

A2A Protocol

Server setup and configuration

Handoffs

Multi-agent communication patterns

A2A Task Management

How It Works

Quick Start

Task Lifecycle

Task States

Python Task States

Task Structure

Task Fields

JSON-RPC Methods

tasks/get — Retrieve a Task

tasks/list — List Tasks

tasks/cancel — Cancel a Task

Python Client Implementation

Task Store

Context-Based Conversations

Error Handling

Best Practices

Related