Agent2Agent Protocol (A2A)

# A2A Streaming with Server-Sent Events (SSE)

function streamTask(agentUrl, task):
    # Open SSE connection
    eventSource = new EventSource(
        agentUrl + "/tasks/sendSubscribe",
        method: POST,
        body: task,
        headers: { "Content-Type": "application/json" }
    )

    eventSource.onMessage = (event) =>
        data = JSON.parse(event.data)

        switch data.type:
            case "status":
                print("Status:", data.status)

            case "artifact":
                # Partial artifact (streaming content)
                if data.append:
                    currentArtifact += data.content
                else:
                    artifacts.append(data)

            case "message":
                # Agent message (e.g., thinking, asking for input)
                print("Agent:", data.content)

            case "done":
                # Task completed
                eventSource.close()
                return { status: data.status, artifacts: artifacts }

            case "error":
                eventSource.close()
                throw Error(data.message)

    eventSource.onError = (error) =>
        eventSource.close()
        throw error

import httpx
from typing import AsyncIterator
from dataclasses import dataclass

@dataclass
class StreamEvent:
    type: str
    data: dict

async def stream_task(
    agent_url: str,
    message: str,
    auth_token: str,
    session_id: str | None = None
) -> AsyncIterator[StreamEvent]:
    """Stream task results using Server-Sent Events."""

    task_data = {
        "id": str(uuid4()),
        "sessionId": session_id or str(uuid4()),
        "message": {
            "role": "user",
            "parts": [{"type": "text", "text": message}]
        }
    }

    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            f"{agent_url}/tasks/sendSubscribe",
            json=task_data,
            headers={
                "Authorization": f"Bearer {auth_token}",
                "Accept": "text/event-stream"
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data = json.loads(line[6:])
                    yield StreamEvent(
                        type=data.get("type", "unknown"),
                        data=data
                    )

                    if data.get("type") in ["done", "error"]:
                        return

# Usage with streaming
async def research_with_streaming():
    full_response = ""

    async for event in stream_task(
        agent_url="https://api.example.com/agents/research",
        message="Research quantum computing breakthroughs in 2025",
        auth_token="your-token"
    ):
        if event.type == "status":
            print(f"Status: {event.data['status']}")

        elif event.type == "artifact":
            if event.data.get("streaming"):
                # Streaming text content
                chunk = event.data.get("content", "")
                print(chunk, end="", flush=True)
                full_response += chunk
            else:
                # Complete artifact
                print(f"\nArtifact: {event.data}")

        elif event.type == "message":
            print(f"Agent: {event.data.get('content')}")

        elif event.type == "done":
            print(f"\nCompleted with status: {event.data['status']}")

        elif event.type == "error":
            print(f"Error: {event.data.get('message')}")
            raise Exception(event.data.get("message"))

asyncio.run(research_with_streaming())

using System.Net.Http.Json;
using System.Runtime.CompilerServices;

public record StreamEvent(string Type, JsonElement Data);

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

    public A2AStreamingClient(string agentUrl, string authToken)
    {
        _agentUrl = agentUrl.TrimEnd('/');
        _client = new HttpClient { Timeout = TimeSpan.FromMinutes(10) };
        _client.DefaultRequestHeaders.Add("Authorization", $"Bearer {authToken}");
        _client.DefaultRequestHeaders.Add("Accept", "text/event-stream");
    }

    public async IAsyncEnumerable<StreamEvent> StreamTaskAsync(
        string message,
        string? sessionId = null,
        [EnumeratorCancellation] 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 } }
            }
        };

        using var request = new HttpRequestMessage(
            HttpMethod.Post,
            $"{_agentUrl}/tasks/sendSubscribe"
        );
        request.Content = JsonContent.Create(taskData);

        using var response = await _client.SendAsync(
            request,
            HttpCompletionOption.ResponseHeadersRead,
            ct
        );
        response.EnsureSuccessStatusCode();

        await using var stream = await response.Content.ReadAsStreamAsync(ct);
        using var reader = new StreamReader(stream);

        while (!reader.EndOfStream && !ct.IsCancellationRequested)
        {
            var line = await reader.ReadLineAsync(ct);

            if (string.IsNullOrEmpty(line) || !line.StartsWith("data: "))
                continue;

            var jsonData = line.Substring(6);
            var data = JsonDocument.Parse(jsonData).RootElement;
            var eventType = data.GetProperty("type").GetString() ?? "unknown";

            yield return new StreamEvent(eventType, data);

            if (eventType is "done" or "error")
                yield break;
        }
    }

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

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

