This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 66b8fee2f4 feat(agent-gateway): record logical MCP tool call metrics
(#7522)
66b8fee2f4 is described below
commit 66b8fee2f4046990ee377221a3d39ebbd8f94547
Author: BobSong <[email protected]>
AuthorDate: Sat Oct 10 09:44:37 2026 +0800
feat(agent-gateway): record logical MCP tool call metrics (#7522)
Co-authored-by: BobSong-dev <[email protected]>
---
.../apache/shenyu/common/constant/Constants.java | 5 +
.../common/metrics/AgentMcpCallObserver.java | 52 ++++
.../shenyu-plugin-agent-gateway/README.md | 40 +++
.../gateway/protocol/AgentMcpCallObservation.java | 87 ++++++
.../gateway/protocol/AgentMcpHttpHandler.java | 13 +-
.../protocol/AgentMcpCallObservationTest.java | 346 +++++++++++++++++++++
.../shenyu/plugin/metrics/MetricsPlugin.java | 7 +
.../shenyu/plugin/metrics/constant/LabelNames.java | 6 +
.../plugin/metrics/reporter/MetricsReporter.java | 2 +
.../shenyu/plugin/metrics/MetricsPluginTest.java | 49 +++
.../metrics/reporter/MetricsReporterTest.java | 8 +-
11 files changed, 609 insertions(+), 6 deletions(-)
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java
index b4654e8e8a..1f58a4916a 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java
@@ -896,6 +896,11 @@ public interface Constants {
*/
String METRICS_HYSTRIX = "metricsHystrix";
+ /**
+ * Optional server-installed MCP terminal observation callback.
+ */
+ String METRICS_AGENT_MCP_CALL = "metricsAgentMcpCall";
+
/**
* The constant shenyu namespace id.
*/
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/AgentMcpCallObserver.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/AgentMcpCallObserver.java
new file mode 100644
index 0000000000..3c4efb1b80
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/AgentMcpCallObserver.java
@@ -0,0 +1,52 @@
+/*
+ * 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.shenyu.common.metrics;
+
+/**
+ * Server-side terminal observation for a parsed, authenticated MCP tools/call.
+ * Implementations must be nonblocking and must not retain request data.
+ */
+@FunctionalInterface
+public interface AgentMcpCallObserver {
+
+ /**
+ * Record one logical call after execution and response writing terminate.
+ *
+ * @param outcome bounded terminal classification
+ * @param elapsedMillis monotonic elapsed time since the validated call
was parsed
+ */
+ void record(Outcome outcome, long elapsedMillis);
+
+ /**
+ * Bounded labels; no tool names, principals or client-controlled
identifiers.
+ */
+ enum Outcome {
+ /** Successful tool result and completed response write. */
+ SUCCESS,
+ /** Completed MCP result with isError=true. */
+ TOOL_ERROR,
+ /** Protocol or authorization rejection after a valid call was parsed.
*/
+ REJECTED,
+ /** Internal, encoding or response-writing failure. */
+ SERVER_ERROR,
+ /** The gateway request deadline expired. */
+ TIMEOUT,
+ /** The downstream subscription was cancelled. */
+ CANCELLED
+ }
+}
diff --git a/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
b/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
index ecc56fbbdb..38e63a83a0 100644
--- a/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
+++ b/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
@@ -262,6 +262,46 @@ the request body, credentials, response content, or
mutable plugin-chain state.
Each subscription gets its own context. The same exchange should not be
subscribed concurrently, because exchange attributes are shared by that
request.
+## Tool call metrics
+
+When the existing ShenYu Metrics plugin runs before Agent Gateway, it installs
+a server-side terminal callback. No new exporter, endpoint, dependency or Admin
+configuration is required. Without that plugin, calls behave as before and no
+MCP statistics are recorded.
+
+`shenyu_agent_mcp_calls_total{outcome}` counts one logical call per
subscription,
+starting only after authentication and valid `tools/call` parsing. It includes
+subsequent authorization/argument rejection, not just provider invocations.
+Discovery, listing, malformed input and pre-authentication refusals remain in
+the existing HTTP metrics; they are not tool calls. Retries and upstream
attempts
+are not introduced or counted separately.
+
+The six fixed outcomes are `success`, `tool_error` (including invalid tool
+arguments reported as MCP `isError=true`), `rejected`, `server_error`,
`timeout`
+and `cancelled`. A result computed by a tool is not a successful call until its
+response write completes. Encoding/write failures count as `server_error`;
+deadline expiry and downstream cancellation are distinct. Each subscription
+records at most one terminal event, even with repeated RPC ids or late signals.
+
+`shenyu_agent_mcp_call_latency_millis{outcome}` uses monotonic elapsed time
from
+validated parsing through the terminal response write/error/cancellation. Its
+`_sum` and `_count` can be used for mean latency in milliseconds. This first
+version reuses the existing histogram registration/buckets; it does not claim
+a new percentile policy. No subject, tool name, RPC/request id, configuration
+version, arguments or response body is a metric label.
+
+For a time window, the logical-call failure ratio is
+`(tool_error + server_error + timeout) / (success + tool_error + server_error
+ timeout)`.
+Use rates/increases of the terminal counter, not cumulative process totals;
+report authorization refusals and cancellations separately. An empty
denominator
+means no applicable observations, not a zero failure rate. Callback failures
are
+best-effort observation failures and do not change a response or trigger
retries.
+
+No token or cost metric is emitted: MCP does not supply a trusted model-usage
+source here, and missing usage must not be represented as zero or estimated
from
+tool-call count/result length. Remote-catalog-specific labels, per-tool
metrics,
+in-flight gauges and persistent reporting remain outside this first version.
+
## Scope
This module provides LLM request correlation and a tools-only MCP entry with
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
new file mode 100644
index 0000000000..2e47c4354d
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
@@ -0,0 +1,87 @@
+/*
+ * 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.shenyu.plugin.agent.gateway.protocol;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.SignalType;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * One subscription's bounded observation state, independent of RPC ids and
sessions.
+ */
+final class AgentMcpCallObservation {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AgentMcpCallObservation.class);
+
+ private final AgentMcpCallObserver observer;
+
+ private final AtomicBoolean recorded = new AtomicBoolean();
+
+ private volatile long started;
+
+ private volatile boolean call;
+
+ private volatile Outcome outcome = Outcome.SERVER_ERROR;
+
+ AgentMcpCallObservation(final Object observer) {
+ this.observer = observer instanceof AgentMcpCallObserver ?
(AgentMcpCallObserver) observer : null;
+ }
+
+ void parsed(final AgentMcpRequest request) {
+ if (Objects.nonNull(observer) &&
"tools/call".equals(request.getMethod())) {
+ started = System.nanoTime();
+ call = true;
+ }
+ }
+
+ void response(final ObjectNode response) {
+ outcome = response.path("result").path("isError").asBoolean(false) ?
Outcome.TOOL_ERROR : Outcome.SUCCESS;
+ }
+
+ void failure(final Throwable error) {
+ if (error instanceof TimeoutException) {
+ outcome = Outcome.TIMEOUT;
+ } else if (error instanceof AgentMcpProtocolException &&
((AgentMcpProtocolException) error).getHttpStatus() < 500) {
+ outcome = Outcome.REJECTED;
+ } else {
+ outcome = Outcome.SERVER_ERROR;
+ }
+ }
+
+ void finish(final SignalType signal) {
+ if (!call || !recorded.compareAndSet(false, true)) {
+ return;
+ }
+ Outcome terminal = signal == SignalType.CANCEL ? Outcome.CANCELLED :
outcome;
+ long millis = TimeUnit.NANOSECONDS.toMillis(Math.max(0,
System.nanoTime() - started));
+ try {
+ observer.record(terminal, millis);
+ } catch (RuntimeException error) {
+ // Observability must not fail a call or log potentially sensitive
callback errors.
+ LOG.debug("MCP call statistics callback failed");
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
index 610fcebd31..b0edc084dc 100644
---
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.plugin.agent.gateway.protocol;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.AgentGatewayMcpConfig;
import org.apache.shenyu.plugin.agent.gateway.AgentTrafficContext;
import
org.apache.shenyu.plugin.agent.gateway.security.AgentMcpSecurityResolver;
@@ -79,6 +80,7 @@ public final class AgentMcpHttpHandler {
final AgentTrafficContext traffic, final long
configurationVersion) {
return Mono.defer(() -> {
final Instant deadline =
Instant.now().plusMillis(config.getTimeoutMs());
+ final AgentMcpCallObservation observation = new
AgentMcpCallObservation(exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL));
AtomicReference<JsonNode> rpcId = new AtomicReference<>();
return Mono.fromRunnable(() -> preflight(exchange, config))
.then(Mono.defer(() -> securityResolver.resolve(exchange))
@@ -86,12 +88,19 @@ public final class AgentMcpHttpHandler {
.flatMap(identity -> read(exchange, config).map(body ->
parser.parse(body, exchange.getRequest().getHeaders(),
config.getMaxRequestBytes()))
.flatMap(request -> {
rpcId.set(request.getId());
+ observation.parsed(request);
return dispatcher.dispatch(request, () -> new
AgentMcpExecutionContext(traffic.getRequestId(), identity.getSubject(),
traffic.getRuleId(),
configurationVersion, config.getAllowedTools(), identity.getToolGrants(),
deadline));
}))
- .flatMap(response -> write(exchange, response, config,
"sse".equals(config.getResponseMode()), 200))
+ .flatMap(response -> {
+ observation.response(response);
+ return write(exchange, response, config,
"sse".equals(config.getResponseMode()), 200);
+ })
.timeout(Duration.ofMillis(config.getTimeoutMs()))
- .onErrorResume(error -> handleFailure(exchange, config,
rpcId.get(), error));
+ .doOnError(observation::failure)
+ .onErrorResume(error -> handleFailure(exchange, config,
rpcId.get(), error))
+ .doOnError(observation::failure)
+ .doFinally(observation::finish);
});
}
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
new file mode 100644
index 0000000000..dbde556e0f
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
@@ -0,0 +1,346 @@
+/*
+ * 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.shenyu.plugin.agent.gateway.protocol;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.dto.AgentGatewayMcpConfig;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.apache.shenyu.plugin.agent.gateway.AgentTrafficContext;
+import org.apache.shenyu.plugin.agent.gateway.security.AgentMcpIdentity;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolExecutionException;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolProvider;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolRegistry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Logical call observation through execution, encoding, writing and
cancellation.
+ */
+class AgentMcpCallObservationTest {
+
+ private final AgentTrafficContext traffic = new
AgentTrafficContext("private-id", "mcp", "selector", "rule");
+
+ @ParameterizedTest
+ @CsvSource({"json,SUCCESS", "sse,SUCCESS", "json,TOOL_ERROR",
"sse,TOOL_ERROR"})
+ void recordsBusinessOutcomeNotOnlyHttpStatus(final String mode, final
Outcome expected) {
+ AgentToolProvider tool = tool(expected == Outcome.TOOL_ERROR ?
Mono.error(new AgentToolExecutionException("private-diagnostic")) :
Mono.just(new JsonObject()));
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ StepVerifier.create(handler(tool, Set.of("read")).handle(exchange,
config(mode, 30000), traffic, 1)).verifyComplete();
+ observer.assertOnly(expected);
+ assertEquals(200, exchange.getResponse().getStatusCode().value());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsRejectedCallWithoutInvokingProvider(final String mode) {
+ AtomicInteger invocations = new AtomicInteger();
+ AgentToolProvider tool = tool(Mono.defer(() -> {
+ invocations.incrementAndGet();
+ return Mono.just(new JsonObject());
+ }));
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ StepVerifier.create(handler(tool, Set.of()).handle(exchange,
config(mode, 30000), traffic, 1)).verifyComplete();
+ observer.assertOnly(Outcome.REJECTED);
+ assertEquals(0, invocations.get());
+ assertEquals(403, exchange.getResponse().getStatusCode().value());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsInternalErrorNotRecoveredHttpCompletion(final String mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ StepVerifier.create(handler(tool(Mono.error(new
IllegalStateException("private-error"))), Set.of("read"))
+ .handle(exchange, config(mode, 30000), traffic,
1)).verifyComplete();
+ observer.assertOnly(Outcome.SERVER_ERROR);
+ assertEquals(500, exchange.getResponse().getStatusCode().value());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsDeadlineOnceInsteadOfProviderCancellation(final String mode) {
+ Recording observer = new Recording();
+ AtomicBoolean cancelled = new AtomicBoolean();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ AgentMcpHttpHandler handler =
handler(tool(Mono.<JsonObject>never().doOnCancel(() -> cancelled.set(true))),
Set.of("read"));
+ StepVerifier.withVirtualTime(() -> handler.handle(exchange,
config(mode, 100), traffic, 1))
+ .thenAwait(Duration.ofMillis(101)).verifyComplete();
+ observer.assertOnly(Outcome.TIMEOUT);
+ assertTrue(cancelled.get());
+ assertEquals(504, exchange.getResponse().getStatusCode().value());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsDownstreamCancellationDuringExecution(final String mode) {
+ Recording observer = new Recording();
+ AtomicBoolean started = new AtomicBoolean();
+ AtomicBoolean cancelled = new AtomicBoolean();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ AgentMcpHttpHandler handler =
handler(tool(Mono.<JsonObject>never().doOnSubscribe(subscription ->
started.set(true))
+ .doOnCancel(() -> cancelled.set(true))), Set.of("read"));
+ StepVerifier.create(handler.handle(exchange, config(mode, 30000),
traffic, 1)).then(() -> assertTrue(started.get())).thenCancel().verify();
+ observer.assertOnly(Outcome.CANCELLED);
+ assertTrue(cancelled.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void doesNotCountComputedResultAsSuccessDuringStalledWriting(final String
mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release).then(Mono.never()));
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read")).handle(exchange, config(mode, 30000), traffic, 1))
+ .then(() ->
assertTrue(observer.outcomes.isEmpty())).thenCancel().verify();
+ observer.assertOnly(Outcome.CANCELLED);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsResponseWriteErrorNotComputedSuccess(final String mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release)
+ .then(Mono.error(new
IllegalStateException("private-write-error"))));
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read")).handle(exchange, config(mode, 30000), traffic, 1))
+ .expectError(IllegalStateException.class).verify();
+ observer.assertOnly(Outcome.SERVER_ERROR);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void recordsResponseLimitFailureWithoutLeakingSuccess(final String mode) {
+ JsonObject result = new JsonObject();
+ result.addProperty("large", "x".repeat(2000));
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ JsonObject settings =
JsonParser.parseString("{\"allowedTools\":[\"read\"],\"maxResponseBytes\":1024,\"responseMode\":\""
+ mode + "\"}").getAsJsonObject();
+ StepVerifier.create(handler(tool(Mono.just(result)),
Set.of("read")).handle(exchange, AgentGatewayMcpConfig.parse(settings, true),
traffic, 1)).verifyComplete();
+ observer.assertOnly(Outcome.SERVER_ERROR);
+ assertEquals(500, exchange.getResponse().getStatusCode().value());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void timeoutWhileWritingIsNotSuccessfulToolCompletion(final String mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release).then(Mono.never()));
+ AgentMcpHttpHandler handler = handler(tool(Mono.just(new
JsonObject())), Set.of("read"));
+ StepVerifier.withVirtualTime(() -> handler.handle(exchange,
config(mode, 100), traffic, 1))
+
.thenAwait(Duration.ofMillis(101)).expectError(TimeoutException.class).verify();
+ observer.assertOnly(Outcome.TIMEOUT);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"server/discover", "tools/list"})
+ void excludesDiscoveryAndListing(final String method) {
+ Recording observer = new Recording();
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read"))
+ .handle(exchange(method, observer), config("json", 30000),
traffic, 1)).verifyComplete();
+ assertTrue(observer.outcomes.isEmpty());
+ }
+
+ @Test
+ void excludesUnparsedAndUnauthenticatedRequests() {
+ Recording observer = new Recording();
+ MockServerWebExchange malformed =
MockServerWebExchange.from(MockServerHttpRequest.post("/agent")
+
.contentType(MediaType.APPLICATION_JSON).header(HttpHeaders.ACCEPT,
"application/json, text/event-stream").body("{}"));
+ malformed.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
observer);
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read"))
+ .handle(malformed, config("json", 30000), traffic,
1)).verifyComplete();
+ AgentMcpHttpHandler unauthenticated = new AgentMcpHttpHandler(new
AgentMcpDispatcher(new AgentToolRegistry(List.of()), "test", "1"), ignored ->
Mono.empty());
+ StepVerifier.create(unauthenticated.handle(exchange("tools/call",
observer), config("json", 30000), traffic, 1)).verifyComplete();
+ assertTrue(observer.outcomes.isEmpty());
+ }
+
+ @Test
+ void callbackFailuresDoNotChangeSuccessfulResponse() {
+ AtomicInteger records = new AtomicInteger();
+ AgentMcpCallObserver observer = (outcome, millis) -> {
+ records.incrementAndGet();
+ throw new IllegalStateException("private-observer-error");
+ };
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read"))
+ .handle(exchange, config("json", 30000), traffic,
1)).verifyComplete();
+ assertEquals(1, records.get());
+ assertEquals(200, exchange.getResponse().getStatusCode().value());
+ }
+
+ @Test
+ void concurrentSameRpcIdsHaveIndependentTerminalRecords() {
+ Recording observer = new Recording();
+ AgentMcpHttpHandler handler = handler(tool(Mono.just(new
JsonObject()).publishOn(Schedulers.parallel())), Set.of("read"));
+ StepVerifier.create(Flux.range(0, 32).flatMap(index ->
handler.handle(exchange("tools/call", observer), config("json", 30000),
traffic, index), 32))
+ .verifyComplete();
+ await().atMost(Duration.ofSeconds(3)).until(() ->
observer.outcomes.size() == 32);
+ assertEquals(32, observer.outcomes.size());
+ assertTrue(observer.outcomes.stream().allMatch(outcome -> outcome ==
Outcome.SUCCESS));
+ }
+
+ @Test
+ void aFreshSubscriptionDoesNotReuseThePreviousTerminalRecord() {
+ Recording observer = new Recording();
+ AgentMcpHttpHandler handler = handler(tool(Mono.just(new
JsonObject())), Set.of("read"));
+ Mono<Void> call = Mono.defer(() ->
handler.handle(exchange("tools/call", observer), config("json", 30000),
traffic, 1));
+ StepVerifier.create(call).verifyComplete();
+ StepVerifier.create(call).verifyComplete();
+ assertEquals(List.of(Outcome.SUCCESS, Outcome.SUCCESS),
List.copyOf(observer.outcomes));
+ }
+
+ @Test
+ void cancellingOneCallDoesNotFinalizeAnotherWithTheSameRpcId() {
+ Recording cancelled = new Recording();
+ Recording successful = new Recording();
+ AtomicBoolean started = new AtomicBoolean();
+ Sinks.One<JsonObject> pending = Sinks.one();
+ StepVerifier.create(handler(tool(pending.asMono()), Set.of("read"))
+ .handle(exchange("tools/call", successful), config("json",
30000), traffic, 1))
+ .then(() -> {
+
StepVerifier.create(handler(tool(Mono.<JsonObject>never().doOnSubscribe(subscription
-> started.set(true))), Set.of("read"))
+ .handle(exchange("tools/call", cancelled),
config("json", 30000), traffic, 1))
+ .then(() ->
assertTrue(started.get())).thenCancel().verify();
+ cancelled.assertOnly(Outcome.CANCELLED);
+ assertTrue(successful.outcomes.isEmpty());
+ assertEquals(Sinks.EmitResult.OK, pending.tryEmitValue(new
JsonObject()));
+ }).verifyComplete();
+ successful.assertOnly(Outcome.SUCCESS);
+ }
+
+ @Test
+ void lateSignalsCannotRecordATerminalTwice() {
+ Recording observer = new Recording();
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation(observer);
+ AgentMcpRequest request = new
AgentMcpRequestParser().parse(body("tools/call").getBytes(StandardCharsets.UTF_8),
+ exchange("tools/call", observer).getRequest().getHeaders(),
262144);
+ observation.parsed(request);
+ observation.response(new ObjectMapper().createObjectNode());
+ observation.finish(SignalType.CANCEL);
+ observation.finish(SignalType.ON_COMPLETE);
+ observation.failure(new IllegalStateException("late-error"));
+ observation.finish(SignalType.ON_ERROR);
+ observer.assertOnly(Outcome.CANCELLED);
+ }
+
+ @Test
+ void absentOrWrongTypeCallbackDoesNotRequireMetricsPlugin() {
+ MockServerWebExchange exchange = exchange("tools/call", null);
+ exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
"unusable-callback");
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read"))
+ .handle(exchange, config("json", 30000), traffic,
1)).verifyComplete();
+ assertEquals(200, exchange.getResponse().getStatusCode().value());
+ }
+
+ @Test
+ void cancellationBeforeParsingIsNotAToolCall() {
+ Recording observer = new Recording();
+ AgentMcpHttpHandler handler = handler(tool(Mono.never()),
Set.of("read"));
+ StepVerifier.create(handler.handle(exchange("tools/call", observer),
config("json", 30000), traffic, 1)).thenCancel().verify();
+ assertTrue(observer.outcomes.isEmpty());
+ }
+
+ private AgentMcpHttpHandler handler(final AgentToolProvider tool, final
Set<String> grants) {
+ return new AgentMcpHttpHandler(new AgentMcpDispatcher(new
AgentToolRegistry(List.of(tool)), "test", "1"),
+ ignored -> Mono.just(new AgentMcpIdentity("private-subject",
grants)));
+ }
+
+ private AgentToolProvider tool(final Mono<JsonObject> action) {
+ AgentToolProvider tool = mock(AgentToolProvider.class);
+ when(tool.getName()).thenReturn("read");
+ when(tool.getDescription()).thenReturn("Read test data");
+
when(tool.getInputSchema()).thenReturn(JsonParser.parseString("{\"type\":\"object\"}").getAsJsonObject());
+ when(tool.getRequiredClientCapabilities()).thenReturn(new
JsonObject());
+ when(tool.invoke(any())).thenReturn(action);
+ return tool;
+ }
+
+ private AgentGatewayMcpConfig config(final String mode, final int timeout)
{
+ return
AgentGatewayMcpConfig.parse(JsonParser.parseString("{\"allowedTools\":[\"read\"],\"responseMode\":\""
+ mode
+ + "\",\"timeoutMs\":" + timeout + "}").getAsJsonObject(),
true);
+ }
+
+ private MockServerWebExchange exchange(final String method, final
AgentMcpCallObserver observer) {
+ MockServerHttpRequest.BodyBuilder builder =
MockServerHttpRequest.post("/agent").contentType(MediaType.APPLICATION_JSON)
+ .header(HttpHeaders.ACCEPT, "application/json,
text/event-stream").header("MCP-Protocol-Version",
AgentMcpRequestParser.VERSION).header("Mcp-Method", method);
+ if ("tools/call".equals(method)) {
+ builder.header("Mcp-Name", "read");
+ }
+ MockServerWebExchange exchange =
MockServerWebExchange.from(builder.body(body(method)));
+ if (Objects.nonNull(observer)) {
+ exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
observer);
+ }
+ return exchange;
+ }
+
+ private String body(final String method) {
+ return "{\"jsonrpc\":\"2.0\",\"id\":\"same-id\",\"method\":\"" +
method + "\",\"params\":{\"name\":\"read\",\"_meta\":{"
+ +
"\"io.modelcontextprotocol/protocolVersion\":\"2026-07-28\",\"io.modelcontextprotocol/clientCapabilities\":{}}}}";
+ }
+
+ private static final class Recording implements AgentMcpCallObserver {
+
+ private final ConcurrentLinkedQueue<Outcome> outcomes = new
ConcurrentLinkedQueue<>();
+
+ @Override
+ public void record(final Outcome outcome, final long millis) {
+ assertTrue(millis >= 0);
+ outcomes.add(outcome);
+ }
+
+ private void assertOnly(final Outcome outcome) {
+ assertEquals(List.of(outcome), List.copyOf(outcomes));
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
index 66c2f9f850..a6527250a6 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
@@ -19,6 +19,7 @@ package org.apache.shenyu.plugin.metrics;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
import org.apache.shenyu.common.utils.DateUtils;
import org.apache.shenyu.plugin.api.ShenyuPlugin;
import org.apache.shenyu.plugin.api.ShenyuPluginChain;
@@ -33,6 +34,7 @@ import reactor.core.publisher.Mono;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Optional;
+import java.util.Locale;
import java.util.function.Consumer;
/**
@@ -56,6 +58,11 @@ public class MetricsPlugin implements ShenyuPlugin {
}
private void setMetricsCallbacks(final ServerWebExchange exchange) {
+ exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
(AgentMcpCallObserver) (outcome, millis) -> {
+ String[] labels = {outcome.name().toLowerCase(Locale.ROOT)};
+ MetricsReporter.counterIncrement(LabelNames.AGENT_MCP_CALLS_TOTAL,
labels);
+ MetricsReporter.recordTime(LabelNames.AGENT_MCP_CALL_LATENCY,
labels, millis);
+ });
exchange.getAttributes().put(Constants.METRICS_SENTINEL,
(Consumer<HttpStatus>) status -> {
if (Objects.equals(HttpStatus.TOO_MANY_REQUESTS, status)) {
MetricsReporter.counterIncrement(LabelNames.SENTINEL_REQUEST_RESTRICT_TOTAL);
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
index 7db4b6d8e2..eccada895e 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
@@ -21,6 +21,12 @@ package org.apache.shenyu.plugin.metrics.constant;
* The Label names.
*/
public final class LabelNames {
+
+ /** Completed authenticated, parsed MCP logical calls, labelled only by
outcome. */
+ public static final String AGENT_MCP_CALLS_TOTAL =
"shenyu_agent_mcp_calls_total";
+
+ /** Time from validated MCP call parsing through terminal response
writing. */
+ public static final String AGENT_MCP_CALL_LATENCY =
"shenyu_agent_mcp_call_latency_millis";
/**
* The constant REQUEST_TOTAL.
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporter.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporter.java
index 38096fa29b..5e4b035340 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporter.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporter.java
@@ -43,6 +43,8 @@ public final class MetricsReporter {
MetricsReporter.registerCounter(LabelNames.REQUEST_TYPE_TOTAL, new
String[]{"type"}, "shenyu http request type total count");
MetricsReporter.registerCounter(LabelNames.REQUEST_THROW_TOTAL,
"shenyu request error total count");
MetricsReporter.registerHistogram(LabelNames.EXECUTE_LATENCY_NAME,
"the shenyu executor latency millis");
+ MetricsReporter.registerCounter(LabelNames.AGENT_MCP_CALLS_TOTAL, new
String[]{"outcome"}, "Terminal authenticated parsed MCP logical calls");
+ MetricsReporter.registerHistogram(LabelNames.AGENT_MCP_CALL_LATENCY,
new String[]{"outcome"}, "MCP logical call terminal latency millis");
MetricsReporter.registerCounter(LabelNames.SENTINEL_REQUEST_RESTRICT_TOTAL,
"shenyu sentinel request restrict total count");
MetricsReporter.registerCounter(LabelNames.SENTINEL_REQUEST_CIRCUITBREAKER_TOTAL,
"shenyu sentinel circuitbreaker request total count");
MetricsReporter.registerCounter(LabelNames.RESILIENCE4J_REQUEST_RESTRICT_TOTAL,
"shenyu resilience4j request restrict total count");
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/MetricsPluginTest.java
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/MetricsPluginTest.java
index 1b07081afb..5465b9a1a8 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/MetricsPluginTest.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/MetricsPluginTest.java
@@ -17,15 +17,18 @@
package org.apache.shenyu.plugin.metrics;
+import io.prometheus.client.CollectorRegistry;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.enums.PluginEnum;
import org.apache.shenyu.common.enums.RpcTypeEnum;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
import org.apache.shenyu.plugin.api.RemoteAddressResolver;
import org.apache.shenyu.plugin.api.ShenyuPluginChain;
import org.apache.shenyu.plugin.api.context.ShenyuContext;
import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
import org.apache.shenyu.plugin.metrics.constant.LabelNames;
import org.apache.shenyu.plugin.metrics.reporter.MetricsReporter;
+import org.apache.shenyu.plugin.metrics.prometheus.PrometheusMetricsRegister;
import org.apache.shenyu.plugin.metrics.spi.MetricsRegister;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -40,6 +43,7 @@ import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.net.InetSocketAddress;
+import java.util.Locale;
/**
* The Test Case For MetricsPlugin.
@@ -111,6 +115,51 @@ public class MetricsPluginTest {
}
}
+ @Test
+ public void testMcpCallbackUsesOnlyBoundedOutcomes() {
+ MetricsRegister metricsRegister = Mockito.mock(MetricsRegister.class);
+ MetricsReporter.register(metricsRegister);
+ try {
+
Mockito.when(chain.execute(ArgumentMatchers.any())).thenReturn(Mono.empty());
+ StepVerifier.create(metricsPlugin.execute(exchange,
chain)).verifyComplete();
+ AgentMcpCallObserver observer =
exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL);
+ Assertions.assertNotNull(observer);
+ for (AgentMcpCallObserver.Outcome outcome :
AgentMcpCallObserver.Outcome.values()) {
+ observer.record(outcome, 42);
+ String[] labels = {outcome.name().toLowerCase(Locale.ROOT)};
+
Mockito.verify(metricsRegister).counterIncrement(LabelNames.AGENT_MCP_CALLS_TOTAL,
labels, 1L);
+
Mockito.verify(metricsRegister).recordTime(LabelNames.AGENT_MCP_CALL_LATENCY,
labels, 42);
+ }
+ } finally {
+ MetricsReporter.clean();
+ }
+ }
+
+ @Test
+ public void testMcpTerminalMetricsAreExposedByExistingPrometheusRegistry()
{
+ MetricsReporter.clean();
+ CollectorRegistry.defaultRegistry.clear();
+ MetricsReporter.register(new PrometheusMetricsRegister());
+ try {
+ ShenyuContext shenyuContext =
exchange.getAttribute(Constants.CONTEXT);
+
Mockito.when(shenyuContext.getRpcType()).thenReturn(RpcTypeEnum.HTTP.getName());
+
Mockito.when(chain.execute(ArgumentMatchers.any())).thenReturn(Mono.empty());
+ StepVerifier.create(metricsPlugin.execute(exchange,
chain)).verifyComplete();
+ AgentMcpCallObserver observer =
exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL);
+ Assertions.assertNotNull(observer);
+ for (AgentMcpCallObserver.Outcome outcome :
AgentMcpCallObserver.Outcome.values()) {
+ observer.record(outcome, 42);
+ String[] labels = {outcome.name().toLowerCase(Locale.ROOT)};
+ Assertions.assertEquals(1.0,
CollectorRegistry.defaultRegistry.getSampleValue(LabelNames.AGENT_MCP_CALLS_TOTAL,
new String[]{"outcome"}, labels));
+ Assertions.assertEquals(1.0,
CollectorRegistry.defaultRegistry.getSampleValue(LabelNames.AGENT_MCP_CALL_LATENCY
+ "_count", new String[]{"outcome"}, labels));
+ Assertions.assertEquals(42.0,
CollectorRegistry.defaultRegistry.getSampleValue(LabelNames.AGENT_MCP_CALL_LATENCY
+ "_sum", new String[]{"outcome"}, labels));
+ }
+ } finally {
+ MetricsReporter.clean();
+ CollectorRegistry.defaultRegistry.clear();
+ }
+ }
+
private ServerWebExchange createExchange(final String path, final
ShenyuContext shenyuContext) {
ServerWebExchange result =
MockServerWebExchange.from(MockServerHttpRequest.get("http://localhost"; +
path).build());
result.getAttributes().put(Constants.CONTEXT, shenyuContext);
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
index 4bb6f35dec..023635a2d9 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
@@ -58,9 +58,9 @@ public final class MetricsReporterTest {
@Test
public void testRegister() throws Exception {
Map<String, Counter> map1 = getPrivateField(metricsRegister,
"COUNTER_MAP", Map.class);
- Assertions.assertEquals(map1.size(), 9);
+ Assertions.assertEquals(map1.size(), 10);
Map<String, Histogram> map2 = getPrivateField(metricsRegister,
"HISTOGRAM_MAP", Map.class);
- Assertions.assertEquals(map2.size(), 3);
+ Assertions.assertEquals(map2.size(), 4);
List<String> labels = new ArrayList<>();
labels.add("shenyu_request_total");
Collection<Metric> metrics = new ArrayList<>();
@@ -69,9 +69,9 @@ public final class MetricsReporterTest {
metrics.add(new Metric(MetricType.HISTOGRAM, "name3", DOCUMENT,
labels));
MetricsReporter.registerMetrics(metrics);
Map<String, Counter> map3 = getPrivateField(metricsRegister,
"COUNTER_MAP", Map.class);
- Assertions.assertEquals(map3.size(), 10);
+ Assertions.assertEquals(map3.size(), 11);
Map<String, Histogram> map4 = getPrivateField(metricsRegister,
"HISTOGRAM_MAP", Map.class);
- Assertions.assertEquals(map4.size(), 4);
+ Assertions.assertEquals(map4.size(), 5);
Map<String, Gauge> map5 = getPrivateField(metricsRegister,
"GAUGE_MAP", Map.class);
Assertions.assertEquals(map5.size(), 3);
}
