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

[via GitHub]

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-unsubscr...@camel.apache.org

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