var fullResponse = new StringBuilder();

await foreach (var evt in client.StreamTaskAsync(
    "Research quantum computing breakthroughs in 2025"
))
{
    switch (evt.Type)
    {
        case "status":
            Console.WriteLine($"Status: {evt.Data.GetProperty("status")}");
            break;

        case "artifact":
            if (evt.Data.TryGetProperty("streaming", out var streaming)
                && streaming.GetBoolean())
            {
                var chunk = evt.Data.GetProperty("content").GetString();
                Console.Write(chunk);
                fullResponse.Append(chunk);
            }
            break;

        case "done":
            Console.WriteLine($"\nCompleted: {evt.Data.GetProperty("status")}");
            break;

        case "error":
            throw new Exception(evt.Data.GetProperty("message").GetString());
    }
}

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from enum import Enum
import asyncio
import json

app = FastAPI()

# In-memory task storage (use Redis/DB in production)
tasks: dict[str, dict] = {}

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

class TaskMessage(BaseModel):
    role: str
    parts: list[dict]

class TaskRequest(BaseModel):
    id: str
    sessionId: str
    message: TaskMessage
    metadata: dict | None = None

# Agent Card endpoint
@app.get("/.well-known/agent.json")
async def get_agent_card():
    return {
        "name": "research-assistant",
        "description": "AI research assistant",
        "url": "https://api.example.com/agents/research",
        "version": "1.0.0",
        "capabilities": {
            "streaming": True,
            "pushNotifications": False
        },
        "skills": [
            {
                "id": "web-search",
                "name": "Web Search",
                "description": "Search the web for information"
            }
        ]
    }

# Send task endpoint
@app.post("/tasks/send")
async def send_task(
    task: TaskRequest,
    background_tasks: BackgroundTasks
):
    # Store task
    tasks[task.id] = {
        "id": task.id,
        "sessionId": task.sessionId,
        "status": TaskStatus.SUBMITTED,
        "message": task.message.dict(),
        "artifacts": []
    }

    # Process in background
    background_tasks.add_task(process_task, task.id)

    return tasks[task.id]

# Get task status
@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    if task_id not in tasks:
        raise HTTPException(404, "Task not found")
    return tasks[task_id]

# Streaming endpoint
@app.post("/tasks/sendSubscribe")
async def send_task_streaming(task: TaskRequest):
    # Store task
    tasks[task.id] = {
        "id": task.id,
        "status": TaskStatus.SUBMITTED,
        "message": task.message.dict()
    }

    async def event_stream():
        # Send status update
        yield f"data: {json.dumps({'type': 'status', 'status': 'working'})}\n\n"

        # Simulate streaming response
        response_text = await generate_response(task.message.parts[0]["text"])

        # Stream response in chunks
        for chunk in response_text.split(" "):
            yield f"data: {json.dumps({'type': 'artifact', 'streaming': True, 'content': chunk + ' '})}\n\n"
            await asyncio.sleep(0.05)

        # Send completion
        yield f"data: {json.dumps({'type': 'done', 'status': 'completed'})}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

async def process_task(task_id: str):
    """Background task processing."""
    tasks[task_id]["status"] = TaskStatus.WORKING

    try:
        # Get the user message
        message = tasks[task_id]["message"]["parts"][0]["text"]

        # Generate response (your agent logic here)
        response = await generate_response(message)

        # Store result
        tasks[task_id]["artifacts"] = [
            {"type": "text", "content": response}
        ]
        tasks[task_id]["status"] = TaskStatus.COMPLETED

    except Exception as e:
        tasks[task_id]["status"] = TaskStatus.FAILED
        tasks[task_id]["error"] = str(e)

