Configuration

# Each of these is a normal request; firing more than AgentMaxConcurrency
# of them at once against one agent makes the overflow return -32603.
curl -X POST http://localhost:9090/a2a/echo-agent-01 \
  -H 'Content-Type: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": { "message": { "parts": [{ "text": "hello" }] } }
  }'

// .kb/integration-a2a/examples/csharp/limits/concurrency-limit/Client.cs
using System.Net.Http.Json;

var kubemqUrl = Environment.GetEnvironmentVariable("KUBEMQ_URL") ?? "http://localhost:9090";
var agentId = "concurrency-agent-01";
const int total = 101; // limit is 100; 1 request will be rejected

using var http = new HttpClient();

async Task<(bool ok, int? code)> SendOne(int i)
{
    var payload = new
    {
        jsonrpc = "2.0",
        id = i,
        method = "message/send",
        @params = new { message = new { parts = new[] { new { text = $"req-{i}" } } } }
    };
    var resp = await http.PostAsJsonAsync($"{kubemqUrl}/a2a/{agentId}", payload);
    var body = await resp.Content.ReadFromJsonAsync<JsonRpcResponse>();
    if (body?.Error is not null)
        return (false, body.Error.Code);
    return (true, null);
}

var results = await Task.WhenAll(Enumerable.Range(0, total).Select(SendOne));
var rejected = results.Count(r => r.code == -32603);
Console.WriteLine($"Successes: {results.Count(r => r.ok)}  Rejected (-32603): {rejected}");

// .kb/integration-a2a/examples/go/limits/concurrency-limit/client.go
package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "os"
    "sync"
    "sync/atomic"
)

func main() {
    kubemqURL := os.Getenv("KUBEMQ_URL")
    if kubemqURL == "" {
        kubemqURL = "http://localhost:9090"
    }
    agentID := "concurrency-agent-01"
    const total = 101 // limit is 100; 1 request is rejected

    var success, rejected int64
    var wg sync.WaitGroup
    for i := 0; i < total; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            body, _ := json.Marshal(map[string]any{
                "jsonrpc": "2.0", "id": i, "method": "message/send",
                "params": map[string]any{"message": map[string]any{
                    "parts": []map[string]string{{"text": fmt.Sprintf("req-%d", i)}}}},
            })
            resp, err := http.Post(kubemqURL+"/a2a/"+agentID, "application/json", bytes.NewReader(body))
            if err != nil {
                return
            }
            defer resp.Body.Close()
            var out struct {
                Error *struct {
                    Code int `json:"code"`
                } `json:"error"`
            }
            _ = json.NewDecoder(resp.Body).Decode(&out)
            if out.Error != nil && out.Error.Code == -32603 {
                atomic.AddInt64(&rejected, 1)
            } else {
                atomic.AddInt64(&success, 1)
            }
        }(i)
    }
    wg.Wait()
    fmt.Printf("Successes: %d  Rejected (-32603): %d\n", success, rejected)
}

// .kb/integration-a2a/examples/java/limits/concurrency-limit/Client.java
import java.net.URI;
import java.net.http.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

public class Client {
    public static void main(String[] args) throws Exception {
        String kubemqUrl = System.getenv().getOrDefault("KUBEMQ_URL", "http://localhost:9090");
        String agentId = "concurrency-agent-01";
        int total = 101; // limit is 100; 1 request is rejected

        HttpClient http = HttpClient.newHttpClient();
        AtomicInteger success = new AtomicInteger();
        AtomicInteger rejected = new AtomicInteger();
        ExecutorService pool = Executors.newFixedThreadPool(total);

        var tasks = IntStream.range(0, total).<Callable<Void>>mapToObj(i -> () -> {
            String payload = """
                {"jsonrpc":"2.0","id":%d,"method":"message/send",
                 "params":{"message":{"parts":[{"text":"req-%d"}]}}}""".formatted(i, i);
            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(kubemqUrl + "/a2a/" + agentId))
                .header("Content-Type", "application/json")
                .POST(HttpRequest.BodyPublishers.ofString(payload))
                .build();
            String body = http.send(req, HttpResponse.BodyHandlers.ofString()).body();
            if (body.contains("-32603")) rejected.incrementAndGet();
            else success.incrementAndGet();
            return null;
        }).toList();

        pool.invokeAll(tasks);
        pool.shutdown();
        System.out.printf("Successes: %d  Rejected (-32603): %d%n", success.get(), rejected.get());
    }
}

# .kb/integration-a2a/examples/python/limits/concurrency-limit/client.py
import asyncio
import os

import httpx

KUBEMQ_URL = os.getenv("KUBEMQ_URL", "http://localhost:9090")
AGENT_ID = "concurrency-agent-01"
TOTAL = 101  # limit is 100; 1 request is rejected


async def send_one(client: httpx.AsyncClient, i: int) -> int | None:
    payload = {
        "jsonrpc": "2.0",
        "id": i,
        "method": "message/send",
        "params": {"message": {"parts": [{"text": f"req-{i}"}]}},
    }
    resp = await client.post(f"{KUBEMQ_URL}/a2a/{AGENT_ID}", json=payload)
    error = resp.json().get("error")
    return error["code"] if error else None


async def main() -> None:
    async with httpx.AsyncClient(timeout=30) as client:
        codes = await asyncio.gather(*(send_one(client, i) for i in range(TOTAL)))
    rejected = sum(1 for c in codes if c == -32603)
    successes = sum(1 for c in codes if c is None)
    print(f"Successes: {successes}  Rejected (-32603): {rejected}")


if __name__ == "__main__":
    asyncio.run(main())

