[PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi opened a new pull request, #23711:
URL: https://github.com/apache/camel/pull/23711

   …l integration
   
   Adds the `camel-a2a` component implementing the [A2A (Agent-to-Agent) protocol](https://a2a-protocol.org/latest/) for Apache Camel. The component provides both producer and consumer support, enabling Camel
     routes to participate in standardized agent-to-agent communication.
   
     Features
   
     - **Consumer** — exposes a Camel route as an A2A server with automatic agent card serving at `/.well-known/agent-card.json`
     - **Producer** — sends A2A messages (MESSAGE_SEND, MESSAGE_STREAM, TASK_GET, TASK_CANCEL, etc.) to remote agents
     - **Protocol bindings** — REST and JSON-RPC 2.0
     - **Streaming** — Server-Sent Events (SSE) for real-time task updates
     - **Authentication** — API key, Bearer token, OAuth 2.0, OpenID Connect
     - **Task management** — in-memory task store with TTL, push notifications, subscriber management
     - **Agent card** — discovery, resolution (file < bean < URI params), and auto-serving
     - **Type converters** — Message/Task to String, plus Simple DSL functions (`a2aText()`, `a2aData()`, `a2aFile()`)
     - **Progress API** — `A2AProgress.emit()` for status updates from any route
   
     **Demo**
   
     A multi-agent demo showcasing `camel-a2a` in action is available at:
   
   [https://github.com/luigidemasi/camel-a2a-morning-routine](https://github.com/luigidemasi/camel-a2a-morning-routine)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["davsclaus (via GitHub)" <gi...@apache.org>]

davsclaus commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348769217


##########
components/camel-ai/camel-a2a/src/generated/resources/a2a.json:
##########
@@ -0,0 +1,15 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "a2a",
+    "title": "A2a",
+    "description": "Camel A2A component for Agent-to-Agent protocol communication",
+    "deprecated": false,
+    "firstVersion": "4.21.0",
+    "label": "ai,a2a",

Review Comment:
   these should be generated from code via the Category.XXX you set in the endpoint class



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["orpiske (via GitHub)" <gi...@apache.org>]

orpiske commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3347387364


##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java:
##########
@@ -0,0 +1,1278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.a2a.model.A2AError;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.Message;
+import org.apache.camel.component.a2a.model.SendMessageRequest;
+import org.apache.camel.component.a2a.model.SendMessageResponse;
+import org.apache.camel.component.a2a.model.StreamResponse;
+import org.apache.camel.component.a2a.model.Task;
+import org.apache.camel.component.a2a.model.TaskPushNotificationConfig;
+import org.apache.camel.component.a2a.model.TaskState;
+import org.apache.camel.component.a2a.model.TaskStatus;
+import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent;
+import org.apache.camel.component.a2a.operation.A2AOperations;
+import org.apache.camel.component.a2a.operation.MessageSendOperation;
+import org.apache.camel.component.a2a.protocol.JsonRpcProtocol;
+import org.apache.camel.component.a2a.state.A2ATaskStore;
+import org.apache.camel.component.a2a.state.A2ATaskSubscriber;
+import org.apache.camel.component.a2a.streaming.DefaultStreamEmitter;
+import org.apache.camel.component.a2a.streaming.QueueStreamEmitter;
+import org.apache.camel.component.a2a.streaming.SseQueueInputStream;
+import org.apache.camel.component.a2a.streaming.StreamSubscriber;
+import org.apache.camel.component.a2a.util.A2AJsonMapper;
+import org.apache.camel.component.rest.RestEndpoint;
+import org.apache.camel.spi.RestConfiguration;
+import org.apache.camel.spi.RestConsumerFactory;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A2A consumer that automatically registers HTTP endpoints via {@link RestConsumerFactory} SPI.
+ * <p>
+ * On startup, discovers a {@link RestConsumerFactory} (typically provided by camel-platform-http) and registers routes
+ * for all A2A operations. For REST binding, creates separate routes per operation path. For JSON-RPC binding, creates a
+ * single POST route that dispatches via the JSON-RPC method field.
+ * <p>
+ * The agent card is always served at {@code /.well-known/agent-card.json} regardless of protocol binding.
+ */
+public class A2AConsumer extends DefaultConsumer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AConsumer.class);
+    private static final ObjectMapper OBJECT_MAPPER = A2AJsonMapper.instance();
+
+    private final MessageSendOperation messageSendOperation;
+    private final List<Consumer> httpConsumers = new ArrayList<>();
+    private final ConcurrentHashMap<String, Future<?>> inFlightTasks = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, java.util.concurrent.ScheduledFuture<?>> timeoutFutures = new ConcurrentHashMap<>();
+    private ExecutorService asyncExecutor;
+    private ScheduledExecutorService asyncTimeoutScheduler;
+    private Semaphore taskPermits;
+    private LinkedBlockingQueue<PendingTask> pendingTaskQueue;
+    private boolean corsEnabled;
+    private Map<String, String> corsHeaders;
+
+    public A2AConsumer(A2AEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+        this.messageSendOperation = new MessageSendOperation();
+    }
+
+    @Override
+    public A2AEndpoint getEndpoint() {
+        return (A2AEndpoint) super.getEndpoint();
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        AgentCard card = getEndpoint().getResolvedCard();
+        String agentName = card != null ? card.getName() : "unknown";
+        LOG.info("A2A Consumer starting for agent: {}", agentName);
+
+        asyncExecutor = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newThreadPool(this, "A2AAsyncProcessor", 1, 10);
+        asyncTimeoutScheduler = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newScheduledThreadPool(this, "A2AAsyncTimeout", 1);
+
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        if (maxTasks > 0) {
+            taskPermits = new Semaphore(maxTasks);
+            int queueSize = getEndpoint().getConfiguration().getTaskQueueSize();
+            if (queueSize > 0) {
+                pendingTaskQueue = new LinkedBlockingQueue<>(queueSize);
+            }
+            LOG.info("A2A capacity limits: maxConcurrentTasks={}, taskQueueSize={}", maxTasks, queueSize);
+        }
+
+        RestConsumerFactory factory = resolveRestConsumerFactory();
+        if (factory == null) {
+            LOG.warn("No RestConsumerFactory found — HTTP endpoints will not be registered automatically. "
+                     + "Add camel-platform-http or another REST consumer implementation to the classpath.");
+            return;
+        }
+
+        RestConfiguration restConfig = getEndpoint().getCamelContext().getRestConfiguration();
+        String basePath = getEndpoint().getConfiguration().getBasePath();
+
+        corsEnabled = restConfig.isEnableCORS();
+        corsHeaders = restConfig.getCorsHeaders();
+
+        // Agent card is always public — no auth required
+        registerRoute(factory, restConfig, "GET", basePath + A2AConstants.WELL_KNOWN_PATH,
+                this::handleAgentCardRequest, true);
+
+        boolean isJsonRpc
+                = A2AConstants.PROTOCOL_JSONRPC.equalsIgnoreCase(getEndpoint().getConfiguration().getProtocolBinding());
+
+        if (isJsonRpc) {
+            registerSseRoute(factory, restConfig, "POST", basePath + "/",
+                    this::handleJsonRpcDispatch);
+        } else {
+            registerRoute(factory, restConfig, "POST", basePath + "/message:send",
+                    this::handleSendMessage);
+            registerSseRoute(factory, restConfig, "POST", basePath + "/message:stream",
+                    this::handleMessageStream);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks",
+                    this::handleListTasks);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}",
+                    this::handleGetTask);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}:cancel",
+                    this::handleCancelTask);
+            registerSseRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}:subscribe",
+                    this::handleTaskSubscribe);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigCreate);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigList);
+            registerRoute(factory, restConfig, "GET",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigGet);
+            registerRoute(factory, restConfig, "DELETE",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigDelete);
+        }
+
+        if (corsEnabled) {
+            registerRoute(factory, restConfig, "OPTIONS", basePath + A2AConstants.WELL_KNOWN_PATH,
+                    this::handleCorsPreFlight, true);
+            if (isJsonRpc) {
+                registerRoute(factory, restConfig, "OPTIONS", basePath + "/",
+                        this::handleCorsPreFlight, true);
+            } else {
+                for (String path : List.of(
+                        "/message:send", "/message:stream", "/tasks", "/tasks/{taskId}",
+                        "/tasks/{taskId}:cancel", "/tasks/{taskId}:subscribe",
+                        "/tasks/{taskId}/pushNotificationConfigs",
+                        "/tasks/{taskId}/pushNotificationConfigs/{configId}")) {
+                    registerRoute(factory, restConfig, "OPTIONS", basePath + path,
+                            this::handleCorsPreFlight, true);
+                }
+            }
+        }
+
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.startService(consumer);
+        }
+
+        LOG.info("A2A Consumer registered {} HTTP endpoint(s) for agent '{}'",
+                httpConsumers.size(), agentName);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        cancelAllInFlight();
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopService(consumer);
+        }
+        httpConsumers.clear();
+        super.doStop();
+    }
+
+    @Override
+    protected void doShutdown() throws Exception {
+        failPendingQueuedTasks();
+        cancelAllInFlight();
+        if (asyncExecutor != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncExecutor);
+        }
+        if (asyncTimeoutScheduler != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncTimeoutScheduler);
+        }
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopAndShutdownService(consumer);
+        }
+        httpConsumers.clear();
+        super.doShutdown();
+    }
+
+    private void cancelAllInFlight() {
+        for (Future<?> future : inFlightTasks.values()) {
+            future.cancel(true);
+        }
+        inFlightTasks.clear();
+    }
+
+    // ---- Capacity limiting ----
+
+    private boolean tryAcquirePermit() {
+        return taskPermits == null || taskPermits.tryAcquire();
+    }
+
+    private void releasePermit() {
+        if (taskPermits != null) {
+            taskPermits.release();
+            drainPendingQueue();
+        }
+    }
+
+    private void drainPendingQueue() {
+        if (pendingTaskQueue == null || pendingTaskQueue.isEmpty()) {
+            return;
+        }
+        if (!taskPermits.tryAcquire()) {
+            return;
+        }
+        PendingTask pending = pendingTaskQueue.poll();
+        if (pending == null) {
+            taskPermits.release();
+            return;
+        }
+        submitAsyncTask(pending.taskId, pending.contextId, pending.processorExchange);
+    }
+
+    private void failPendingQueuedTasks() {
+        if (pendingTaskQueue == null) {
+            return;
+        }
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        PendingTask pending;
+        while ((pending = pendingTaskQueue.poll()) != null) {
+            Task failedTask = Task.builder()
+                    .id(pending.taskId)
+                    .contextId(pending.contextId)
+                    .status(new TaskStatus(TaskState.FAILED))
+                    .build();
+            store.put(pending.taskId, failedTask);
+            store.notifySubscribers(pending.taskId,
+                    StreamResponse.ofStatusUpdate(buildStatusEvent(failedTask)));
+        }
+    }
+
+    private void writeServerBusyError(Exchange exchange) throws Exception {
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                new A2AError(
+                        "ServerBusyError",
+                        "Agent at capacity: " + maxTasks + " concurrent tasks"));
+        exchange.getMessage().setBody(errorJson);
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 429);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    static class ServerBusyException extends RuntimeException {
+        ServerBusyException(String message) {
+            super(message);
+        }
+    }
+
+    private record PendingTask(String taskId, String contextId, Exchange processorExchange) {
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, false, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, isPublic, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerSseRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, false),
+                verb, path, null,
+                null, SSE_CONTENT_TYPE,
+                restConfig, Map.of("useStreaming", "true"));
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A SSE route: {} {}", verb, path);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic,
+            String produces)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, isPublic),
+                verb, path, null,
+                null, produces,
+                restConfig, Collections.emptyMap());
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A route: {} {}", verb, path);
+    }
+
+    private Processor createDispatchProcessor(A2ARequestHandler handler, boolean isPublic) {
+        return exchange -> {
+            try {
+                if (!isPublic) {
+                    Map<String, Object> profile = getEndpoint().getAuthHandler()
+                            .validateConsumerAuth(exchange, getEndpoint().getResolvedCard());
+                    if (profile != null) {
+                        exchange.getMessage().setHeader(A2AConstants.USER_PROFILE, profile);
+                    }
+                }
+                handler.handle(exchange);
+            } catch (SecurityException e) {
+                LOG.debug("Authentication failed: {}", e.getMessage());
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("AuthenticationError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 401);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } catch (Exception e) {
+                LOG.error("A2A request handling failed", e);
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("InternalError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 500);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } finally {
+                if (corsEnabled) {
+                    setCorsHeaders(exchange);
+                }
+            }
+        };
+    }
+
+    void handleCorsPreFlight(Exchange exchange) {
+        exchange.getMessage().setBody("");
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
+    }
+
+    private void setCorsHeaders(Exchange exchange) {
+        String allowOrigin = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Origin") : null;
+        if (allowOrigin == null) {
+            allowOrigin = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_ORIGIN;
+        }
+        String allowMethods = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Methods") : null;
+        if (allowMethods == null) {
+            allowMethods = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_METHODS;
+        }
+        String allowHeaders = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Headers") : null;
+        if (allowHeaders == null) {
+            allowHeaders = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_HEADERS
+                           + ", A2A-Version, A2A-Extensions, Authorization";
+        }
+        String maxAge = corsHeaders != null ? corsHeaders.get("Access-Control-Max-Age") : null;
+        if (maxAge == null) {
+            maxAge = RestConfiguration.CORS_ACCESS_CONTROL_MAX_AGE;
+        }
+        String allowCredentials = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Credentials") : null;
+
+        // Restrict the origin if credentials are allowed (W3C CORS spec section 6.1, point 3)
+        String origin = exchange.getIn().getHeader("Origin", String.class);
+        if ("true".equalsIgnoreCase(allowCredentials) && "*".equals(allowOrigin) && origin != null) {
+            allowOrigin = origin;
+        }
+
+        exchange.getMessage().setHeader("Access-Control-Allow-Origin", allowOrigin);
+        exchange.getMessage().setHeader("Access-Control-Allow-Methods", allowMethods);
+        exchange.getMessage().setHeader("Access-Control-Allow-Headers", allowHeaders);
+        exchange.getMessage().setHeader("Access-Control-Max-Age", maxAge);
+        if (allowCredentials != null) {
+            exchange.getMessage().setHeader("Access-Control-Allow-Credentials", allowCredentials);
+        }
+    }
+
+    private RestConsumerFactory resolveRestConsumerFactory() {
+        // 1. Explicit httpServerComponent takes priority
+        String serverComponent = getEndpoint().getConfiguration().getHttpServerComponent();
+        if (serverComponent != null) {
+            Object component = getEndpoint().getCamelContext().getComponent(serverComponent, true);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+            throw new IllegalArgumentException(
+                    "httpServerComponent '" + serverComponent + "' does not implement RestConsumerFactory");
+        }
+
+        // 2. Scan existing components
+        for (String name : getEndpoint().getCamelContext().getComponentNames()) {
+            Object component = getEndpoint().getCamelContext().getComponent(name, false);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+        }
+
+        // 3. Auto-discover from registry
+        Set<RestConsumerFactory> factories = getEndpoint().getCamelContext()
+                .getRegistry().findByType(RestConsumerFactory.class);
+        if (!factories.isEmpty()) {
+            return factories.iterator().next();
+        }
+
+        // 4. Auto-create default REST consumer components (same list as REST DSL).
+        //    Skip platform-http — it needs a pre-configured engine and is already covered by step 2.
+        for (String name : RestEndpoint.DEFAULT_REST_CONSUMER_COMPONENTS) {
+            if ("platform-http".equals(name)) {
+                continue;
+            }
+            try {
+                Object component = getEndpoint().getCamelContext().getComponent(name, true);
+                if (component instanceof RestConsumerFactory rcf) {
+                    LOG.info("Auto-discovered {} as RestConsumerFactory", name);
+                    return rcf;
+                }
+            } catch (Exception e) {
+                LOG.debug("Could not auto-create component {}: {}", name, e.getMessage());
+            }
+        }
+        return null;
+    }
+
+    // ---- JSON-RPC dispatch ----
+
+    void handleJsonRpcDispatch(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        byte[] body = exchange.getMessage().getBody(byte[].class);
+        if (body == null || body.length == 0) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: empty body", null);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (body.length > maxSize) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST,
+                    "Request body exceeds maximum size: " + maxSize + " bytes", null);
+            return;
+        }
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        String method;
+        Object requestId;
+        Map<String, Object> params;
+        try {
+            method = jsonRpc.detectMethod(body);
+            requestId = jsonRpc.extractId(body);
+            params = jsonRpc.extractParams(body);
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.PARSE_ERROR, "Parse error: " + e.getMessage(), null);
+            return;
+        }
+
+        if (method == null) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: missing method", requestId);
+            return;
+        }
+
+        // Streaming methods produce text/event-stream, not a single JSON-RPC envelope
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        LOG.info("JSON-RPC dispatch: method={}, op={}", method, op);
+        if (op == A2AOperations.MESSAGE_STREAM) {
+            LOG.info("Entering handleJsonRpcStream for taskId={}", requestId);
+            try {
+                handleJsonRpcStream(exchange, params, requestId);
+                LOG.info("handleJsonRpcStream completed, contentType={}", exchange.getMessage().getHeader("Content-Type"));
+            } catch (Exception e) {
+                LOG.error("handleJsonRpcStream failed: {}", e.getMessage(), e);
+                writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR,
+                        "Internal error: " + e.getMessage(), requestId);
+            }
+            return;
+        }
+
+        Object result;
+        try {
+            result = dispatchJsonRpcMethod(method, params, exchange);
+        } catch (ServerBusyException e) {
+            writeJsonRpcError(exchange, -32000, e.getMessage(), requestId);
+            return;
+        } catch (IllegalArgumentException e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.METHOD_NOT_FOUND, "Method not found: " + method, requestId);
+            return;
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR, "Internal error: " + e.getMessage(), requestId);
+            return;
+        }
+
+        byte[] response = jsonRpc.wrapJsonRpcResponse(result, requestId);
+        exchange.getMessage().setBody(response);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    private Object dispatchJsonRpcMethod(String method, Map<String, Object> params, Exchange exchange) throws Exception {
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        if (op == null) {
+            throw new IllegalArgumentException("Unknown method: " + method);
+        }
+        return switch (op) {
+            case MESSAGE_SEND -> {
+                SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+                yield processSendMessage(request, exchange);
+            }
+            case TASK_GET -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processGetTask(taskId);
+            }
+            case TASK_CANCEL -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processCancelTask(taskId);
+            }
+            case TASK_LIST -> {
+                Integer pageSize = params != null ? (Integer) params.get("pageSize") : null;
+                String contextId = params != null ? (String) params.get("contextId") : null;
+                yield processListTasks(contextId, pageSize);
+            }
+            case PUSH_CONFIG_CREATE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                TaskPushNotificationConfig config = OBJECT_MAPPER.convertValue(params, TaskPushNotificationConfig.class);
+                yield processPushConfigCreate(taskId, config);
+            }
+            case PUSH_CONFIG_GET -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigGet(taskId, configId);
+            }
+            case PUSH_CONFIG_LIST -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                yield processPushConfigList(taskId);
+            }
+            case PUSH_CONFIG_DELETE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigDelete(taskId, configId);
+            }
+            default -> throw new IllegalArgumentException("Unsupported method: " + method);
+        };
+    }
+
+    private void writeJsonRpcError(Exchange exchange, int code, String message, Object requestId) throws Exception {
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        byte[] error = jsonRpc.wrapJsonRpcError(code, message, requestId);
+        exchange.getMessage().setBody(error);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    // ---- JSON-RPC streaming ----
+
+    private void handleJsonRpcStream(Exchange exchange, Map<String, Object> params, Object requestId) throws Exception {
+        if (!tryAcquirePermit()) {
+            writeJsonRpcError(exchange, -32000,
+                    "Agent at capacity: " + getEndpoint().getConfiguration().getMaxConcurrentTasks()
+                                                + " concurrent tasks",
+                    requestId);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        store.put(taskId, Task.builder()
+                .id(taskId).contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED)).build());
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        final Object rpcRequestId = requestId;
+        Function<StreamResponse, String> jsonRpcEncoder = response -> {
+            byte[] envelope = jsonRpc.wrapJsonRpcResponse(response, rpcRequestId);
+            return "data: " + new String(envelope, StandardCharsets.UTF_8) + "\n\n";
+        };
+
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue, jsonRpcEncoder);
+        SseQueueInputStream inputStream = new SseQueueInputStream(
+                queue,
+                getEndpoint().getConfiguration().getAsyncTimeout());
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        submitStreamProcessing(taskId, contextId, request, emitter);
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    // ---- REST handler methods ----
+
+    void handleAgentCardRequest(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        AgentCard card = getEndpoint().getResolvedCard();
+        byte[] cardJson = OBJECT_MAPPER.writeValueAsBytes(card);
+
+        exchange.getMessage().setBody(cardJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleSendMessage(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String requestBody = exchange.getMessage().getBody(String.class);
+        if (requestBody == null || requestBody.isBlank()) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("ContentTypeNotSupportedError", "Request body is empty"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (requestBody.length() > maxSize) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "ContentTypeNotSupportedError",
+                            "Request body exceeds maximum size: " + maxSize + " bytes"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class);
+
+        SendMessageResponse response;
+        try {
+            response = processSendMessage(request, exchange);
+        } catch (ServerBusyException e) {
+            writeServerBusyError(exchange);
+            return;
+        }
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(response);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleGetTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processGetTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleCancelTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processCancelTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (IllegalStateException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotCancelableError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 409);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleListTasks(Exchange exchange) throws Exception {
+        Integer pageSize = exchange.getMessage().getHeader(A2AConstants.LIST_PAGE_SIZE, Integer.class);
+        String contextId = exchange.getMessage().getHeader(A2AConstants.LIST_CONTEXT_ID, String.class);
+
+        filterInboundHeaders(exchange);
+
+        List<Task> tasks = processListTasks(contextId, pageSize);
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(tasks);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    // ---- SSE streaming handlers ----
+
+    private static final String SSE_CONTENT_TYPE = "text/event-stream";
+
+    void handleMessageStream(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String requestBody = exchange.getMessage().getBody(String.class);
+        if (requestBody == null || requestBody.isBlank()) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("ContentTypeNotSupportedError", "Request body is empty"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (requestBody.length() > maxSize) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "ContentTypeNotSupportedError",
+                            "Request body exceeds maximum size: " + maxSize + " bytes"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class);
+
+        if (resolveReturnImmediately(request, exchange)) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "UnsupportedOperationError",
+                            "returnImmediately is not supported for streaming operations per A2A spec"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        if (!tryAcquirePermit()) {
+            writeServerBusyError(exchange);
+            return;
+        }
+
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        store.put(taskId, Task.builder()
+                .id(taskId).contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED)).build());
+
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue);
+        SseQueueInputStream inputStream = new SseQueueInputStream(
+                queue,
+                getEndpoint().getConfiguration().getAsyncTimeout());
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        submitStreamProcessing(taskId, contextId, request, emitter);
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    void handleTaskSubscribe(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Task task = store.get(taskId);
+        if (task == null) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", "Task not found: " + taskId));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long timeout = getEndpoint().getConfiguration().getAsyncTimeout();
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, task.contextId(), queue);
+        SseQueueInputStream inputStream = new SseQueueInputStream(queue, timeout);
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+
+        // Cleanup: unsubscribe when the HTTP component closes the stream
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        // Emit current state as the first event
+        if (task.status() != null) {
+            emitter.emitStatus(task.status().state(), null);
+        }
+
+        // If already terminal, close immediately — queued events + EOF will be read before -1
+        if (task.status() != null && task.status().state() != null
+                && task.status().state().isTerminal()) {
+            emitter.close();
+        }
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    // ---- Shared processing logic (used by both REST and JSON-RPC paths) ----
+
+    private SendMessageResponse processSendMessage(SendMessageRequest request, Exchange exchange) throws Exception {
+        DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext());
+        processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint());
+        messageSendOperation.parseRequest(processorExchange, request);
+
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId);
+        processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId);
+
+        if (resolveReturnImmediately(request, exchange)) {
+            return processSendMessageAsync(processorExchange, taskId, contextId);
+        }
+
+        if (!tryAcquirePermit()) {
+            throw new ServerBusyException(
+                    "Agent at capacity: "
+                                          + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks");
+        }
+        try {
+            getProcessor().process(processorExchange);
+        } finally {
+            releasePermit();
+        }
+
+        SendMessageResponse response = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+
+        if (response.isTaskResponse()) {
+            Task task = response.getTask();
+            A2ATaskStore store = getEndpoint().getTaskStore();
+            store.put(task.id(), task);
+            if (task.status() != null) {
+                store.notifySubscribers(task.id(),
+                        StreamResponse.ofStatusUpdate(buildStatusEvent(task)));
+            }
+        }
+
+        return response;
+    }
+
+    private SendMessageResponse processSendMessageAsync(
+            Exchange processorExchange, String taskId, String contextId) {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Task submittedTask = Task.builder()
+                .id(taskId)
+                .contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED))
+                .build();
+
+        if (tryAcquirePermit()) {
+            store.put(taskId, submittedTask);
+            store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask)));
+            submitAsyncTask(taskId, contextId, processorExchange);
+        } else if (pendingTaskQueue != null
+                && pendingTaskQueue.offer(new PendingTask(taskId, contextId, processorExchange))) {
+            store.put(taskId, submittedTask);
+            store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask)));
+            LOG.debug("Task {} queued (queue size: {})", taskId, pendingTaskQueue.size());
+        } else {
+            throw new ServerBusyException(
+                    "Agent at capacity: "
+                                          + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks");
+        }
+
+        Task responseTask = Task.builder()
+                .id(taskId)
+                .contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED))
+                .build();
+
+        SendMessageResponse response = new SendMessageResponse();
+        response.setTask(responseTask);
+        return response;
+    }
+
+    private void submitAsyncTask(String taskId, String contextId, Exchange processorExchange) {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Future<?> future = asyncExecutor.submit(() -> {
+            try {
+                Task workingTask = store.get(taskId);
+                if (workingTask != null) {
+                    store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.WORKING));
+                }
+                getProcessor().process(processorExchange);
+                SendMessageResponse asyncResponse
+                        = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+                if (asyncResponse.isTaskResponse()) {
+                    Task completed = Task.builder(asyncResponse.getTask())
+                            .id(taskId)
+                            .contextId(contextId)
+                            .build();
+                    store.put(taskId, completed);
+                    store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(completed)));
+                }
+            } catch (Exception e) {
+                LOG.error("Async task processing failed for taskId={}", taskId, e);
+                store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+            } finally {
+                inFlightTasks.remove(taskId);
+                java.util.concurrent.ScheduledFuture<?> tf = timeoutFutures.remove(taskId);
+                if (tf != null) {
+                    tf.cancel(false);
+                }
+                releasePermit();
+            }
+        });
+        inFlightTasks.put(taskId, future);
+
+        long asyncTimeout = getEndpoint().getConfiguration().getAsyncTimeout();
+        if (asyncTimeout > 0) {
+            java.util.concurrent.ScheduledFuture<?> timeoutFuture = asyncTimeoutScheduler.schedule(() -> {
+                timeoutFutures.remove(taskId);
+                Future<?> f = inFlightTasks.get(taskId);
+                if (f != null && !f.isDone()) {
+                    f.cancel(true);
+                    store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+                    inFlightTasks.remove(taskId);
+                    LOG.warn("Async task timed out after {}ms for taskId={}", asyncTimeout, taskId);
+                }
+            }, asyncTimeout, TimeUnit.MILLISECONDS);
+            timeoutFutures.put(taskId, timeoutFuture);
+        }
+    }
+
+    private boolean resolveReturnImmediately(SendMessageRequest request, Exchange exchange) {
+        if (request.getConfiguration() != null) {
+            Object val = request.getConfiguration().get("returnImmediately");
+            if (val instanceof Boolean) {
+                return (Boolean) val;
+            }
+            if (val instanceof String) {
+                return Boolean.parseBoolean((String) val);
+            }
+        }
+        Boolean header = exchange.getMessage().getHeader(A2AConstants.RETURN_IMMEDIATELY, Boolean.class);
+        if (header != null) {
+            return header;
+        }
+        return getEndpoint().getConfiguration().isReturnImmediately();
+    }
+
+    private Task processGetTask(String taskId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        Task task = store.get(taskId);
+        if (task == null) {
+            throw new TaskNotFoundException("Task not found: " + taskId);
+        }
+        return task;
+    }
+
+    private Task processCancelTask(String taskId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        Task task = store.cancelIfNotTerminal(taskId);
+        if (task == null) {
+            throw new TaskNotFoundException("Task not found: " + taskId);
+        }
+        Future<?> inFlight = inFlightTasks.remove(taskId);
+        if (inFlight != null) {
+            inFlight.cancel(true);
+        }
+        store.notifySubscribers(taskId,
+                StreamResponse.ofStatusUpdate(buildStatusEvent(task)));
+        return task;
+    }
+
+    private void submitStreamProcessing(
+            String taskId, String contextId, SendMessageRequest request,
+            QueueStreamEmitter emitter) {
+        DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext());
+        processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint());
+        processorExchange.getMessage().setBody(request.getMessage());
+        processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId);
+        processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId);
+        processorExchange.getMessage().setHeader(A2AConstants.OPERATION, A2AOperations.MESSAGE_STREAM.getMethodName());
+
+        asyncExecutor.submit(() -> {
+            try {
+                getProcessor().process(processorExchange);
+                SendMessageResponse resp = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+                if (resp.isTaskResponse()) {
+                    Task task = resp.getTask();
+                    if (task.history() != null && !task.history().isEmpty()) {
+                        Message agentMessage = task.history().get(task.history().size() - 1);
+                        A2AProgress.emitMessage(processorExchange, agentMessage);
+                    }
+                }
+            } catch (Exception e) {
+                LOG.error("Stream processing failed for task {}: {}", taskId, e.getMessage(), e);
+                A2ATaskStore store = getEndpoint().getTaskStore();
+                store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+            } finally {
+                emitter.close();
+                releasePermit();
+            }
+        });
+    }
+
+    private DefaultStreamEmitter executeStreamProcessing(SendMessageRequest request) throws Exception {

Review Comment:
   This one was flagged as unused. Please, can you double check @luigidemasi ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #23711:
URL: https://github.com/apache/camel/pull/23711#issuecomment-4604443564

   :star2: Thank you for your contribution to the Apache Camel project! :star2:
     :robot: CI automation will test this PR automatically.
   
     :camel: Apache Camel Committers, please review the following items:
   
     * First-time contributors **require MANUAL approval** for the GitHub Actions to run
     * You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
     * You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
     * Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
     
     :warning: Be careful when sharing logs. Review their contents before sharing them publicly.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["davsclaus (via GitHub)" <gi...@apache.org>]

davsclaus commented on PR #23711:
URL: https://github.com/apache/camel/pull/23711#issuecomment-4697909476

   Are we getting in shape for this to be included next week


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["davsclaus (via GitHub)" <gi...@apache.org>]

davsclaus commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348771411


##########
components/camel-ai/camel-a2a/src/generated/resources/a2a.json:
##########
@@ -0,0 +1,15 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "a2a",
+    "title": "A2a",
+    "description": "Camel A2A component for Agent-to-Agent protocol communication",
+    "deprecated": false,
+    "firstVersion": "4.21.0",
+    "label": "ai,a2a",

Review Comment:
   This ensures we have a consistent label(s) across the board.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348760645


##########
components/camel-ai/camel-a2a/src/generated/resources/a2a.json:
##########
@@ -0,0 +1,15 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "a2a",
+    "title": "A2a",
+    "description": "Camel A2A component for Agent-to-Agent protocol communication",
+    "deprecated": false,
+    "firstVersion": "4.21.0",
+    "label": "ai,a2a",

Review Comment:
   changed to `"label": "ai,agent"`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["davsclaus (via GitHub)" <gi...@apache.org>]

davsclaus commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3346989330


##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/push/PushNotificationDispatcher.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.push;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.component.a2a.A2AConstants;
+import org.apache.camel.component.a2a.model.AuthenticationInfo;
+import org.apache.camel.component.a2a.model.StreamResponse;
+import org.apache.camel.component.a2a.model.TaskPushNotificationConfig;
+import org.apache.camel.component.a2a.state.A2ATaskStore;
+import org.apache.camel.component.a2a.util.A2AJsonMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Dispatches push notification events to registered webhook URLs. When task state changes, the task store calls
+ * {@link #dispatch(String, StreamResponse)} which POSTs the event to all registered push configs for that task in
+ * parallel. Supports configurable retry with exponential backoff.
+ */
+public class PushNotificationDispatcher {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PushNotificationDispatcher.class);
+    private static final ObjectMapper OBJECT_MAPPER = A2AJsonMapper.instance();
+    private static final int HTTP_GONE = 410;
+    private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(30);
+
+    private final HttpClient httpClient;
+    private final A2ATaskStore store;
+    private final int maxRetries;
+    private final long initialBackoffMs;
+    private final ScheduledExecutorService executor;
+
+    public PushNotificationDispatcher(HttpClient httpClient, A2ATaskStore store,
+                                      int maxRetries, long initialBackoffMs) {
+        this.httpClient = httpClient;
+        this.store = store;
+        this.maxRetries = maxRetries;
+        this.initialBackoffMs = initialBackoffMs;
+        this.executor = Executors.newScheduledThreadPool(4, r -> {
+            Thread t = new Thread(r, "A2APushDispatcher");

Review Comment:
   **[Medium]** This creates a `ScheduledExecutorService` via `Executors.newScheduledThreadPool()` directly, bypassing Camel's `ExecutorServiceManager`. The threads won't be tracked by JMX/monitoring, won't participate in graceful shutdown, and won't respect Camel's thread pool profile.
   
   The `HttpClient` executor in `A2AEndpoint.doStart()` correctly uses `getCamelContext().getExecutorServiceManager()` — this should follow the same pattern. Since `PushNotificationDispatcher` doesn't have access to `CamelContext`, consider passing in a `ScheduledExecutorService` created by the endpoint instead.



##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java:
##########
@@ -0,0 +1,1278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.a2a.model.A2AError;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.Message;
+import org.apache.camel.component.a2a.model.SendMessageRequest;
+import org.apache.camel.component.a2a.model.SendMessageResponse;
+import org.apache.camel.component.a2a.model.StreamResponse;
+import org.apache.camel.component.a2a.model.Task;
+import org.apache.camel.component.a2a.model.TaskPushNotificationConfig;
+import org.apache.camel.component.a2a.model.TaskState;
+import org.apache.camel.component.a2a.model.TaskStatus;
+import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent;
+import org.apache.camel.component.a2a.operation.A2AOperations;
+import org.apache.camel.component.a2a.operation.MessageSendOperation;
+import org.apache.camel.component.a2a.protocol.JsonRpcProtocol;
+import org.apache.camel.component.a2a.state.A2ATaskStore;
+import org.apache.camel.component.a2a.state.A2ATaskSubscriber;
+import org.apache.camel.component.a2a.streaming.DefaultStreamEmitter;
+import org.apache.camel.component.a2a.streaming.QueueStreamEmitter;
+import org.apache.camel.component.a2a.streaming.SseQueueInputStream;
+import org.apache.camel.component.a2a.streaming.StreamSubscriber;
+import org.apache.camel.component.a2a.util.A2AJsonMapper;
+import org.apache.camel.component.rest.RestEndpoint;
+import org.apache.camel.spi.RestConfiguration;
+import org.apache.camel.spi.RestConsumerFactory;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A2A consumer that automatically registers HTTP endpoints via {@link RestConsumerFactory} SPI.
+ * <p>
+ * On startup, discovers a {@link RestConsumerFactory} (typically provided by camel-platform-http) and registers routes
+ * for all A2A operations. For REST binding, creates separate routes per operation path. For JSON-RPC binding, creates a
+ * single POST route that dispatches via the JSON-RPC method field.
+ * <p>
+ * The agent card is always served at {@code /.well-known/agent-card.json} regardless of protocol binding.
+ */
+public class A2AConsumer extends DefaultConsumer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AConsumer.class);
+    private static final ObjectMapper OBJECT_MAPPER = A2AJsonMapper.instance();
+
+    private final MessageSendOperation messageSendOperation;
+    private final List<Consumer> httpConsumers = new ArrayList<>();
+    private final ConcurrentHashMap<String, Future<?>> inFlightTasks = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, java.util.concurrent.ScheduledFuture<?>> timeoutFutures = new ConcurrentHashMap<>();
+    private ExecutorService asyncExecutor;
+    private ScheduledExecutorService asyncTimeoutScheduler;
+    private Semaphore taskPermits;
+    private LinkedBlockingQueue<PendingTask> pendingTaskQueue;
+    private boolean corsEnabled;
+    private Map<String, String> corsHeaders;
+
+    public A2AConsumer(A2AEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+        this.messageSendOperation = new MessageSendOperation();
+    }
+
+    @Override
+    public A2AEndpoint getEndpoint() {
+        return (A2AEndpoint) super.getEndpoint();
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        AgentCard card = getEndpoint().getResolvedCard();
+        String agentName = card != null ? card.getName() : "unknown";
+        LOG.info("A2A Consumer starting for agent: {}", agentName);
+
+        asyncExecutor = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newThreadPool(this, "A2AAsyncProcessor", 1, 10);
+        asyncTimeoutScheduler = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newScheduledThreadPool(this, "A2AAsyncTimeout", 1);
+
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        if (maxTasks > 0) {
+            taskPermits = new Semaphore(maxTasks);
+            int queueSize = getEndpoint().getConfiguration().getTaskQueueSize();
+            if (queueSize > 0) {
+                pendingTaskQueue = new LinkedBlockingQueue<>(queueSize);
+            }
+            LOG.info("A2A capacity limits: maxConcurrentTasks={}, taskQueueSize={}", maxTasks, queueSize);
+        }
+
+        RestConsumerFactory factory = resolveRestConsumerFactory();
+        if (factory == null) {
+            LOG.warn("No RestConsumerFactory found — HTTP endpoints will not be registered automatically. "
+                     + "Add camel-platform-http or another REST consumer implementation to the classpath.");
+            return;
+        }
+
+        RestConfiguration restConfig = getEndpoint().getCamelContext().getRestConfiguration();
+        String basePath = getEndpoint().getConfiguration().getBasePath();
+
+        corsEnabled = restConfig.isEnableCORS();
+        corsHeaders = restConfig.getCorsHeaders();
+
+        // Agent card is always public — no auth required
+        registerRoute(factory, restConfig, "GET", basePath + A2AConstants.WELL_KNOWN_PATH,
+                this::handleAgentCardRequest, true);
+
+        boolean isJsonRpc
+                = A2AConstants.PROTOCOL_JSONRPC.equalsIgnoreCase(getEndpoint().getConfiguration().getProtocolBinding());
+
+        if (isJsonRpc) {
+            registerSseRoute(factory, restConfig, "POST", basePath + "/",
+                    this::handleJsonRpcDispatch);
+        } else {
+            registerRoute(factory, restConfig, "POST", basePath + "/message:send",
+                    this::handleSendMessage);
+            registerSseRoute(factory, restConfig, "POST", basePath + "/message:stream",
+                    this::handleMessageStream);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks",
+                    this::handleListTasks);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}",
+                    this::handleGetTask);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}:cancel",
+                    this::handleCancelTask);
+            registerSseRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}:subscribe",
+                    this::handleTaskSubscribe);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigCreate);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigList);
+            registerRoute(factory, restConfig, "GET",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigGet);
+            registerRoute(factory, restConfig, "DELETE",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigDelete);
+        }
+
+        if (corsEnabled) {
+            registerRoute(factory, restConfig, "OPTIONS", basePath + A2AConstants.WELL_KNOWN_PATH,
+                    this::handleCorsPreFlight, true);
+            if (isJsonRpc) {
+                registerRoute(factory, restConfig, "OPTIONS", basePath + "/",
+                        this::handleCorsPreFlight, true);
+            } else {
+                for (String path : List.of(
+                        "/message:send", "/message:stream", "/tasks", "/tasks/{taskId}",
+                        "/tasks/{taskId}:cancel", "/tasks/{taskId}:subscribe",
+                        "/tasks/{taskId}/pushNotificationConfigs",
+                        "/tasks/{taskId}/pushNotificationConfigs/{configId}")) {
+                    registerRoute(factory, restConfig, "OPTIONS", basePath + path,
+                            this::handleCorsPreFlight, true);
+                }
+            }
+        }
+
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.startService(consumer);
+        }
+
+        LOG.info("A2A Consumer registered {} HTTP endpoint(s) for agent '{}'",
+                httpConsumers.size(), agentName);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        cancelAllInFlight();
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopService(consumer);
+        }
+        httpConsumers.clear();
+        super.doStop();
+    }
+
+    @Override
+    protected void doShutdown() throws Exception {
+        failPendingQueuedTasks();
+        cancelAllInFlight();
+        if (asyncExecutor != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncExecutor);
+        }
+        if (asyncTimeoutScheduler != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncTimeoutScheduler);
+        }
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopAndShutdownService(consumer);
+        }
+        httpConsumers.clear();
+        super.doShutdown();
+    }
+
+    private void cancelAllInFlight() {
+        for (Future<?> future : inFlightTasks.values()) {
+            future.cancel(true);
+        }
+        inFlightTasks.clear();
+    }
+
+    // ---- Capacity limiting ----
+
+    private boolean tryAcquirePermit() {
+        return taskPermits == null || taskPermits.tryAcquire();
+    }
+
+    private void releasePermit() {
+        if (taskPermits != null) {
+            taskPermits.release();
+            drainPendingQueue();
+        }
+    }
+
+    private void drainPendingQueue() {
+        if (pendingTaskQueue == null || pendingTaskQueue.isEmpty()) {
+            return;
+        }
+        if (!taskPermits.tryAcquire()) {
+            return;
+        }
+        PendingTask pending = pendingTaskQueue.poll();
+        if (pending == null) {
+            taskPermits.release();
+            return;
+        }
+        submitAsyncTask(pending.taskId, pending.contextId, pending.processorExchange);
+    }
+
+    private void failPendingQueuedTasks() {
+        if (pendingTaskQueue == null) {
+            return;
+        }
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        PendingTask pending;
+        while ((pending = pendingTaskQueue.poll()) != null) {
+            Task failedTask = Task.builder()
+                    .id(pending.taskId)
+                    .contextId(pending.contextId)
+                    .status(new TaskStatus(TaskState.FAILED))
+                    .build();
+            store.put(pending.taskId, failedTask);
+            store.notifySubscribers(pending.taskId,
+                    StreamResponse.ofStatusUpdate(buildStatusEvent(failedTask)));
+        }
+    }
+
+    private void writeServerBusyError(Exchange exchange) throws Exception {
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                new A2AError(
+                        "ServerBusyError",
+                        "Agent at capacity: " + maxTasks + " concurrent tasks"));
+        exchange.getMessage().setBody(errorJson);
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 429);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    static class ServerBusyException extends RuntimeException {
+        ServerBusyException(String message) {
+            super(message);
+        }
+    }
+
+    private record PendingTask(String taskId, String contextId, Exchange processorExchange) {
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, false, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, isPublic, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerSseRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, false),
+                verb, path, null,
+                null, SSE_CONTENT_TYPE,
+                restConfig, Map.of("useStreaming", "true"));
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A SSE route: {} {}", verb, path);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic,
+            String produces)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, isPublic),
+                verb, path, null,
+                null, produces,
+                restConfig, Collections.emptyMap());
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A route: {} {}", verb, path);
+    }
+
+    private Processor createDispatchProcessor(A2ARequestHandler handler, boolean isPublic) {
+        return exchange -> {
+            try {
+                if (!isPublic) {
+                    Map<String, Object> profile = getEndpoint().getAuthHandler()
+                            .validateConsumerAuth(exchange, getEndpoint().getResolvedCard());
+                    if (profile != null) {
+                        exchange.getMessage().setHeader(A2AConstants.USER_PROFILE, profile);
+                    }
+                }
+                handler.handle(exchange);
+            } catch (SecurityException e) {
+                LOG.debug("Authentication failed: {}", e.getMessage());
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("AuthenticationError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 401);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } catch (Exception e) {
+                LOG.error("A2A request handling failed", e);
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("InternalError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 500);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } finally {
+                if (corsEnabled) {
+                    setCorsHeaders(exchange);
+                }
+            }
+        };
+    }
+
+    void handleCorsPreFlight(Exchange exchange) {
+        exchange.getMessage().setBody("");
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
+    }
+
+    private void setCorsHeaders(Exchange exchange) {
+        String allowOrigin = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Origin") : null;
+        if (allowOrigin == null) {
+            allowOrigin = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_ORIGIN;
+        }
+        String allowMethods = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Methods") : null;
+        if (allowMethods == null) {
+            allowMethods = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_METHODS;
+        }
+        String allowHeaders = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Headers") : null;
+        if (allowHeaders == null) {
+            allowHeaders = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_HEADERS
+                           + ", A2A-Version, A2A-Extensions, Authorization";
+        }
+        String maxAge = corsHeaders != null ? corsHeaders.get("Access-Control-Max-Age") : null;
+        if (maxAge == null) {
+            maxAge = RestConfiguration.CORS_ACCESS_CONTROL_MAX_AGE;
+        }
+        String allowCredentials = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Credentials") : null;
+
+        // Restrict the origin if credentials are allowed (W3C CORS spec section 6.1, point 3)
+        String origin = exchange.getIn().getHeader("Origin", String.class);
+        if ("true".equalsIgnoreCase(allowCredentials) && "*".equals(allowOrigin) && origin != null) {
+            allowOrigin = origin;
+        }
+
+        exchange.getMessage().setHeader("Access-Control-Allow-Origin", allowOrigin);
+        exchange.getMessage().setHeader("Access-Control-Allow-Methods", allowMethods);
+        exchange.getMessage().setHeader("Access-Control-Allow-Headers", allowHeaders);
+        exchange.getMessage().setHeader("Access-Control-Max-Age", maxAge);
+        if (allowCredentials != null) {
+            exchange.getMessage().setHeader("Access-Control-Allow-Credentials", allowCredentials);
+        }
+    }
+
+    private RestConsumerFactory resolveRestConsumerFactory() {
+        // 1. Explicit httpServerComponent takes priority
+        String serverComponent = getEndpoint().getConfiguration().getHttpServerComponent();
+        if (serverComponent != null) {
+            Object component = getEndpoint().getCamelContext().getComponent(serverComponent, true);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+            throw new IllegalArgumentException(
+                    "httpServerComponent '" + serverComponent + "' does not implement RestConsumerFactory");
+        }
+
+        // 2. Scan existing components
+        for (String name : getEndpoint().getCamelContext().getComponentNames()) {
+            Object component = getEndpoint().getCamelContext().getComponent(name, false);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+        }
+
+        // 3. Auto-discover from registry
+        Set<RestConsumerFactory> factories = getEndpoint().getCamelContext()
+                .getRegistry().findByType(RestConsumerFactory.class);
+        if (!factories.isEmpty()) {
+            return factories.iterator().next();
+        }
+
+        // 4. Auto-create default REST consumer components (same list as REST DSL).
+        //    Skip platform-http — it needs a pre-configured engine and is already covered by step 2.
+        for (String name : RestEndpoint.DEFAULT_REST_CONSUMER_COMPONENTS) {
+            if ("platform-http".equals(name)) {
+                continue;
+            }
+            try {
+                Object component = getEndpoint().getCamelContext().getComponent(name, true);
+                if (component instanceof RestConsumerFactory rcf) {
+                    LOG.info("Auto-discovered {} as RestConsumerFactory", name);
+                    return rcf;
+                }
+            } catch (Exception e) {
+                LOG.debug("Could not auto-create component {}: {}", name, e.getMessage());
+            }
+        }
+        return null;
+    }
+
+    // ---- JSON-RPC dispatch ----
+
+    void handleJsonRpcDispatch(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        byte[] body = exchange.getMessage().getBody(byte[].class);
+        if (body == null || body.length == 0) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: empty body", null);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (body.length > maxSize) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST,
+                    "Request body exceeds maximum size: " + maxSize + " bytes", null);
+            return;
+        }
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        String method;
+        Object requestId;
+        Map<String, Object> params;
+        try {
+            method = jsonRpc.detectMethod(body);
+            requestId = jsonRpc.extractId(body);
+            params = jsonRpc.extractParams(body);
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.PARSE_ERROR, "Parse error: " + e.getMessage(), null);
+            return;
+        }
+
+        if (method == null) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: missing method", requestId);
+            return;
+        }
+
+        // Streaming methods produce text/event-stream, not a single JSON-RPC envelope
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        LOG.info("JSON-RPC dispatch: method={}, op={}", method, op);
+        if (op == A2AOperations.MESSAGE_STREAM) {
+            LOG.info("Entering handleJsonRpcStream for taskId={}", requestId);
+            try {
+                handleJsonRpcStream(exchange, params, requestId);
+                LOG.info("handleJsonRpcStream completed, contentType={}", exchange.getMessage().getHeader("Content-Type"));
+            } catch (Exception e) {
+                LOG.error("handleJsonRpcStream failed: {}", e.getMessage(), e);
+                writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR,
+                        "Internal error: " + e.getMessage(), requestId);
+            }
+            return;
+        }
+
+        Object result;
+        try {
+            result = dispatchJsonRpcMethod(method, params, exchange);
+        } catch (ServerBusyException e) {
+            writeJsonRpcError(exchange, -32000, e.getMessage(), requestId);
+            return;
+        } catch (IllegalArgumentException e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.METHOD_NOT_FOUND, "Method not found: " + method, requestId);
+            return;
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR, "Internal error: " + e.getMessage(), requestId);
+            return;
+        }
+
+        byte[] response = jsonRpc.wrapJsonRpcResponse(result, requestId);
+        exchange.getMessage().setBody(response);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    private Object dispatchJsonRpcMethod(String method, Map<String, Object> params, Exchange exchange) throws Exception {
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        if (op == null) {
+            throw new IllegalArgumentException("Unknown method: " + method);
+        }
+        return switch (op) {
+            case MESSAGE_SEND -> {
+                SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+                yield processSendMessage(request, exchange);
+            }
+            case TASK_GET -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processGetTask(taskId);
+            }
+            case TASK_CANCEL -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processCancelTask(taskId);
+            }
+            case TASK_LIST -> {
+                Integer pageSize = params != null ? (Integer) params.get("pageSize") : null;
+                String contextId = params != null ? (String) params.get("contextId") : null;
+                yield processListTasks(contextId, pageSize);
+            }
+            case PUSH_CONFIG_CREATE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                TaskPushNotificationConfig config = OBJECT_MAPPER.convertValue(params, TaskPushNotificationConfig.class);
+                yield processPushConfigCreate(taskId, config);
+            }
+            case PUSH_CONFIG_GET -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigGet(taskId, configId);
+            }
+            case PUSH_CONFIG_LIST -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                yield processPushConfigList(taskId);
+            }
+            case PUSH_CONFIG_DELETE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigDelete(taskId, configId);
+            }
+            default -> throw new IllegalArgumentException("Unsupported method: " + method);
+        };
+    }
+
+    private void writeJsonRpcError(Exchange exchange, int code, String message, Object requestId) throws Exception {
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        byte[] error = jsonRpc.wrapJsonRpcError(code, message, requestId);
+        exchange.getMessage().setBody(error);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    // ---- JSON-RPC streaming ----
+
+    private void handleJsonRpcStream(Exchange exchange, Map<String, Object> params, Object requestId) throws Exception {
+        if (!tryAcquirePermit()) {
+            writeJsonRpcError(exchange, -32000,
+                    "Agent at capacity: " + getEndpoint().getConfiguration().getMaxConcurrentTasks()
+                                                + " concurrent tasks",
+                    requestId);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        store.put(taskId, Task.builder()
+                .id(taskId).contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED)).build());
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        final Object rpcRequestId = requestId;
+        Function<StreamResponse, String> jsonRpcEncoder = response -> {
+            byte[] envelope = jsonRpc.wrapJsonRpcResponse(response, rpcRequestId);
+            return "data: " + new String(envelope, StandardCharsets.UTF_8) + "\n\n";
+        };
+
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue, jsonRpcEncoder);
+        SseQueueInputStream inputStream = new SseQueueInputStream(
+                queue,
+                getEndpoint().getConfiguration().getAsyncTimeout());
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        submitStreamProcessing(taskId, contextId, request, emitter);
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    // ---- REST handler methods ----
+
+    void handleAgentCardRequest(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        AgentCard card = getEndpoint().getResolvedCard();
+        byte[] cardJson = OBJECT_MAPPER.writeValueAsBytes(card);
+
+        exchange.getMessage().setBody(cardJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleSendMessage(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String requestBody = exchange.getMessage().getBody(String.class);
+        if (requestBody == null || requestBody.isBlank()) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("ContentTypeNotSupportedError", "Request body is empty"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (requestBody.length() > maxSize) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "ContentTypeNotSupportedError",
+                            "Request body exceeds maximum size: " + maxSize + " bytes"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class);
+
+        SendMessageResponse response;
+        try {
+            response = processSendMessage(request, exchange);
+        } catch (ServerBusyException e) {
+            writeServerBusyError(exchange);
+            return;
+        }
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(response);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleGetTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processGetTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleCancelTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processCancelTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (IllegalStateException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotCancelableError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 409);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleListTasks(Exchange exchange) throws Exception {
+        Integer pageSize = exchange.getMessage().getHeader(A2AConstants.LIST_PAGE_SIZE, Integer.class);
+        String contextId = exchange.getMessage().getHeader(A2AConstants.LIST_CONTEXT_ID, String.class);
+
+        filterInboundHeaders(exchange);
+
+        List<Task> tasks = processListTasks(contextId, pageSize);
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(tasks);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    // ---- SSE streaming handlers ----
+
+    private static final String SSE_CONTENT_TYPE = "text/event-stream";
+
+    void handleMessageStream(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String requestBody = exchange.getMessage().getBody(String.class);
+        if (requestBody == null || requestBody.isBlank()) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("ContentTypeNotSupportedError", "Request body is empty"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (requestBody.length() > maxSize) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "ContentTypeNotSupportedError",
+                            "Request body exceeds maximum size: " + maxSize + " bytes"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class);
+
+        if (resolveReturnImmediately(request, exchange)) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "UnsupportedOperationError",
+                            "returnImmediately is not supported for streaming operations per A2A spec"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        if (!tryAcquirePermit()) {
+            writeServerBusyError(exchange);
+            return;
+        }
+
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        store.put(taskId, Task.builder()
+                .id(taskId).contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED)).build());
+
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue);
+        SseQueueInputStream inputStream = new SseQueueInputStream(
+                queue,
+                getEndpoint().getConfiguration().getAsyncTimeout());
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        submitStreamProcessing(taskId, contextId, request, emitter);
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    void handleTaskSubscribe(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Task task = store.get(taskId);
+        if (task == null) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", "Task not found: " + taskId));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long timeout = getEndpoint().getConfiguration().getAsyncTimeout();
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, task.contextId(), queue);
+        SseQueueInputStream inputStream = new SseQueueInputStream(queue, timeout);
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+
+        // Cleanup: unsubscribe when the HTTP component closes the stream
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        // Emit current state as the first event
+        if (task.status() != null) {
+            emitter.emitStatus(task.status().state(), null);
+        }
+
+        // If already terminal, close immediately — queued events + EOF will be read before -1
+        if (task.status() != null && task.status().state() != null
+                && task.status().state().isTerminal()) {
+            emitter.close();
+        }
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    // ---- Shared processing logic (used by both REST and JSON-RPC paths) ----
+
+    private SendMessageResponse processSendMessage(SendMessageRequest request, Exchange exchange) throws Exception {
+        DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext());
+        processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint());
+        messageSendOperation.parseRequest(processorExchange, request);
+
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId);
+        processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId);
+
+        if (resolveReturnImmediately(request, exchange)) {
+            return processSendMessageAsync(processorExchange, taskId, contextId);
+        }
+
+        if (!tryAcquirePermit()) {
+            throw new ServerBusyException(
+                    "Agent at capacity: "
+                                          + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks");
+        }
+        try {
+            getProcessor().process(processorExchange);
+        } finally {
+            releasePermit();
+        }
+
+        SendMessageResponse response = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+
+        if (response.isTaskResponse()) {
+            Task task = response.getTask();
+            A2ATaskStore store = getEndpoint().getTaskStore();
+            store.put(task.id(), task);
+            if (task.status() != null) {
+                store.notifySubscribers(task.id(),
+                        StreamResponse.ofStatusUpdate(buildStatusEvent(task)));
+            }
+        }
+
+        return response;
+    }
+
+    private SendMessageResponse processSendMessageAsync(
+            Exchange processorExchange, String taskId, String contextId) {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Task submittedTask = Task.builder()
+                .id(taskId)
+                .contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED))
+                .build();
+
+        if (tryAcquirePermit()) {
+            store.put(taskId, submittedTask);
+            store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask)));
+            submitAsyncTask(taskId, contextId, processorExchange);
+        } else if (pendingTaskQueue != null
+                && pendingTaskQueue.offer(new PendingTask(taskId, contextId, processorExchange))) {
+            store.put(taskId, submittedTask);
+            store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(submittedTask)));
+            LOG.debug("Task {} queued (queue size: {})", taskId, pendingTaskQueue.size());
+        } else {
+            throw new ServerBusyException(
+                    "Agent at capacity: "
+                                          + getEndpoint().getConfiguration().getMaxConcurrentTasks() + " concurrent tasks");
+        }
+
+        Task responseTask = Task.builder()
+                .id(taskId)
+                .contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED))
+                .build();
+
+        SendMessageResponse response = new SendMessageResponse();
+        response.setTask(responseTask);
+        return response;
+    }
+
+    private void submitAsyncTask(String taskId, String contextId, Exchange processorExchange) {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        Future<?> future = asyncExecutor.submit(() -> {
+            try {
+                Task workingTask = store.get(taskId);
+                if (workingTask != null) {
+                    store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.WORKING));
+                }
+                getProcessor().process(processorExchange);
+                SendMessageResponse asyncResponse
+                        = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+                if (asyncResponse.isTaskResponse()) {
+                    Task completed = Task.builder(asyncResponse.getTask())
+                            .id(taskId)
+                            .contextId(contextId)
+                            .build();
+                    store.put(taskId, completed);
+                    store.notifySubscribers(taskId, StreamResponse.ofStatusUpdate(buildStatusEvent(completed)));
+                }
+            } catch (Exception e) {
+                LOG.error("Async task processing failed for taskId={}", taskId, e);
+                store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+            } finally {
+                inFlightTasks.remove(taskId);
+                java.util.concurrent.ScheduledFuture<?> tf = timeoutFutures.remove(taskId);
+                if (tf != null) {
+                    tf.cancel(false);
+                }
+                releasePermit();
+            }
+        });
+        inFlightTasks.put(taskId, future);
+
+        long asyncTimeout = getEndpoint().getConfiguration().getAsyncTimeout();
+        if (asyncTimeout > 0) {
+            java.util.concurrent.ScheduledFuture<?> timeoutFuture = asyncTimeoutScheduler.schedule(() -> {
+                timeoutFutures.remove(taskId);
+                Future<?> f = inFlightTasks.get(taskId);
+                if (f != null && !f.isDone()) {
+                    f.cancel(true);
+                    store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+                    inFlightTasks.remove(taskId);
+                    LOG.warn("Async task timed out after {}ms for taskId={}", asyncTimeout, taskId);
+                }
+            }, asyncTimeout, TimeUnit.MILLISECONDS);
+            timeoutFutures.put(taskId, timeoutFuture);
+        }
+    }
+
+    private boolean resolveReturnImmediately(SendMessageRequest request, Exchange exchange) {
+        if (request.getConfiguration() != null) {
+            Object val = request.getConfiguration().get("returnImmediately");
+            if (val instanceof Boolean) {
+                return (Boolean) val;
+            }
+            if (val instanceof String) {
+                return Boolean.parseBoolean((String) val);
+            }
+        }
+        Boolean header = exchange.getMessage().getHeader(A2AConstants.RETURN_IMMEDIATELY, Boolean.class);
+        if (header != null) {
+            return header;
+        }
+        return getEndpoint().getConfiguration().isReturnImmediately();
+    }
+
+    private Task processGetTask(String taskId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        Task task = store.get(taskId);
+        if (task == null) {
+            throw new TaskNotFoundException("Task not found: " + taskId);
+        }
+        return task;
+    }
+
+    private Task processCancelTask(String taskId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        Task task = store.cancelIfNotTerminal(taskId);
+        if (task == null) {
+            throw new TaskNotFoundException("Task not found: " + taskId);
+        }
+        Future<?> inFlight = inFlightTasks.remove(taskId);
+        if (inFlight != null) {
+            inFlight.cancel(true);
+        }
+        store.notifySubscribers(taskId,
+                StreamResponse.ofStatusUpdate(buildStatusEvent(task)));
+        return task;
+    }
+
+    private void submitStreamProcessing(
+            String taskId, String contextId, SendMessageRequest request,
+            QueueStreamEmitter emitter) {
+        DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext());
+        processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint());
+        processorExchange.getMessage().setBody(request.getMessage());
+        processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId);
+        processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId);
+        processorExchange.getMessage().setHeader(A2AConstants.OPERATION, A2AOperations.MESSAGE_STREAM.getMethodName());
+
+        asyncExecutor.submit(() -> {
+            try {
+                getProcessor().process(processorExchange);
+                SendMessageResponse resp = (SendMessageResponse) messageSendOperation.buildResponse(processorExchange);
+                if (resp.isTaskResponse()) {
+                    Task task = resp.getTask();
+                    if (task.history() != null && !task.history().isEmpty()) {
+                        Message agentMessage = task.history().get(task.history().size() - 1);
+                        A2AProgress.emitMessage(processorExchange, agentMessage);
+                    }
+                }
+            } catch (Exception e) {
+                LOG.error("Stream processing failed for task {}: {}", taskId, e.getMessage(), e);
+                A2ATaskStore store = getEndpoint().getTaskStore();
+                store.updateStatusAndNotify(taskId, new TaskStatus(TaskState.FAILED));
+            } finally {
+                emitter.close();
+                releasePermit();
+            }
+        });
+    }
+
+    private DefaultStreamEmitter executeStreamProcessing(SendMessageRequest request) throws Exception {
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        Task streamTask = Task.builder()
+                .id(taskId)
+                .contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED))
+                .build();
+        store.put(taskId, streamTask);
+
+        DefaultStreamEmitter emitter = new DefaultStreamEmitter(taskId, contextId);
+        store.addSubscriber(taskId, new StreamSubscriber(emitter));
+        DefaultExchange processorExchange = new DefaultExchange(getEndpoint().getCamelContext());
+        processorExchange.getExchangeExtension().setFromEndpoint(getEndpoint());
+        processorExchange.getMessage().setBody(request.getMessage());
+        processorExchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId);
+        processorExchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId);
+        processorExchange.getMessage().setHeader(A2AConstants.OPERATION, A2AOperations.MESSAGE_STREAM.getMethodName());
+
+        getProcessor().process(processorExchange);
+        emitter.close();
+
+        return emitter;
+    }
+
+    private List<Task> processListTasks(String contextId, Integer pageSize) {
+        int maxResults = pageSize != null ? Math.min(pageSize, 100) : 50;
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        return store.list(contextId, null, maxResults);
+    }
+
+    // ---- Push notification config handlers ----
+
+    void handlePushConfigCreate(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        TaskPushNotificationConfig config = OBJECT_MAPPER.readValue(
+                exchange.getMessage().getBody(byte[].class), TaskPushNotificationConfig.class);
+
+        try {
+            config = processPushConfigCreate(taskId, config);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(config);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handlePushConfigGet(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String[] ids = extractPushConfigIds(exchange);
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        TaskPushNotificationConfig config = store.getPushConfig(ids[0], ids[1]);
+        if (config == null) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", "Push config not found: " + ids[1]));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(config);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handlePushConfigList(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        List<TaskPushNotificationConfig> configs = store.listPushConfigs(taskId);
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(configs);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handlePushConfigDelete(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String[] ids = extractPushConfigIds(exchange);
+        A2ATaskStore store = getEndpoint().getTaskStore();
+
+        boolean deleted = store.deletePushConfig(ids[0], ids[1]);
+        if (!deleted) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", "Push config not found: " + ids[1]));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        exchange.getMessage().setBody(OBJECT_MAPPER.writeValueAsBytes(Collections.emptyMap()));
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    private TaskPushNotificationConfig processPushConfigCreate(String taskId, TaskPushNotificationConfig config)
+            throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        if (!store.contains(taskId)) {
+            throw new TaskNotFoundException("Task not found: " + taskId);
+        }
+        store.putPushConfig(taskId, config);
+        return config;
+    }
+
+    private TaskPushNotificationConfig processPushConfigGet(String taskId, String configId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        TaskPushNotificationConfig config = store.getPushConfig(taskId, configId);
+        if (config == null) {
+            throw new TaskNotFoundException("Push config not found: " + configId);
+        }
+        return config;
+    }
+
+    private List<TaskPushNotificationConfig> processPushConfigList(String taskId) {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        return store.listPushConfigs(taskId);
+    }
+
+    private Object processPushConfigDelete(String taskId, String configId) throws TaskNotFoundException {
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        if (!store.deletePushConfig(taskId, configId)) {
+            throw new TaskNotFoundException("Push config not found: " + configId);
+        }
+        return Collections.emptyMap();
+    }
+
+    // ---- Utilities ----
+
+    private static TaskStatusUpdateEvent buildStatusEvent(Task task) {
+        return TaskStatusUpdateEvent.builder()
+                .taskId(task.id())
+                .contextId(task.contextId())
+                .status(task.status())
+                .build();
+    }
+
+    private void filterInboundHeaders(Exchange exchange) {
+        exchange.getMessage().getHeaders().entrySet().removeIf(entry -> {
+            String lower = entry.getKey().toLowerCase(Locale.ENGLISH);
+            return lower.startsWith("camela2a") || lower.startsWith("org.apache.camel.");
+        });
+    }

Review Comment:
   **[Blocking — Security]** This filter only strips `CamelA2A*` and `org.apache.camel.*` headers but does **not** strip all `Camel*` headers. Per the project's [security model](https://github.com/apache/camel/blob/main/docs/user-manual/modules/ROOT/pages/security-model.adoc) and the committer review checklist, consumers exposed to untrusted input must block `Camel*` / `camel*` / `org.apache.camel.*` **case-insensitively**.
   
   The current filter leaves headers like `CamelBeanMethodName`, `CamelExecCommandExecutable`, etc. unfiltered — this is the exact attack vector from CVE-2025-27636 and five follow-on CVEs.
   
   ```suggestion
               return lower.startsWith("camel") || lower.startsWith("org.apache.camel.");
   ```



##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java:
##########
@@ -0,0 +1,1278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.a2a.model.A2AError;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.Message;
+import org.apache.camel.component.a2a.model.SendMessageRequest;
+import org.apache.camel.component.a2a.model.SendMessageResponse;
+import org.apache.camel.component.a2a.model.StreamResponse;
+import org.apache.camel.component.a2a.model.Task;
+import org.apache.camel.component.a2a.model.TaskPushNotificationConfig;
+import org.apache.camel.component.a2a.model.TaskState;
+import org.apache.camel.component.a2a.model.TaskStatus;
+import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent;
+import org.apache.camel.component.a2a.operation.A2AOperations;
+import org.apache.camel.component.a2a.operation.MessageSendOperation;
+import org.apache.camel.component.a2a.protocol.JsonRpcProtocol;
+import org.apache.camel.component.a2a.state.A2ATaskStore;
+import org.apache.camel.component.a2a.state.A2ATaskSubscriber;
+import org.apache.camel.component.a2a.streaming.DefaultStreamEmitter;
+import org.apache.camel.component.a2a.streaming.QueueStreamEmitter;
+import org.apache.camel.component.a2a.streaming.SseQueueInputStream;
+import org.apache.camel.component.a2a.streaming.StreamSubscriber;
+import org.apache.camel.component.a2a.util.A2AJsonMapper;
+import org.apache.camel.component.rest.RestEndpoint;
+import org.apache.camel.spi.RestConfiguration;
+import org.apache.camel.spi.RestConsumerFactory;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A2A consumer that automatically registers HTTP endpoints via {@link RestConsumerFactory} SPI.
+ * <p>
+ * On startup, discovers a {@link RestConsumerFactory} (typically provided by camel-platform-http) and registers routes
+ * for all A2A operations. For REST binding, creates separate routes per operation path. For JSON-RPC binding, creates a
+ * single POST route that dispatches via the JSON-RPC method field.
+ * <p>
+ * The agent card is always served at {@code /.well-known/agent-card.json} regardless of protocol binding.
+ */
+public class A2AConsumer extends DefaultConsumer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AConsumer.class);
+    private static final ObjectMapper OBJECT_MAPPER = A2AJsonMapper.instance();
+
+    private final MessageSendOperation messageSendOperation;
+    private final List<Consumer> httpConsumers = new ArrayList<>();
+    private final ConcurrentHashMap<String, Future<?>> inFlightTasks = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, java.util.concurrent.ScheduledFuture<?>> timeoutFutures = new ConcurrentHashMap<>();
+    private ExecutorService asyncExecutor;
+    private ScheduledExecutorService asyncTimeoutScheduler;
+    private Semaphore taskPermits;
+    private LinkedBlockingQueue<PendingTask> pendingTaskQueue;
+    private boolean corsEnabled;
+    private Map<String, String> corsHeaders;
+
+    public A2AConsumer(A2AEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+        this.messageSendOperation = new MessageSendOperation();
+    }
+
+    @Override
+    public A2AEndpoint getEndpoint() {
+        return (A2AEndpoint) super.getEndpoint();
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        AgentCard card = getEndpoint().getResolvedCard();
+        String agentName = card != null ? card.getName() : "unknown";
+        LOG.info("A2A Consumer starting for agent: {}", agentName);
+
+        asyncExecutor = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newThreadPool(this, "A2AAsyncProcessor", 1, 10);
+        asyncTimeoutScheduler = getEndpoint().getCamelContext()
+                .getExecutorServiceManager()
+                .newScheduledThreadPool(this, "A2AAsyncTimeout", 1);
+
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        if (maxTasks > 0) {
+            taskPermits = new Semaphore(maxTasks);
+            int queueSize = getEndpoint().getConfiguration().getTaskQueueSize();
+            if (queueSize > 0) {
+                pendingTaskQueue = new LinkedBlockingQueue<>(queueSize);
+            }
+            LOG.info("A2A capacity limits: maxConcurrentTasks={}, taskQueueSize={}", maxTasks, queueSize);
+        }
+
+        RestConsumerFactory factory = resolveRestConsumerFactory();
+        if (factory == null) {
+            LOG.warn("No RestConsumerFactory found — HTTP endpoints will not be registered automatically. "
+                     + "Add camel-platform-http or another REST consumer implementation to the classpath.");
+            return;
+        }
+
+        RestConfiguration restConfig = getEndpoint().getCamelContext().getRestConfiguration();
+        String basePath = getEndpoint().getConfiguration().getBasePath();
+
+        corsEnabled = restConfig.isEnableCORS();
+        corsHeaders = restConfig.getCorsHeaders();
+
+        // Agent card is always public — no auth required
+        registerRoute(factory, restConfig, "GET", basePath + A2AConstants.WELL_KNOWN_PATH,
+                this::handleAgentCardRequest, true);
+
+        boolean isJsonRpc
+                = A2AConstants.PROTOCOL_JSONRPC.equalsIgnoreCase(getEndpoint().getConfiguration().getProtocolBinding());
+
+        if (isJsonRpc) {
+            registerSseRoute(factory, restConfig, "POST", basePath + "/",
+                    this::handleJsonRpcDispatch);
+        } else {
+            registerRoute(factory, restConfig, "POST", basePath + "/message:send",
+                    this::handleSendMessage);
+            registerSseRoute(factory, restConfig, "POST", basePath + "/message:stream",
+                    this::handleMessageStream);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks",
+                    this::handleListTasks);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}",
+                    this::handleGetTask);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}:cancel",
+                    this::handleCancelTask);
+            registerSseRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}:subscribe",
+                    this::handleTaskSubscribe);
+            registerRoute(factory, restConfig, "POST", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigCreate);
+            registerRoute(factory, restConfig, "GET", basePath + "/tasks/{taskId}/pushNotificationConfigs",
+                    this::handlePushConfigList);
+            registerRoute(factory, restConfig, "GET",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigGet);
+            registerRoute(factory, restConfig, "DELETE",
+                    basePath + "/tasks/{taskId}/pushNotificationConfigs/{configId}",
+                    this::handlePushConfigDelete);
+        }
+
+        if (corsEnabled) {
+            registerRoute(factory, restConfig, "OPTIONS", basePath + A2AConstants.WELL_KNOWN_PATH,
+                    this::handleCorsPreFlight, true);
+            if (isJsonRpc) {
+                registerRoute(factory, restConfig, "OPTIONS", basePath + "/",
+                        this::handleCorsPreFlight, true);
+            } else {
+                for (String path : List.of(
+                        "/message:send", "/message:stream", "/tasks", "/tasks/{taskId}",
+                        "/tasks/{taskId}:cancel", "/tasks/{taskId}:subscribe",
+                        "/tasks/{taskId}/pushNotificationConfigs",
+                        "/tasks/{taskId}/pushNotificationConfigs/{configId}")) {
+                    registerRoute(factory, restConfig, "OPTIONS", basePath + path,
+                            this::handleCorsPreFlight, true);
+                }
+            }
+        }
+
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.startService(consumer);
+        }
+
+        LOG.info("A2A Consumer registered {} HTTP endpoint(s) for agent '{}'",
+                httpConsumers.size(), agentName);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        cancelAllInFlight();
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopService(consumer);
+        }
+        httpConsumers.clear();
+        super.doStop();
+    }
+
+    @Override
+    protected void doShutdown() throws Exception {
+        failPendingQueuedTasks();
+        cancelAllInFlight();
+        if (asyncExecutor != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncExecutor);
+        }
+        if (asyncTimeoutScheduler != null) {
+            getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(asyncTimeoutScheduler);
+        }
+        for (Consumer consumer : httpConsumers) {
+            ServiceHelper.stopAndShutdownService(consumer);
+        }
+        httpConsumers.clear();
+        super.doShutdown();
+    }
+
+    private void cancelAllInFlight() {
+        for (Future<?> future : inFlightTasks.values()) {
+            future.cancel(true);
+        }
+        inFlightTasks.clear();
+    }
+
+    // ---- Capacity limiting ----
+
+    private boolean tryAcquirePermit() {
+        return taskPermits == null || taskPermits.tryAcquire();
+    }
+
+    private void releasePermit() {
+        if (taskPermits != null) {
+            taskPermits.release();
+            drainPendingQueue();
+        }
+    }
+
+    private void drainPendingQueue() {
+        if (pendingTaskQueue == null || pendingTaskQueue.isEmpty()) {
+            return;
+        }
+        if (!taskPermits.tryAcquire()) {
+            return;
+        }
+        PendingTask pending = pendingTaskQueue.poll();
+        if (pending == null) {
+            taskPermits.release();
+            return;
+        }
+        submitAsyncTask(pending.taskId, pending.contextId, pending.processorExchange);
+    }
+
+    private void failPendingQueuedTasks() {
+        if (pendingTaskQueue == null) {
+            return;
+        }
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        PendingTask pending;
+        while ((pending = pendingTaskQueue.poll()) != null) {
+            Task failedTask = Task.builder()
+                    .id(pending.taskId)
+                    .contextId(pending.contextId)
+                    .status(new TaskStatus(TaskState.FAILED))
+                    .build();
+            store.put(pending.taskId, failedTask);
+            store.notifySubscribers(pending.taskId,
+                    StreamResponse.ofStatusUpdate(buildStatusEvent(failedTask)));
+        }
+    }
+
+    private void writeServerBusyError(Exchange exchange) throws Exception {
+        int maxTasks = getEndpoint().getConfiguration().getMaxConcurrentTasks();
+        byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                new A2AError(
+                        "ServerBusyError",
+                        "Agent at capacity: " + maxTasks + " concurrent tasks"));
+        exchange.getMessage().setBody(errorJson);
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 429);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    static class ServerBusyException extends RuntimeException {
+        ServerBusyException(String message) {
+            super(message);
+        }
+    }
+
+    private record PendingTask(String taskId, String contextId, Exchange processorExchange) {
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, false, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic)
+            throws Exception {
+        registerRoute(factory, restConfig, verb, path, handler, isPublic, A2AConstants.CONTENT_TYPE);
+    }
+
+    private void registerSseRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, false),
+                verb, path, null,
+                null, SSE_CONTENT_TYPE,
+                restConfig, Map.of("useStreaming", "true"));
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A SSE route: {} {}", verb, path);
+    }
+
+    private void registerRoute(
+            RestConsumerFactory factory, RestConfiguration restConfig,
+            String verb, String path, A2ARequestHandler handler, boolean isPublic,
+            String produces)
+            throws Exception {
+        Consumer consumer = factory.createConsumer(
+                getEndpoint().getCamelContext(),
+                createDispatchProcessor(handler, isPublic),
+                verb, path, null,
+                null, produces,
+                restConfig, Collections.emptyMap());
+        httpConsumers.add(consumer);
+        LOG.debug("Registered A2A route: {} {}", verb, path);
+    }
+
+    private Processor createDispatchProcessor(A2ARequestHandler handler, boolean isPublic) {
+        return exchange -> {
+            try {
+                if (!isPublic) {
+                    Map<String, Object> profile = getEndpoint().getAuthHandler()
+                            .validateConsumerAuth(exchange, getEndpoint().getResolvedCard());
+                    if (profile != null) {
+                        exchange.getMessage().setHeader(A2AConstants.USER_PROFILE, profile);
+                    }
+                }
+                handler.handle(exchange);
+            } catch (SecurityException e) {
+                LOG.debug("Authentication failed: {}", e.getMessage());
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("AuthenticationError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 401);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } catch (Exception e) {
+                LOG.error("A2A request handling failed", e);
+                byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                        new A2AError("InternalError", e.getMessage()));
+                exchange.getMessage().setBody(errorJson);
+                exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 500);
+                exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            } finally {
+                if (corsEnabled) {
+                    setCorsHeaders(exchange);
+                }
+            }
+        };
+    }
+
+    void handleCorsPreFlight(Exchange exchange) {
+        exchange.getMessage().setBody("");
+        exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
+    }
+
+    private void setCorsHeaders(Exchange exchange) {
+        String allowOrigin = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Origin") : null;
+        if (allowOrigin == null) {
+            allowOrigin = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_ORIGIN;
+        }
+        String allowMethods = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Methods") : null;
+        if (allowMethods == null) {
+            allowMethods = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_METHODS;
+        }
+        String allowHeaders = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Headers") : null;
+        if (allowHeaders == null) {
+            allowHeaders = RestConfiguration.CORS_ACCESS_CONTROL_ALLOW_HEADERS
+                           + ", A2A-Version, A2A-Extensions, Authorization";
+        }
+        String maxAge = corsHeaders != null ? corsHeaders.get("Access-Control-Max-Age") : null;
+        if (maxAge == null) {
+            maxAge = RestConfiguration.CORS_ACCESS_CONTROL_MAX_AGE;
+        }
+        String allowCredentials = corsHeaders != null ? corsHeaders.get("Access-Control-Allow-Credentials") : null;
+
+        // Restrict the origin if credentials are allowed (W3C CORS spec section 6.1, point 3)
+        String origin = exchange.getIn().getHeader("Origin", String.class);
+        if ("true".equalsIgnoreCase(allowCredentials) && "*".equals(allowOrigin) && origin != null) {
+            allowOrigin = origin;
+        }
+
+        exchange.getMessage().setHeader("Access-Control-Allow-Origin", allowOrigin);
+        exchange.getMessage().setHeader("Access-Control-Allow-Methods", allowMethods);
+        exchange.getMessage().setHeader("Access-Control-Allow-Headers", allowHeaders);
+        exchange.getMessage().setHeader("Access-Control-Max-Age", maxAge);
+        if (allowCredentials != null) {
+            exchange.getMessage().setHeader("Access-Control-Allow-Credentials", allowCredentials);
+        }
+    }
+
+    private RestConsumerFactory resolveRestConsumerFactory() {
+        // 1. Explicit httpServerComponent takes priority
+        String serverComponent = getEndpoint().getConfiguration().getHttpServerComponent();
+        if (serverComponent != null) {
+            Object component = getEndpoint().getCamelContext().getComponent(serverComponent, true);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+            throw new IllegalArgumentException(
+                    "httpServerComponent '" + serverComponent + "' does not implement RestConsumerFactory");
+        }
+
+        // 2. Scan existing components
+        for (String name : getEndpoint().getCamelContext().getComponentNames()) {
+            Object component = getEndpoint().getCamelContext().getComponent(name, false);
+            if (component instanceof RestConsumerFactory rcf) {
+                return rcf;
+            }
+        }
+
+        // 3. Auto-discover from registry
+        Set<RestConsumerFactory> factories = getEndpoint().getCamelContext()
+                .getRegistry().findByType(RestConsumerFactory.class);
+        if (!factories.isEmpty()) {
+            return factories.iterator().next();
+        }
+
+        // 4. Auto-create default REST consumer components (same list as REST DSL).
+        //    Skip platform-http — it needs a pre-configured engine and is already covered by step 2.
+        for (String name : RestEndpoint.DEFAULT_REST_CONSUMER_COMPONENTS) {
+            if ("platform-http".equals(name)) {
+                continue;
+            }
+            try {
+                Object component = getEndpoint().getCamelContext().getComponent(name, true);
+                if (component instanceof RestConsumerFactory rcf) {
+                    LOG.info("Auto-discovered {} as RestConsumerFactory", name);
+                    return rcf;
+                }
+            } catch (Exception e) {
+                LOG.debug("Could not auto-create component {}: {}", name, e.getMessage());
+            }
+        }
+        return null;
+    }
+
+    // ---- JSON-RPC dispatch ----
+
+    void handleJsonRpcDispatch(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        byte[] body = exchange.getMessage().getBody(byte[].class);
+        if (body == null || body.length == 0) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: empty body", null);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (body.length > maxSize) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST,
+                    "Request body exceeds maximum size: " + maxSize + " bytes", null);
+            return;
+        }
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        String method;
+        Object requestId;
+        Map<String, Object> params;
+        try {
+            method = jsonRpc.detectMethod(body);
+            requestId = jsonRpc.extractId(body);
+            params = jsonRpc.extractParams(body);
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.PARSE_ERROR, "Parse error: " + e.getMessage(), null);
+            return;
+        }
+
+        if (method == null) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INVALID_REQUEST, "Invalid Request: missing method", requestId);
+            return;
+        }
+
+        // Streaming methods produce text/event-stream, not a single JSON-RPC envelope
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        LOG.info("JSON-RPC dispatch: method={}, op={}", method, op);
+        if (op == A2AOperations.MESSAGE_STREAM) {
+            LOG.info("Entering handleJsonRpcStream for taskId={}", requestId);
+            try {
+                handleJsonRpcStream(exchange, params, requestId);
+                LOG.info("handleJsonRpcStream completed, contentType={}", exchange.getMessage().getHeader("Content-Type"));
+            } catch (Exception e) {
+                LOG.error("handleJsonRpcStream failed: {}", e.getMessage(), e);
+                writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR,
+                        "Internal error: " + e.getMessage(), requestId);
+            }
+            return;
+        }
+
+        Object result;
+        try {
+            result = dispatchJsonRpcMethod(method, params, exchange);
+        } catch (ServerBusyException e) {
+            writeJsonRpcError(exchange, -32000, e.getMessage(), requestId);
+            return;
+        } catch (IllegalArgumentException e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.METHOD_NOT_FOUND, "Method not found: " + method, requestId);
+            return;
+        } catch (Exception e) {
+            writeJsonRpcError(exchange, JsonRpcProtocol.INTERNAL_ERROR, "Internal error: " + e.getMessage(), requestId);
+            return;
+        }
+
+        byte[] response = jsonRpc.wrapJsonRpcResponse(result, requestId);
+        exchange.getMessage().setBody(response);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    private Object dispatchJsonRpcMethod(String method, Map<String, Object> params, Exchange exchange) throws Exception {
+        A2AOperations op = A2AOperations.fromMethodName(method);
+        if (op == null) {
+            throw new IllegalArgumentException("Unknown method: " + method);
+        }
+        return switch (op) {
+            case MESSAGE_SEND -> {
+                SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+                yield processSendMessage(request, exchange);
+            }
+            case TASK_GET -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processGetTask(taskId);
+            }
+            case TASK_CANCEL -> {
+                String taskId = params != null ? (String) params.get("id") : null;
+                yield processCancelTask(taskId);
+            }
+            case TASK_LIST -> {
+                Integer pageSize = params != null ? (Integer) params.get("pageSize") : null;
+                String contextId = params != null ? (String) params.get("contextId") : null;
+                yield processListTasks(contextId, pageSize);
+            }
+            case PUSH_CONFIG_CREATE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                TaskPushNotificationConfig config = OBJECT_MAPPER.convertValue(params, TaskPushNotificationConfig.class);
+                yield processPushConfigCreate(taskId, config);
+            }
+            case PUSH_CONFIG_GET -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigGet(taskId, configId);
+            }
+            case PUSH_CONFIG_LIST -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                yield processPushConfigList(taskId);
+            }
+            case PUSH_CONFIG_DELETE -> {
+                String taskId = params != null ? (String) params.get("taskId") : null;
+                String configId = params != null ? (String) params.get("id") : null;
+                yield processPushConfigDelete(taskId, configId);
+            }
+            default -> throw new IllegalArgumentException("Unsupported method: " + method);
+        };
+    }
+
+    private void writeJsonRpcError(Exchange exchange, int code, String message, Object requestId) throws Exception {
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        byte[] error = jsonRpc.wrapJsonRpcError(code, message, requestId);
+        exchange.getMessage().setBody(error);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    // ---- JSON-RPC streaming ----
+
+    private void handleJsonRpcStream(Exchange exchange, Map<String, Object> params, Object requestId) throws Exception {
+        if (!tryAcquirePermit()) {
+            writeJsonRpcError(exchange, -32000,
+                    "Agent at capacity: " + getEndpoint().getConfiguration().getMaxConcurrentTasks()
+                                                + " concurrent tasks",
+                    requestId);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.convertValue(params, SendMessageRequest.class);
+        String taskId = UUID.randomUUID().toString();
+        String contextId = request.getMessage().contextId();
+        if (contextId == null || contextId.isEmpty()) {
+            contextId = UUID.randomUUID().toString();
+        }
+
+        A2ATaskStore store = getEndpoint().getTaskStore();
+        store.put(taskId, Task.builder()
+                .id(taskId).contextId(contextId)
+                .status(new TaskStatus(TaskState.SUBMITTED)).build());
+
+        JsonRpcProtocol jsonRpc = (JsonRpcProtocol) getEndpoint().getProtocol();
+        final Object rpcRequestId = requestId;
+        Function<StreamResponse, String> jsonRpcEncoder = response -> {
+            byte[] envelope = jsonRpc.wrapJsonRpcResponse(response, rpcRequestId);
+            return "data: " + new String(envelope, StandardCharsets.UTF_8) + "\n\n";
+        };
+
+        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
+        QueueStreamEmitter emitter = new QueueStreamEmitter(taskId, contextId, queue, jsonRpcEncoder);
+        SseQueueInputStream inputStream = new SseQueueInputStream(
+                queue,
+                getEndpoint().getConfiguration().getAsyncTimeout());
+
+        A2ATaskSubscriber subscriber = new StreamSubscriber(emitter);
+        store.addSubscriber(taskId, subscriber);
+        inputStream.setOnClose(() -> {
+            store.removeSubscriber(taskId, subscriber);
+            emitter.close();
+        });
+
+        submitStreamProcessing(taskId, contextId, request, emitter);
+
+        exchange.getExchangeExtension().setStreamCacheDisabled(true);
+        exchange.getMessage().setBody(inputStream);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, SSE_CONTENT_TYPE);
+        exchange.getMessage().setHeader("Cache-Control", "no-cache");
+        exchange.getMessage().setHeader("Connection", "keep-alive");
+    }
+
+    // ---- REST handler methods ----
+
+    void handleAgentCardRequest(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        AgentCard card = getEndpoint().getResolvedCard();
+        byte[] cardJson = OBJECT_MAPPER.writeValueAsBytes(card);
+
+        exchange.getMessage().setBody(cardJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleSendMessage(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String requestBody = exchange.getMessage().getBody(String.class);
+        if (requestBody == null || requestBody.isBlank()) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("ContentTypeNotSupportedError", "Request body is empty"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        long maxSize = getEndpoint().getConfiguration().getMaxPayloadSize();
+        if (requestBody.length() > maxSize) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError(
+                            "ContentTypeNotSupportedError",
+                            "Request body exceeds maximum size: " + maxSize + " bytes"));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 413);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+            return;
+        }
+
+        SendMessageRequest request = OBJECT_MAPPER.readValue(requestBody, SendMessageRequest.class);
+
+        SendMessageResponse response;
+        try {
+            response = processSendMessage(request, exchange);
+        } catch (ServerBusyException e) {
+            writeServerBusyError(exchange);
+            return;
+        }
+
+        byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(response);
+        exchange.getMessage().setBody(responseJson);
+        exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+    }
+
+    void handleGetTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processGetTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleCancelTask(Exchange exchange) throws Exception {
+        filterInboundHeaders(exchange);
+        String taskId = extractTaskIdFromPath(exchange);
+
+        try {
+            Task task = processCancelTask(taskId);
+            byte[] responseJson = OBJECT_MAPPER.writeValueAsBytes(task);
+            exchange.getMessage().setBody(responseJson);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (TaskNotFoundException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotFoundError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 404);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        } catch (IllegalStateException e) {
+            byte[] errorJson = OBJECT_MAPPER.writeValueAsBytes(
+                    new A2AError("TaskNotCancelableError", e.getMessage()));
+            exchange.getMessage().setBody(errorJson);
+            exchange.getMessage().setHeader(Exchange.HTTP_RESPONSE_CODE, 409);
+            exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, A2AConstants.CONTENT_TYPE);
+        }
+    }
+
+    void handleListTasks(Exchange exchange) throws Exception {
+        Integer pageSize = exchange.getMessage().getHeader(A2AConstants.LIST_PAGE_SIZE, Integer.class);
+        String contextId = exchange.getMessage().getHeader(A2AConstants.LIST_CONTEXT_ID, String.class);
+
+        filterInboundHeaders(exchange);
+

Review Comment:
   **[Low]** `pageSize` and `contextId` are read from exchange headers **before** `filterInboundHeaders` is called (lines 734-735). An untrusted HTTP client could inject `CamelA2AListPageSize` and `CamelA2AListContextId` headers. While the impact is low (pageSize is capped at 100, contextId is just a filter key), the filter should run before any header reads — consistent with all other handlers in this class.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348908007


##########
components/camel-ai/camel-a2a/src/generated/resources/a2a.json:
##########
@@ -0,0 +1,15 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "a2a",
+    "title": "A2a",
+    "description": "Camel A2A component for Agent-to-Agent protocol communication",
+    "deprecated": false,
+    "firstVersion": "4.21.0",
+    "label": "ai,a2a",

Review Comment:
   ok let's keep only `ai` then



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348312076


##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.auth;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.SecurityScheme;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Authentication coordinator for the A2A component. Delegates to {@link A2ASecuritySchemeHandler} implementations based
+ * on the agent card's declared security schemes. Configuration parameters ({@code oauthProfile}, {@code bearerToken},
+ * {@code apiKey}) influence scheme selection priority.
+ */
+public class A2AAuthHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AAuthHandler.class);
+
+    private final A2AConfiguration configuration;
+    private final Map<String, A2ASecuritySchemeHandler> handlers;
+
+    public A2AAuthHandler(A2AConfiguration configuration, Map<String, A2ASecuritySchemeHandler> handlers) {
+        this.configuration = configuration;
+        this.handlers = handlers;
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange. Selects the appropriate scheme handler based on the resolved
+     * card's security schemes and configuration hints.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context) {
+        applyProducerAuth(exchange, context, null);
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange using the given card's security schemes.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context, AgentCard card) {
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                handler.applyCredentials(exchange, scheme, configuration, context);
+                return;
+            }
+        }
+
+        LOG.debug("No matching security scheme handler found — request will be unauthenticated");
+    }
+
+    /**
+     * Validate credentials from an incoming consumer request. Returns a user profile map on success, or null if
+     * {@code validateAuth} is disabled. Throws {@link SecurityException} if validation fails.
+     */
+    public Map<String, Object> validateConsumerAuth(Exchange exchange, AgentCard card) {
+        if (!configuration.isValidateAuth()) {
+            return null;
+        }
+
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+        if (schemes.isEmpty()) {
+            LOG.warn("validateAuth is enabled but agent card declares no security schemes — allowing request");
+            return null;
+        }
+
+        SecurityException lastFailure = null;
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                try {
+                    return handler.validateCredentials(exchange, scheme, configuration);
+                } catch (SecurityException e) {
+                    lastFailure = e;
+                }
+            }
+        }
+
+        throw lastFailure != null
+                ? lastFailure
+                : new SecurityException("No security scheme handler available for the declared schemes");
+    }
+
+    /**
+     * Resolves and prioritizes security schemes from the card, using config hints to determine order. If
+     * {@code oauthProfile} is configured, OAuth2/OIDC schemes come first. If {@code bearerToken} is configured, HTTP
+     * bearer schemes come first. If {@code apiKey} is configured, API key schemes come first.
+     */
+    List<Map.Entry<String, SecurityScheme>> resolveSchemes(AgentCard card) {
+        if (card == null || card.getSecuritySchemes() == null || card.getSecuritySchemes().isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<Map.Entry<String, SecurityScheme>> all = new ArrayList<>(card.getSecuritySchemes().entrySet());
+        String preferredType = resolvePreferredSchemeType();
+        if (preferredType != null) {
+            all.sort((a, b) -> {
+                boolean aMatch = preferredType.equals(a.getValue().getType());
+                boolean bMatch = preferredType.equals(b.getValue().getType());
+                if (aMatch == bMatch) {
+                    return 0;
+                }
+                return aMatch ? -1 : 1;
+            });
+        }
+        return all;
+    }
+
+    private String resolvePreferredSchemeType() {
+        if (configuration.getOauthProfile() != null) {
+            return "oauth2";
+        }
+        if (configuration.getBearerToken() != null) {
+            return "http";

Review Comment:
   @apupier there's a misunderstanding here, the `"http"` here is **not** the transport protocol, it's the [OpenAPI 3.0 Security Scheme `type`](https://spec.openapis.org/oas/v3.0.3#security-scheme-object) constant, which the A2A protocol adopts for its `securitySchemes` field.
   
     The `type` enum has four values:
   
     | `type` value | Meaning |
     |---|---|
     | `"apiKey"` | API key (header, query, or cookie) |
     | `"http"` | RFC 7235 HTTP Authentication — Bearer, Basic, Digest, etc. |
     | `"oauth2"` | OAuth 2.0 flows |
     | `"openIdConnect"` | OpenID Connect discovery |
   
     So `"http"` means "this scheme uses the `Authorization` header" (e.g. `Authorization: Bearer <token>`), which works identically over both HTTP and HTTPS. The actual
     transport encryption is determined by the endpoint URI scheme, not by this field.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["zbendhiba (via GitHub)" <gi...@apache.org>]

zbendhiba commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3354555016


##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.auth;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.SecurityScheme;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Authentication coordinator for the A2A component. Delegates to {@link A2ASecuritySchemeHandler} implementations based
+ * on the agent card's declared security schemes. Configuration parameters ({@code oauthProfile}, {@code bearerToken},
+ * {@code apiKey}) influence scheme selection priority.
+ */
+public class A2AAuthHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AAuthHandler.class);
+
+    private final A2AConfiguration configuration;
+    private final Map<String, A2ASecuritySchemeHandler> handlers;
+
+    public A2AAuthHandler(A2AConfiguration configuration, Map<String, A2ASecuritySchemeHandler> handlers) {
+        this.configuration = configuration;
+        this.handlers = handlers;
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange. Selects the appropriate scheme handler based on the resolved
+     * card's security schemes and configuration hints.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context) {
+        applyProducerAuth(exchange, context, null);
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange using the given card's security schemes.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context, AgentCard card) {
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                handler.applyCredentials(exchange, scheme, configuration, context);
+                return;
+            }
+        }
+
+        LOG.debug("No matching security scheme handler found — request will be unauthenticated");
+    }
+
+    /**
+     * Validate credentials from an incoming consumer request. Returns a user profile map on success, or null if
+     * {@code validateAuth} is disabled. Throws {@link SecurityException} if validation fails.
+     */
+    public Map<String, Object> validateConsumerAuth(Exchange exchange, AgentCard card) {
+        if (!configuration.isValidateAuth()) {
+            return null;
+        }
+
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+        if (schemes.isEmpty()) {
+            LOG.warn("validateAuth is enabled but agent card declares no security schemes — allowing request");
+            return null;
+        }
+
+        SecurityException lastFailure = null;
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                try {
+                    return handler.validateCredentials(exchange, scheme, configuration);
+                } catch (SecurityException e) {
+                    lastFailure = e;
+                }
+            }
+        }
+
+        throw lastFailure != null
+                ? lastFailure
+                : new SecurityException("No security scheme handler available for the declared schemes");
+    }
+
+    /**
+     * Resolves and prioritizes security schemes from the card, using config hints to determine order. If
+     * {@code oauthProfile} is configured, OAuth2/OIDC schemes come first. If {@code bearerToken} is configured, HTTP
+     * bearer schemes come first. If {@code apiKey} is configured, API key schemes come first.
+     */
+    List<Map.Entry<String, SecurityScheme>> resolveSchemes(AgentCard card) {
+        if (card == null || card.getSecuritySchemes() == null || card.getSecuritySchemes().isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<Map.Entry<String, SecurityScheme>> all = new ArrayList<>(card.getSecuritySchemes().entrySet());
+        String preferredType = resolvePreferredSchemeType();
+        if (preferredType != null) {
+            all.sort((a, b) -> {
+                boolean aMatch = preferredType.equals(a.getValue().getType());
+                boolean bMatch = preferredType.equals(b.getValue().getType());
+                if (aMatch == bMatch) {
+                    return 0;
+                }
+                return aMatch ? -1 : 1;
+            });
+        }
+        return all;
+    }
+
+    private String resolvePreferredSchemeType() {
+        if (configuration.getOauthProfile() != null) {
+            return "oauth2";
+        }
+        if (configuration.getBearerToken() != null) {
+            return "http";

Review Comment:
   Maybe we could some meaningful variables, declared in a section that explains this spec. WDYT ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["davsclaus (via GitHub)" <gi...@apache.org>]

davsclaus commented on PR #23711:
URL: https://github.com/apache/camel/pull/23711#issuecomment-4667831837

   LGTM @luigidemasi its a massive work so when you think its ready then go ahead and merge - we need it in 4.21 so we have both MCP and A2A


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["apupier (via GitHub)" <gi...@apache.org>]

apupier commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3342485493


##########
components/camel-ai/camel-a2a/src/generated/resources/a2a.json:
##########
@@ -0,0 +1,15 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "a2a",
+    "title": "A2a",
+    "description": "Camel A2A component for Agent-to-Agent protocol communication",
+    "deprecated": false,
+    "firstVersion": "4.21.0",
+    "label": "ai,a2a",

Review Comment:
   What is the reason for the a2a label? Plan to have several agent to agents components?
   
   What about ai and agent labels? maybe we will have some components acting as agents or doing agent to something else communication?



##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.auth;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.SecurityScheme;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Authentication coordinator for the A2A component. Delegates to {@link A2ASecuritySchemeHandler} implementations based
+ * on the agent card's declared security schemes. Configuration parameters ({@code oauthProfile}, {@code bearerToken},
+ * {@code apiKey}) influence scheme selection priority.
+ */
+public class A2AAuthHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger(A2AAuthHandler.class);
+
+    private final A2AConfiguration configuration;
+    private final Map<String, A2ASecuritySchemeHandler> handlers;
+
+    public A2AAuthHandler(A2AConfiguration configuration, Map<String, A2ASecuritySchemeHandler> handlers) {
+        this.configuration = configuration;
+        this.handlers = handlers;
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange. Selects the appropriate scheme handler based on the resolved
+     * card's security schemes and configuration hints.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context) {
+        applyProducerAuth(exchange, context, null);
+    }
+
+    /**
+     * Apply authentication headers to a producer exchange using the given card's security schemes.
+     */
+    public void applyProducerAuth(Exchange exchange, CamelContext context, AgentCard card) {
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                handler.applyCredentials(exchange, scheme, configuration, context);
+                return;
+            }
+        }
+
+        LOG.debug("No matching security scheme handler found — request will be unauthenticated");
+    }
+
+    /**
+     * Validate credentials from an incoming consumer request. Returns a user profile map on success, or null if
+     * {@code validateAuth} is disabled. Throws {@link SecurityException} if validation fails.
+     */
+    public Map<String, Object> validateConsumerAuth(Exchange exchange, AgentCard card) {
+        if (!configuration.isValidateAuth()) {
+            return null;
+        }
+
+        List<Map.Entry<String, SecurityScheme>> schemes = resolveSchemes(card);
+        if (schemes.isEmpty()) {
+            LOG.warn("validateAuth is enabled but agent card declares no security schemes — allowing request");
+            return null;
+        }
+
+        SecurityException lastFailure = null;
+        for (Map.Entry<String, SecurityScheme> entry : schemes) {
+            SecurityScheme scheme = entry.getValue();
+            A2ASecuritySchemeHandler handler = handlers.get(scheme.getType());
+            if (handler != null) {
+                try {
+                    return handler.validateCredentials(exchange, scheme, configuration);
+                } catch (SecurityException e) {
+                    lastFailure = e;
+                }
+            }
+        }
+
+        throw lastFailure != null
+                ? lastFailure
+                : new SecurityException("No security scheme handler available for the declared schemes");
+    }
+
+    /**
+     * Resolves and prioritizes security schemes from the card, using config hints to determine order. If
+     * {@code oauthProfile} is configured, OAuth2/OIDC schemes come first. If {@code bearerToken} is configured, HTTP
+     * bearer schemes come first. If {@code apiKey} is configured, API key schemes come first.
+     */
+    List<Map.Entry<String, SecurityScheme>> resolveSchemes(AgentCard card) {
+        if (card == null || card.getSecuritySchemes() == null || card.getSecuritySchemes().isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<Map.Entry<String, SecurityScheme>> all = new ArrayList<>(card.getSecuritySchemes().entrySet());
+        String preferredType = resolvePreferredSchemeType();
+        if (preferredType != null) {
+            all.sort((a, b) -> {
+                boolean aMatch = preferredType.equals(a.getValue().getType());
+                boolean bMatch = preferredType.equals(b.getValue().getType());
+                if (aMatch == bMatch) {
+                    return 0;
+                }
+                return aMatch ? -1 : 1;
+            });
+        }
+        return all;
+    }
+
+    private String resolvePreferredSchemeType() {
+        if (configuration.getOauthProfile() != null) {
+            return "oauth2";
+        }
+        if (configuration.getBearerToken() != null) {
+            return "http";

Review Comment:
   Is it for both http and https?



##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/model/Task.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.model;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+/**
+ * A2A protocol task.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record Task(
+        String id,
+        String contextId,
+        TaskStatus status,
+        List<Artifact> artifacts,
+        List<Message> history,
+        Map<String, Object> metadata) {
+
+    /**
+     * Convenience accessor for Camel OGNL: ${body.latest.parts[0].text}
+     */
+    public Message latest() {
+        return (history != null && !history.isEmpty()) ? history.get(history.size() - 1) : null;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static Builder builder(Task task) {
+        return new Builder()
+                .id(task.id)
+                .contextId(task.contextId)
+                .status(task.status)
+                .artifacts(task.artifacts != null ? new ArrayList<>(task.artifacts) : null)
+                .history(task.history != null ? new ArrayList<>(task.history) : null)
+                .metadata(task.metadata != null ? new HashMap<>(task.metadata) : null);

Review Comment:
   What about providing empty List or Map instead of null?
   
   it avoids having to check for nullness



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi commented on PR #23711:
URL: https://github.com/apache/camel/pull/23711#issuecomment-4706796957

   > Are we getting in shape for this to be included next week
   @davsclaus yes, it is almost ready and I think we are in good shape for next week. Before marking it ready, I want to finish and incorporate the OAuth authentication support for HTTP consumers from CAMEL-23685 (#23866) and CAMEL-23723 (#23926), so A2A lands with the right auth path for 4.21. 
   I will update the PR later today.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi merged PR #23711:
URL: https://github.com/apache/camel/pull/23711


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-23063: Add camel-a2a component for Agent-to-Agent (A2A) protoco… [camel]

["luigidemasi (via GitHub)" <gi...@apache.org>]

luigidemasi commented on code in PR #23711:
URL: https://github.com/apache/camel/pull/23711#discussion_r3348401575


##########
components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/model/Task.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.a2a.model;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+/**
+ * A2A protocol task.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record Task(
+        String id,
+        String contextId,
+        TaskStatus status,
+        List<Artifact> artifacts,
+        List<Message> history,
+        Map<String, Object> metadata) {
+
+    /**
+     * Convenience accessor for Camel OGNL: ${body.latest.parts[0].text}
+     */
+    public Message latest() {
+        return (history != null && !history.isEmpty()) ? history.get(history.size() - 1) : null;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static Builder builder(Task task) {
+        return new Builder()
+                .id(task.id)
+                .contextId(task.contextId)
+                .status(task.status)
+                .artifacts(task.artifacts != null ? new ArrayList<>(task.artifacts) : null)
+                .history(task.history != null ? new ArrayList<>(task.history) : null)
+                .metadata(task.metadata != null ? new HashMap<>(task.metadata) : null);

Review Comment:
   This is a protocol model class serialized by Jackson, `null` fields are omitted from JSON output while empty collections serialize as `[]` / `{}`. 
   The A2A spec distinguishes between "field absent" (not provided) and "field present but empty" (explicitly no items), so the copy constructor preserves the original semantics.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org