async def generate_response(message: str) -> str:
    """Your agent logic here."""
    # This is where you'd call your LLM, tools, etc.
    await asyncio.sleep(1)  # Simulate processing
    return f"Research results for: {message}"

using Microsoft.AspNetCore.Mvc;
using System.Text.Json;
using System.Collections.Concurrent;

[ApiController]
[Route("")]
public class A2AAgentController : ControllerBase
{
    private static readonly ConcurrentDictionary<string, AgentTask> Tasks = new();

    // Agent Card endpoint
    [HttpGet(".well-known/agent.json")]
    public IActionResult GetAgentCard()
    {
        return Ok(new
        {
            name = "research-assistant",
            description = "AI research assistant",
            url = "https://api.example.com/agents/research",
            version = "1.0.0",
            capabilities = new { streaming = true },
            skills = new[]
            {
                new {
                    id = "web-search",
                    name = "Web Search",
                    description = "Search the web for information"
                }
            }
        });
    }

    // Send task endpoint
    [HttpPost("tasks/send")]
    public async Task<IActionResult> SendTask(
        [FromBody] TaskRequest request,
        [FromServices] IBackgroundTaskQueue taskQueue)
    {
        var task = new AgentTask
        {
            Id = request.Id,
            SessionId = request.SessionId,
            Status = TaskStatus.Submitted,
            Message = request.Message
        };

        Tasks[task.Id] = task;

        // Queue background processing
        await taskQueue.QueueAsync(async ct =>
        {
            await ProcessTaskAsync(task.Id, ct);
        });

        return Ok(task);
    }

    // Get task status
    [HttpGet("tasks/{taskId}")]
    public IActionResult GetTask(string taskId)
    {
        if (!Tasks.TryGetValue(taskId, out var task))
            return NotFound();

        return Ok(task);
    }

    // Streaming endpoint
    [HttpPost("tasks/sendSubscribe")]
    public async Task SendTaskStreaming(
        [FromBody] TaskRequest request,
        CancellationToken ct)
    {
        Response.ContentType = "text/event-stream";
        Response.Headers.CacheControl = "no-cache";

        var writer = Response.BodyWriter;

        // Send status
        await WriteEventAsync(writer, new { type = "status", status = "working" });

        // Generate response (your agent logic)
        var message = request.Message.Parts[0].Text;
        var response = await GenerateResponseAsync(message, ct);

        // Stream response
        foreach (var word in response.Split(' '))
        {
            await WriteEventAsync(writer, new
            {
                type = "artifact",
                streaming = true,
                content = word + " "
            });
            await Task.Delay(50, ct);
        }

        // Send completion
        await WriteEventAsync(writer, new { type = "done", status = "completed" });
    }

    private static async Task WriteEventAsync(
        System.IO.Pipelines.PipeWriter writer,
        object data)
    {
        var json = JsonSerializer.Serialize(data);
        var bytes = System.Text.Encoding.UTF8.GetBytes($"data: {json}\n\n");
        await writer.WriteAsync(bytes);
        await writer.FlushAsync();
    }

    private static async Task ProcessTaskAsync(string taskId, CancellationToken ct)
    {
        if (!Tasks.TryGetValue(taskId, out var task))
            return;

        task.Status = TaskStatus.Working;

        try
        {
            var message = task.Message.Parts[0].Text;
            var response = await GenerateResponseAsync(message, ct);

            task.Artifacts = new[]
            {
                new { type = "text", content = response }
            };
            task.Status = TaskStatus.Completed;
        }
        catch (Exception ex)
        {
            task.Status = TaskStatus.Failed;
            task.Error = ex.Message;
        }
    }

    private static async Task<string> GenerateResponseAsync(
        string message,
        CancellationToken ct)
    {
        await Task.Delay(1000, ct);
        return $"Research results for: {message}";
    }
}