Agent2Agent Protocol (A2A)

# A2A Task Lifecycle

# 1. CLIENT SENDS TASK
task = {
    id: generateUUID(),
    sessionId: "session-123",  # For multi-turn conversations
    message: {
        role: "user",
        parts: [
            { type: "text", text: "Research the latest AI safety papers" }
        ]
    },
    metadata: {
        priority: "high",
        deadline: "2025-01-27T00:00:00Z"
    }
}

response = POST(agentUrl + "/tasks/send", task)

# 2. SERVER ACKNOWLEDGES AND STARTS PROCESSING
# Response includes task status
assert response.status == "submitted" or response.status == "working"

# 3. CLIENT POLLS FOR UPDATES (or uses streaming)
while task.status not in ["completed", "failed", "canceled"]:
    statusResponse = GET(agentUrl + "/tasks/" + task.id)
    task = statusResponse.task

    if task.status == "input-required":
        # Agent needs more information
        userInput = getUserInput(task.message)
        POST(agentUrl + "/tasks/" + task.id + "/send", userInput)

# 4. GET FINAL RESULT
if task.status == "completed":
    artifacts = task.artifacts
    for artifact in artifacts:
        print(artifact.type, artifact.content)

import httpx
import asyncio
from uuid import uuid4
from dataclasses import dataclass
from enum import Enum

class TaskStatus(Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

@dataclass
class A2ATask:
    id: str
    status: TaskStatus
    message: dict | None = None
    artifacts: list | None = None
    history: list | None = None

class A2AClient:
    def __init__(self, agent_url: str, auth_token: str):
        self.agent_url = agent_url.rstrip("/")
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {auth_token}"}
        )

    async def send_task(
        self,
        message: str,
        session_id: str | None = None,
        metadata: dict | None = None
    ) -> A2ATask:
        """Send a new task to the agent."""
        task_data = {
            "id": str(uuid4()),
            "sessionId": session_id or str(uuid4()),
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": message}]
            }
        }
        if metadata:
            task_data["metadata"] = metadata

        response = await self.client.post(
            f"{self.agent_url}/tasks/send",
            json=task_data
        )
        response.raise_for_status()
        return self._parse_task(response.json())

    async def get_task_status(self, task_id: str) -> A2ATask:
        """Get current status of a task."""
        response = await self.client.get(
            f"{self.agent_url}/tasks/{task_id}"
        )
        response.raise_for_status()
        return self._parse_task(response.json())

    async def send_input(
        self,
        task_id: str,
        message: str
    ) -> A2ATask:
        """Send additional input for a task that requires it."""
        response = await self.client.post(
            f"{self.agent_url}/tasks/{task_id}/send",
            json={
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": message}]
                }
            }
        )
        response.raise_for_status()
        return self._parse_task(response.json())

    async def wait_for_completion(
        self,
        task_id: str,
        poll_interval: float = 1.0,
        timeout: float = 300.0
    ) -> A2ATask:
        """Poll until task completes or times out."""
        elapsed = 0.0

        while elapsed < timeout:
            task = await self.get_task_status(task_id)

            if task.status in [
                TaskStatus.COMPLETED,
                TaskStatus.FAILED,
                TaskStatus.CANCELED
            ]:
                return task

            if task.status == TaskStatus.INPUT_REQUIRED:
                raise InterruptedError(
                    f"Task requires input: {task.message}"
                )

            await asyncio.sleep(poll_interval)
            elapsed += poll_interval

        raise TimeoutError(f"Task {task_id} did not complete")

    def _parse_task(self, data: dict) -> A2ATask:
        return A2ATask(
            id=data["id"],
            status=TaskStatus(data["status"]),
            message=data.get("message"),
            artifacts=data.get("artifacts"),
            history=data.get("history")
        )

# Usage
async def main():
    client = A2AClient(
        agent_url="https://api.example.com/agents/research",
        auth_token="your-token"
    )

    # Send a task
    task = await client.send_task(
        "Research the latest developments in AI safety"
    )
    print(f"Task submitted: {task.id}")

    # Wait for completion
    result = await client.wait_for_completion(task.id)

    if result.status == TaskStatus.COMPLETED:
        for artifact in result.artifacts or []:
            print(f"Artifact: {artifact}")

asyncio.run(main())

using System.Net.Http.Json;
using System.Text.Json;

public enum TaskStatus
{
    Submitted,
    Working,
    InputRequired,
    Completed,
    Failed,
    Canceled
}

public record A2ATask(
    string Id,
    TaskStatus Status,
    JsonElement? Message = null,
    JsonElement[]? Artifacts = null,
    JsonElement[]? History = null
);

public class A2AClient : IDisposable
{
    private readonly HttpClient _client;
    private readonly string _agentUrl;

    public A2AClient(string agentUrl, string authToken)
    {
        _agentUrl = agentUrl.TrimEnd('/');
        _client = new HttpClient();
        _client.DefaultRequestHeaders.Add("Authorization", $"Bearer {authToken}");
    }

    public async Task<A2ATask> SendTaskAsync(
        string message,
        string? sessionId = null,
        Dictionary<string, object>? metadata = null,
        CancellationToken ct = default)
    {
        var taskData = new
        {
            id = Guid.NewGuid().ToString(),
            sessionId = sessionId ?? Guid.NewGuid().ToString(),
            message = new
            {
                role = "user",
                parts = new[] { new { type = "text", text = message } }
            },
            metadata
        };

        var response = await _client.PostAsJsonAsync(
            $"{_agentUrl}/tasks/send",
            taskData,
            ct
        );
        response.EnsureSuccessStatusCode();

        return await response.Content.ReadFromJsonAsync<A2ATask>(ct)
            ?? throw new InvalidOperationException("Invalid response");
    }

    public async Task<A2ATask> GetTaskStatusAsync(
        string taskId,
        CancellationToken ct = default)
    {
        var response = await _client.GetAsync(
            $"{_agentUrl}/tasks/{taskId}",
            ct
        );
        response.EnsureSuccessStatusCode();

        return await response.Content.ReadFromJsonAsync<A2ATask>(ct)
            ?? throw new InvalidOperationException("Invalid response");
    }

    public async Task<A2ATask> WaitForCompletionAsync(
        string taskId,
        TimeSpan? pollInterval = null,
        TimeSpan? timeout = null,
        CancellationToken ct = default)
    {
        var interval = pollInterval ?? TimeSpan.FromSeconds(1);
        var maxWait = timeout ?? TimeSpan.FromMinutes(5);
        var elapsed = TimeSpan.Zero;

        while (elapsed < maxWait)
        {
            var task = await GetTaskStatusAsync(taskId, ct);

            if (task.Status is TaskStatus.Completed
                or TaskStatus.Failed
                or TaskStatus.Canceled)
            {
                return task;
            }

            if (task.Status == TaskStatus.InputRequired)
            {
                throw new InvalidOperationException(
                    $"Task requires input: {task.Message}"
                );
            }

            await Task.Delay(interval, ct);
            elapsed += interval;
        }

        throw new TimeoutException($"Task {taskId} did not complete");
    }

    public void Dispose() => _client.Dispose();
}

// Usage
await using var client = new A2AClient(
    "https://api.example.com/agents/research",
    "your-token"
);

var task = await client.SendTaskAsync(
    "Research the latest developments in AI safety"
);
Console.WriteLine($"Task submitted: {task.Id}");

var result = await client.WaitForCompletionAsync(task.Id);
if (result.Status == TaskStatus.Completed)
{
    foreach (var artifact in result.Artifacts ?? Array.Empty<JsonElement>())
    {
        Console.WriteLine($"Artifact: {artifact}");
    }
}