// .kb/integration-a2a/examples/typescript/limits/concurrency-limit/client.ts
const kubemqUrl = process.env.KUBEMQ_URL ?? "http://localhost:9090";
const agentId = "concurrency-agent-01";
const total = 101; // limit is 100; 1 request is rejected

async function sendOne(i: number): Promise<number | null> {
  const resp = await fetch(`${kubemqUrl}/a2a/${agentId}`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      jsonrpc: "2.0",
      id: i,
      method: "message/send",
      params: { message: { parts: [{ text: `req-${i}` }] } },
    }),
  });
  const body = (await resp.json()) as { error?: { code: number } };
  return body.error?.code ?? null;
}

const codes = await Promise.all(
  Array.from({ length: total }, (_, i) => sendOne(i)),
);
const rejected = codes.filter((c) => c === -32603).length;
const successes = codes.filter((c) => c === null).length;
console.log(`Successes: ${successes}  Rejected (-32603): ${rejected}`);

# A normal request; the cap is hit only when the agent's reply exceeds
# AgentMaxResponseBytes, in which case the response carries error -32603.
curl -X POST http://localhost:9090/a2a/oversize-agent-01 \
  -H 'Content-Type: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": { "message": { "parts": [{ "text": "return a big payload" }] } }
  }'

// .kb/integration-a2a/examples/csharp/limits/response-size/Client.cs
using System.Net.Http.Json;

var kubemqUrl = Environment.GetEnvironmentVariable("KUBEMQ_URL") ?? "http://localhost:9090";
var agentId = "oversize-agent-01";

using var http = new HttpClient();
var payload = new
{
    jsonrpc = "2.0",
    id = 1,
    method = "message/send",
    @params = new { message = new { parts = new[] { new { text = "return a big payload" } } } }
};

var resp = await http.PostAsJsonAsync($"{kubemqUrl}/a2a/{agentId}", payload);
var body = await resp.Content.ReadFromJsonAsync<JsonRpcResponse>();
if (body?.Error is not null)
    Console.WriteLine($"Error code: {body.Error.Code}  Message: {body.Error.Message}");
else
    Console.WriteLine("Response within the size limit.");

// .kb/integration-a2a/examples/go/limits/response-size/client.go
package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "os"
)

func main() {
    kubemqURL := os.Getenv("KUBEMQ_URL")
    if kubemqURL == "" {
        kubemqURL = "http://localhost:9090"
    }
    agentID := "oversize-agent-01"

    body, _ := json.Marshal(map[string]any{
        "jsonrpc": "2.0", "id": 1, "method": "message/send",
        "params": map[string]any{"message": map[string]any{
            "parts": []map[string]string{{"text": "return a big payload"}}}},
    })
    resp, err := http.Post(kubemqURL+"/a2a/"+agentID, "application/json", bytes.NewReader(body))
    if err != nil {
        panic(err)
    }
    defer resp.Body.Close()

    var out struct {
        Error *struct {
            Code    int    `json:"code"`
            Message string `json:"message"`
        } `json:"error"`
    }
    _ = json.NewDecoder(resp.Body).Decode(&out)
    if out.Error != nil {
        fmt.Printf("Error code: %d  Message: %s\n", out.Error.Code, out.Error.Message)
    } else {
        fmt.Println("Response within the size limit.")
    }
}

// .kb/integration-a2a/examples/java/limits/response-size/Client.java
import java.net.URI;
import java.net.http.*;

public class Client {
    public static void main(String[] args) throws Exception {
        String kubemqUrl = System.getenv().getOrDefault("KUBEMQ_URL", "http://localhost:9090");
        String agentId = "oversize-agent-01";

        String payload = """
            {"jsonrpc":"2.0","id":1,"method":"message/send",
             "params":{"message":{"parts":[{"text":"return a big payload"}]}}}""";

        HttpClient http = HttpClient.newHttpClient();
        HttpRequest req = HttpRequest.newBuilder()
            .uri(URI.create(kubemqUrl + "/a2a/" + agentId))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(payload))
            .build();

        String body = http.send(req, HttpResponse.BodyHandlers.ofString()).body();
        if (body.contains("-32603"))
            System.out.println("Error code: -32603  (response too large)");
        else
            System.out.println("Response within the size limit.");
    }
}

# .kb/integration-a2a/examples/python/limits/response-size/client.py
import os

import httpx

KUBEMQ_URL = os.getenv("KUBEMQ_URL", "http://localhost:9090")
AGENT_ID = "oversize-agent-01"


def main() -> None:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "message/send",
        "params": {"message": {"parts": [{"text": "return a big payload"}]}},
    }
    resp = httpx.post(f"{KUBEMQ_URL}/a2a/{AGENT_ID}", json=payload, timeout=30)
    error = resp.json().get("error")
    if error:
        print(f"Error code: {error['code']}  Message: {error['message']}")
    else:
        print("Response within the size limit.")


if __name__ == "__main__":
    main()

// .kb/integration-a2a/examples/typescript/limits/response-size/client.ts
const kubemqUrl = process.env.KUBEMQ_URL ?? "http://localhost:9090";
const agentId = "oversize-agent-01";

const resp = await fetch(`${kubemqUrl}/a2a/${agentId}`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    jsonrpc: "2.0",
    id: 1,
    method: "message/send",
    params: { message: { parts: [{ text: "return a big payload" }] } },
  }),
});

const body = (await resp.json()) as {
  error?: { code: number; message: string };
};
if (body.error) {
  console.log(`Error code: ${body.error.code}  Message: ${body.error.message}`);
} else {
  console.log("Response within the size limit.");
}