You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shenyu.apache.org by li...@apache.org on 2025/07/21 08:40:45 UTC
(shenyu) branch master updated: [feat] mcp streamable http (#6061)
This is an automated email from the ASF dual-hosted git repository.
liuhongyu 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 7de2558788 [feat] mcp streamable http (#6061)
7de2558788 is described below
commit 7de25587883c26ce9e55f2c37d551a2308679ba6
Author: 有若旭 <48...@users.noreply.github.com>
AuthorDate: Mon Jul 21 16:40:40 2025 +0800
[feat] mcp streamable http (#6061)
* [feat] mcp server plugin, init to support streamable http.
* [feat] mcp server plugin, let streamable http and sse use single McpAsyncServer.
* [feat] mcp server plugin, remove default tool for McpAsyncServer.
* [feat] mcp server plugin, session reconnection and stateless session support for Streamable http.
* [feat] mcp server plugin, debug session reconnection and stateless session support for Streamable http.
* [feat] mcp server plugin, Remove handle noSessionId exchange for ShenyuToolCallback .
* [feat] mcp server plugin, Add tools config examples .
* [feat] mcp server plugin, Modify Body writer for POST request .
* [feat] mcp server plugin, Modify rpc type when need do call.
* [feat] mcp server plugin, Modify parameter format & support grpc.
* [feat] mcp server plugin, fix streamable http after support custom message endpoint.
* [feat] mcp server plugin, Modify javadoc for more.
* [feat] mcp server plugin, fix check style.
* [feat] mcp server plugin, use compositeTransportMap remove sseTransportMap & streamableHttpTransportMap.
* [feat] mcp server plugin, reuse ObjectMapper & remove protocolMap.
* [feat] mcp server plugin, fix mistakes for Copilot check.
---
.../shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES.md | 492 ++++++++++
.../MCP_TOOL_EXAMPLES_EN.md | 499 +++++++++++
.../shenyu/plugin/mcp/server/McpServerPlugin.java | 659 +++++++++++---
.../mcp/server/callback/ShenyuToolCallback.java | 681 +++++++++++---
.../server/handler/McpServerPluginDataHandler.java | 48 +-
.../mcp/server/manager/ShenyuMcpServerManager.java | 584 +++++++++---
.../mcp/server/request/BodyWriterExchange.java | 90 +-
.../mcp/server/request/ParameterFormatter.java | 60 ++
.../mcp/server/request/RequestConfigHelper.java | 196 ++--
.../NonCommittingMcpResponseDecorator.java | 276 ++++++
.../server/transport/MessageHandlingResult.java | 99 ++
...henyuStreamableHttpServerTransportProvider.java | 996 +++++++++++++++++++++
.../transport/StreamableHttpProviderBuilder.java | 81 ++
.../plugin/mcp/server/utils/JsonSchemaUtil.java | 4 +-
14 files changed, 4272 insertions(+), 493 deletions(-)
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES.md b/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES.md
new file mode 100644
index 0000000000..78a029ab12
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES.md
@@ -0,0 +1,492 @@
+# MCP Tool 配置示例大全
+
+本文档基于 **shenyu-examples-http** 项目中的真实接口，提供了 Shenyu MCP Server Plugin 的各种工具配置示例，涵盖不同的 HTTP 请求方法、参数类型和配置方式。
+
+## 1. 简单 GET 请求示例
+
+### 1.1 无参数 GET 请求
+```json
+{
+ "name": "hello",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/hello\",\"method\":\"GET\",\"headers\":[],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "Shenyu 客户端问候接口"
+}
+```
+
+### 1.2 带参数的 GET 请求
+```json
+{
+ "name": "hi",
+ "parameters": [
+ {
+ "name": "name",
+ "type": "string",
+ "description": "用户名",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/hi\",\"method\":\"GET\",\"headers\":[],\"timeout\":30000,\"queryParams\":[{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"name\":\"query\"}}",
+ "description": "Shenyu 客户端带参数问候接口"
+}
+```
+
+## 2. 用户相关 GET 请求示例
+
+### 2.1 根据用户ID查找用户
+```json
+{
+ "name": "findByUserId",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByUserId\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"userId\",\"value\":\"${userId}\"}]},\"argsPosition\":{\"userId\":\"query\"}}",
+ "description": "根据用户ID查找用户信息"
+}
+```
+
+### 2.2 根据用户ID和姓名查找用户
+```json
+{
+ "name": "findByUserIdName",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "用户姓名",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByUserIdName\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"userId\",\"value\":\"${userId}\"},{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"userId\":\"query\",\"name\":\"query\"}}",
+ "description": "根据用户ID和姓名查找用户"
+}
+```
+
+### 2.3 分页查找用户
+```json
+{
+ "name": "findByPage",
+ "parameters": [
+ {
+ "name": "keyword",
+ "type": "string",
+ "description": "搜索关键词",
+ "required": false
+ },
+ {
+ "name": "page",
+ "type": "integer",
+ "description": "页码",
+ "required": false
+ },
+ {
+ "name": "pageSize",
+ "type": "integer",
+ "description": "每页大小",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByPage\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"keyword\",\"value\":\"${keyword}\"},{\"key\":\"page\",\"value\":\"${page}\"},{\"key\":\"pageSize\",\"value\":\"${pageSize}\"}]},\"argsPosition\":{\"keyword\":\"query\",\"page\":\"query\",\"pageSize\":\"query\"}}",
+ "description": "分页查找用户"
+}
+```
+
+### 2.4 通过路径参数获取用户信息
+```json
+{
+ "name": "getUserByPath",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "用户姓名",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/path/{{.id}}\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"id\":\"path\",\"name\":\"query\"}}",
+ "description": "通过路径参数获取用户信息"
+}
+```
+
+## 3. POST 请求示例
+
+### 3.1 用户支付接口
+```json
+{
+ "name": "payment",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ },
+ {
+ "name": "userName",
+ "type": "string",
+ "description": "用户姓名",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/payment\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"userId\":\"body\",\"userName\":\"body\"}}",
+ "description": "用户支付接口"
+}
+```
+
+### 3.2 WAF 通过测试
+```json
+{
+ "name": "wafPass",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/waf/pass\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "WAF 通过测试"
+}
+```
+
+### 3.3 Shenyu 客户端 POST 请求
+```json
+{
+ "name": "postHi",
+ "parameters": [
+ {
+ "name": "name",
+ "type": "string",
+ "description": "用户名",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/post/hi\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/x-www-form-urlencoded\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"name\":\"body\"}}",
+ "description": "Shenyu 客户端 POST 问候接口"
+}
+```
+
+## 4. PUT 请求示例
+
+### 4.1 更新用户信息
+```json
+{
+ "name": "updateUser",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ },
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "用户ID",
+ "required": true
+ },
+ {
+ "name": "userName",
+ "type": "string",
+ "description": "用户姓名",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/putPathBody/{{.id}}\",\"method\":\"PUT\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"id\":\"path\",\"userId\":\"body\",\"userName\":\"body\"}}",
+ "description": "更新用户信息"
+}
+```
+
+## 5. 订单相关示例
+
+### 5.1 保存订单
+```json
+{
+ "name": "saveOrder",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "订单ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "订单名称",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/save\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"id\":\"body\",\"name\":\"body\"}}",
+ "description": "保存订单"
+}
+```
+
+### 5.2 根据ID查找订单
+```json
+{
+ "name": "findOrderById",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "订单ID",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/findById\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"id\",\"value\":\"${id}\"}]},\"argsPosition\":{\"id\":\"query\"}}",
+ "description": "根据ID查找订单"
+}
+```
+
+### 5.3 RESTful 风格订单查询
+```json
+{
+ "name": "getOrderByPath",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "订单ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "订单名称",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/path/{{.id}}/{{.name}}\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{\"id\":\"path\",\"name\":\"path\"}}",
+ "description": "RESTful 风格订单查询"
+}
+```
+
+## 6. 请求处理示例
+
+### 6.1 请求头测试
+```json
+{
+ "name": "testRequestHeader",
+ "parameters": [
+ {
+ "name": "headerKey1",
+ "type": "string",
+ "description": "请求头参数",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/header\",\"method\":\"GET\",\"headers\":[{\"key\":\"header_key1\",\"value\":\"{{.headerKey1}}\"}],\"timeout\":30000},\"argsPosition\":{\"headerKey1\":\"query\"}}",
+ "description": "请求头测试"
+}
+```
+
+### 6.2 请求参数测试
+```json
+{
+ "name": "testRequestParameter",
+ "parameters": [
+ {
+ "name": "parameterKey1",
+ "type": "string",
+ "description": "请求参数",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/parameter\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/x-www-form-urlencoded\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"parameterKey1\":\"body\"}}",
+ "description": "请求参数测试"
+}
+```
+
+### 6.3 Cookie 测试
+```json
+{
+ "name": "testRequestCookie",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "用户ID（Cookie值）",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/cookie\",\"method\":\"GET\",\"headers\":[{\"key\":\"Cookie\",\"value\":\"userId={{.userId}}\"}],\"timeout\":30000},\"argsPosition\":{\"userId\":\"query\"}}",
+ "description": "Cookie 测试"
+}
+```
+
+## 7. 文件上传示例
+
+### 7.1 单文件上传
+```json
+{
+ "name": "uploadSingleFile",
+ "parameters": [
+ {
+ "name": "file",
+ "type": "string",
+ "description": "文件内容（Base64编码）",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/upload/webFluxSingle\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"multipart/form-data\"}],\"timeout\":60000,\"argsToJsonBody\":true},\"argsPosition\":{\"file\":\"body\"}}",
+ "description": "单文件上传"
+}
+```
+
+### 7.2 多文件上传
+```json
+{
+ "name": "uploadMultipleFiles",
+ "parameters": [
+ {
+ "name": "files",
+ "type": "array",
+ "description": "多个文件内容（Base64编码数组）",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/upload/webFluxFiles\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"multipart/form-data\"}],\"timeout\":60000,\"argsToJsonBody\":true},\"argsPosition\":{\"files\":\"body\"}}",
+ "description": "多文件上传"
+}
+```
+
+## 8. 高级功能示例
+
+### 8.1 OAuth2 认证测试
+```json
+{
+ "name": "testOAuth2",
+ "parameters": [
+ {
+ "name": "token",
+ "type": "string",
+ "description": "OAuth2 访问令牌",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/oauth2/test\",\"method\":\"GET\",\"headers\":[{\"key\":\"Authorization\",\"value\":\"Bearer {{.token}}\"}],\"timeout\":30000},\"argsPosition\":{\"token\":\"query\"}}",
+ "description": "OAuth2 认证测试"
+}
+```
+
+### 8.2 缓存测试
+```json
+{
+ "name": "testCache",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/cache\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "缓存测试"
+}
+```
+
+### 8.3 大对象响应测试
+```json
+{
+ "name": "testBigObject",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/bigObject\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":60000},\"argsPosition\":{}}",
+ "description": "大对象响应测试"
+}
+```
+
+## 参数位置映射说明 (argsPosition)
+
+### 支持的映射位置：
+
+1. **URL 路径参数**: `"path"` - 对应 URL 模板中的 `{{.paramName}}`
+2. **查询参数**: `"query"` - 对应 `queryParams` 数组中的参数
+3. **请求体**: `"body"` - 配合 `argsToJsonBody: true` 使用
+
+### 示例映射：
+```json
+{
+ "argsPosition": {
+ "id": "path", // 路径参数 /test/path/{{.id}}
+ "userId": "query", // 查询参数 queryParams: [{"key": "userId", "value": "${userId}"}]
+ "userName": "body" // JSON Body (需设置 argsToJsonBody: true)
+ }
+}
+```
+
+### 配置结构说明：
+
+#### requestTemplate 配置项：
+- `url`: URL 模板，路径参数支持 `{{.paramName}}` 占位符
+- `method`: HTTP 方法 (GET, POST, PUT, DELETE)
+- `headers`: 请求头数组，每个元素包含 `key` 和 `value`
+- `queryParams`: 查询参数数组，每个元素包含 `key` 和 `value`
+- `timeout`: 超时时间（毫秒）
+- `argsToJsonBody`: 是否将参数转换为 JSON 请求体
+
+#### 头部配置格式：
+```json
+{
+ "headers": [
+ {
+ "key": "Content-Type",
+ "value": "application/json"
+ },
+ {
+ "key": "Authorization",
+ "value": "Bearer {{.token}}"
+ }
+ ]
+}
+```
+
+#### 查询参数配置格式：
+```json
+{
+ "queryParams": [
+ {
+ "key": "userId",
+ "value": "${userId}"
+ },
+ {
+ "key": "page",
+ "value": "${page}"
+ }
+ ]
+}
+```
+
+## 使用说明
+
+1. **参数类型**: 支持 `string`, `integer`, `number`, `boolean`, `array`, `object`
+2. **必填参数**: 通过 `required` 字段控制
+3. **超时设置**: 通过 `timeout` 字段设置（毫秒）
+4. **请求头**: 使用 `key` 和 `value` 字段配置
+5. **查询参数**: 使用 `queryParams` 数组配置，值使用 `${paramName}` 格式
+6. **路径参数**: 在 URL 中使用 `{{.paramName}}` 格式
+7. **参数映射**: 通过 `argsPosition` 指定参数位置：`path`、`query`、`body`
+
+## 最佳实践
+
+1. **从简单开始**: 先配置基本的 GET/POST 示例，逐步增加复杂度
+2. **超时配置**:
+ - 简单查询: 30秒
+ - 复杂操作: 45-120秒
+ - 文件上传: 60秒以上
+3. **错误处理**: 在业务层实现重试机制和合适的错误响应
+4. **安全性**:
+ - 生产环境必须使用 HTTPS
+ - 访问安全接口时包含认证头
+ - 验证输入参数
+5. **性能优化**:
+ - 设置合适的超时时间
+ - 高频调用使用连接池
+ - 监控和记录API性能
+
+这些示例基于 **shenyu-examples-http** 项目中的真实接口，可以直接在 Shenyu 环境中使用和测试。
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES_EN.md b/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES_EN.md
new file mode 100644
index 0000000000..6dfb50f5ce
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/MCP_TOOL_EXAMPLES_EN.md
@@ -0,0 +1,499 @@
+# MCP Tool Configuration Examples
+
+This document provides comprehensive examples of tool configurations for the Shenyu MCP Server Plugin based on **real interfaces from the shenyu-examples-http project**, covering different HTTP request methods, parameter types, and configuration patterns.
+
+## 1. Simple GET Request Examples
+
+### 1.1 GET Request with No Parameters
+```json
+{
+ "name": "hello",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/hello\",\"method\":\"GET\",\"headers\":[],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "Shenyu client hello endpoint"
+}
+```
+
+### 1.2 GET Request with Parameters
+```json
+{
+ "name": "hi",
+ "parameters": [
+ {
+ "name": "name",
+ "type": "string",
+ "description": "User name",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/hi\",\"method\":\"GET\",\"headers\":[],\"timeout\":30000,\"queryParams\":[{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"name\":\"query\"}}",
+ "description": "Shenyu client hi endpoint with parameter"
+}
+```
+
+## 2. User-Related GET Request Examples
+
+### 2.1 Find User by ID
+```json
+{
+ "name": "findByUserId",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByUserId\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"userId\",\"value\":\"${userId}\"}]},\"argsPosition\":{\"userId\":\"query\"}}",
+ "description": "Find user information by user ID"
+}
+```
+
+### 2.2 Find User by ID and Name
+```json
+{
+ "name": "findByUserIdName",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "User name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByUserIdName\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"userId\",\"value\":\"${userId}\"},{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"userId\":\"query\",\"name\":\"query\"}}",
+ "description": "Find user by ID and name"
+}
+```
+
+### 2.3 Find Users with Pagination
+```json
+{
+ "name": "findByPage",
+ "parameters": [
+ {
+ "name": "keyword",
+ "type": "string",
+ "description": "Search keyword",
+ "required": false
+ },
+ {
+ "name": "page",
+ "type": "integer",
+ "description": "Page number",
+ "required": false
+ },
+ {
+ "name": "pageSize",
+ "type": "integer",
+ "description": "Page size",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/findByPage\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"keyword\",\"value\":\"${keyword}\"},{\"key\":\"page\",\"value\":\"${page}\"},{\"key\":\"pageSize\",\"value\":\"${pageSize}\"}]},\"argsPosition\":{\"keyword\":\"query\",\"page\":\"query\",\"pageSize\":\"query\"}}",
+ "description": "Find users with pagination"
+}
+```
+
+### 2.4 Get User by Path Parameter
+```json
+{
+ "name": "getUserByPath",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "User name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/path/{{.id}}\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"name\",\"value\":\"${name}\"}]},\"argsPosition\":{\"id\":\"path\",\"name\":\"query\"}}",
+ "description": "Get user information by path parameter"
+}
+```
+
+## 3. POST Request Examples
+
+### 3.1 User Payment Interface
+```json
+{
+ "name": "payment",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ },
+ {
+ "name": "userName",
+ "type": "string",
+ "description": "User name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/payment\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"userId\":\"body\",\"userName\":\"body\"}}",
+ "description": "User payment interface"
+}
+```
+
+### 3.2 WAF Pass Test
+```json
+{
+ "name": "wafPass",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/waf/pass\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "WAF pass test"
+}
+```
+
+### 3.3 Shenyu Client POST Request
+```json
+{
+ "name": "postHi",
+ "parameters": [
+ {
+ "name": "name",
+ "type": "string",
+ "description": "User name",
+ "required": false
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/shenyu/client/post/hi\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/x-www-form-urlencoded\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"name\":\"body\"}}",
+ "description": "Shenyu client POST hi endpoint"
+}
+```
+
+## 4. PUT Request Examples
+
+### 4.1 Update User Information
+```json
+{
+ "name": "updateUser",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ },
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "User ID",
+ "required": true
+ },
+ {
+ "name": "userName",
+ "type": "string",
+ "description": "User name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/putPathBody/{{.id}}\",\"method\":\"PUT\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"id\":\"path\",\"userId\":\"body\",\"userName\":\"body\"}}",
+ "description": "Update user information"
+}
+```
+
+## 5. Order-Related Examples
+
+### 5.1 Save Order
+```json
+{
+ "name": "saveOrder",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "Order ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "Order name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/save\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"id\":\"body\",\"name\":\"body\"}}",
+ "description": "Save order"
+}
+```
+
+### 5.2 Find Order by ID
+```json
+{
+ "name": "findOrderById",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "Order ID",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/findById\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000,\"queryParams\":[{\"key\":\"id\",\"value\":\"${id}\"}]},\"argsPosition\":{\"id\":\"query\"}}",
+ "description": "Find order by ID"
+}
+```
+
+### 5.3 RESTful Order Query
+```json
+{
+ "name": "getOrderByPath",
+ "parameters": [
+ {
+ "name": "id",
+ "type": "string",
+ "description": "Order ID",
+ "required": true
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "description": "Order name",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/path/{{.id}}/{{.name}}\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{\"id\":\"path\",\"name\":\"path\"}}",
+ "description": "RESTful style order query"
+}
+```
+
+## 6. Request Processing Examples
+
+### 6.1 Request Header Test
+```json
+{
+ "name": "testRequestHeader",
+ "parameters": [
+ {
+ "name": "headerKey1",
+ "type": "string",
+ "description": "Request header parameter",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/header\",\"method\":\"GET\",\"headers\":[{\"key\":\"header_key1\",\"value\":\"{{.headerKey1}}\"}],\"timeout\":30000},\"argsPosition\":{\"headerKey1\":\"query\"}}",
+ "description": "Request header test"
+}
+```
+
+### 6.2 Request Parameter Test
+```json
+{
+ "name": "testRequestParameter",
+ "parameters": [
+ {
+ "name": "parameterKey1",
+ "type": "string",
+ "description": "Request parameter",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/parameter\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/x-www-form-urlencoded\"}],\"timeout\":30000,\"argsToJsonBody\":true},\"argsPosition\":{\"parameterKey1\":\"body\"}}",
+ "description": "Request parameter test"
+}
+```
+
+### 6.3 Cookie Test
+```json
+{
+ "name": "testRequestCookie",
+ "parameters": [
+ {
+ "name": "userId",
+ "type": "string",
+ "description": "User ID (Cookie value)",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/request/cookie\",\"method\":\"GET\",\"headers\":[{\"key\":\"Cookie\",\"value\":\"userId={{.userId}}\"}],\"timeout\":30000},\"argsPosition\":{\"userId\":\"query\"}}",
+ "description": "Cookie test"
+}
+```
+
+## 7. File Upload Examples
+
+### 7.1 Single File Upload
+```json
+{
+ "name": "uploadSingleFile",
+ "parameters": [
+ {
+ "name": "file",
+ "type": "string",
+ "description": "File content (Base64 encoded)",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/upload/webFluxSingle\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"multipart/form-data\"}],\"timeout\":60000,\"argsToJsonBody\":true},\"argsPosition\":{\"file\":\"body\"}}",
+ "description": "Single file upload"
+}
+```
+
+### 7.2 Multiple Files Upload
+```json
+{
+ "name": "uploadMultipleFiles",
+ "parameters": [
+ {
+ "name": "files",
+ "type": "array",
+ "description": "Multiple file contents (Base64 encoded array)",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/upload/webFluxFiles\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"multipart/form-data\"}],\"timeout\":60000,\"argsToJsonBody\":true},\"argsPosition\":{\"files\":\"body\"}}",
+ "description": "Multiple files upload"
+}
+```
+
+## 8. Advanced Feature Examples
+
+### 8.1 OAuth2 Authentication Test
+```json
+{
+ "name": "testOAuth2",
+ "parameters": [
+ {
+ "name": "token",
+ "type": "string",
+ "description": "OAuth2 access token",
+ "required": true
+ }
+ ],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/order/oauth2/test\",\"method\":\"GET\",\"headers\":[{\"key\":\"Authorization\",\"value\":\"Bearer {{.token}}\"}],\"timeout\":30000},\"argsPosition\":{\"token\":\"query\"}}",
+ "description": "OAuth2 authentication test"
+}
+```
+
+### 8.2 Cache Test
+```json
+{
+ "name": "testCache",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/cache\",\"method\":\"GET\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":30000},\"argsPosition\":{}}",
+ "description": "Cache test"
+}
+```
+
+### 8.3 Big Object Response Test
+```json
+{
+ "name": "testBigObject",
+ "parameters": [],
+ "requestConfig": "{\"requestTemplate\":{\"url\":\"/test/bigObject\",\"method\":\"POST\",\"headers\":[{\"key\":\"Content-Type\",\"value\":\"application/json\"}],\"timeout\":60000},\"argsPosition\":{}}",
+ "description": "Big object response test"
+}
+```
+
+## Parameter Position Mapping (argsPosition)
+
+### Supported Mapping Locations:
+
+1. **URL Path Parameters**: `"path"` - Corresponds to `{{.paramName}}` in URL template
+2. **Query Parameters**: `"query"` - Corresponds to parameters in `queryParams` array
+3. **Request Body**: `"body"` - Used with `argsToJsonBody: true`
+
+### Mapping Examples:
+```json
+{
+ "argsPosition": {
+ "id": "path", // Path parameter /test/path/{{.id}}
+ "userId": "query", // Query parameter queryParams: [{"key": "userId", "value": "${userId}"}]
+ "userName": "body" // JSON Body (requires argsToJsonBody: true)
+ }
+}
+```
+
+### Configuration Structure:
+
+#### requestTemplate Configuration Items:
+- `url`: URL template with path parameter support using `{{.paramName}}` placeholders
+- `method`: HTTP method (GET, POST, PUT, DELETE)
+- `headers`: Headers array, each element contains `key` and `value`
+- `queryParams`: Query parameters array, each element contains `key` and `value`
+- `timeout`: Timeout in milliseconds
+- `argsToJsonBody`: Whether to convert parameters to JSON request body
+
+#### Header Configuration Format:
+```json
+{
+ "headers": [
+ {
+ "key": "Content-Type",
+ "value": "application/json"
+ },
+ {
+ "key": "Authorization",
+ "value": "Bearer {{.token}}"
+ }
+ ]
+}
+```
+
+#### Query Parameters Configuration Format:
+```json
+{
+ "queryParams": [
+ {
+ "key": "userId",
+ "value": "${userId}"
+ },
+ {
+ "key": "page",
+ "value": "${page}"
+ }
+ ]
+}
+```
+
+## Usage Guidelines
+
+1. **Parameter Types**: Supports `string`, `integer`, `number`, `boolean`, `array`, `object`
+2. **Required Parameters**: Controlled by the `required` field
+3. **Timeout Settings**: Configure timeout values in milliseconds via the `timeout` field
+4. **Request Headers**: Configure using `key` and `value` fields
+5. **Query Parameters**: Configure using `queryParams` array with `${paramName}` format values
+6. **Path Parameters**: Use `{{.paramName}}` format in URL
+7. **Parameter Mapping**: Use `argsPosition` to specify parameter location: `path`, `query`, `body`
+
+## Best Practices
+
+### 1. **Start Simple**
+Begin with basic GET/POST examples and gradually add complexity.
+
+### 2. **Timeout Configuration**
+- Simple queries: 30 seconds
+- Complex operations: 45-120 seconds
+- File uploads: 60+ seconds
+
+### 3. **Error Handling**
+Implement retry mechanisms and proper error responses at the business layer.
+
+### 4. **Security**
+- Always use HTTPS in production
+- Include authentication headers when accessing secure endpoints
+- Validate input parameters
+
+### 5. **Performance**
+- Set appropriate timeouts
+- Use connection pooling for high-frequency calls
+- Monitor and log API performance
+
+These examples are based on **real interfaces from the shenyu-examples-http project** and can be used and tested directly in a Shenyu environment.
+
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/McpServerPlugin.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/McpServerPlugin.java
index 0ac672d156..fce5ac904c 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/McpServerPlugin.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/McpServerPlugin.java
@@ -31,7 +31,9 @@ import org.apache.shenyu.plugin.mcp.server.holder.ShenyuMcpExchangeHolder;
import org.apache.shenyu.plugin.mcp.server.manager.ShenyuMcpServerManager;
import org.apache.shenyu.plugin.mcp.server.model.ShenyuMcpServer;
import org.apache.shenyu.plugin.mcp.server.transport.ShenyuSseServerTransportProvider;
+import org.apache.shenyu.plugin.mcp.server.transport.ShenyuStreamableHttpServerTransportProvider;
import org.apache.shenyu.plugin.mcp.server.transport.SseEventFormatter;
+import org.apache.shenyu.plugin.mcp.server.transport.MessageHandlingResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
@@ -39,187 +41,604 @@ import org.springframework.http.codec.HttpMessageReader;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
+import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
+import java.nio.charset.StandardCharsets;
/**
- * MCP Server Plugin.
- * Handles MCP (Message Control Protocol) server-side functionality.
+ * MCP (Model Context Protocol) Server Plugin for Shenyu Gateway.
+ *
+ * <p>Provides MCP server functionality supporting both SSE and Streamable HTTP transport protocols.
+ * Enables AI models to interact with Shenyu Gateway services through standardized MCP tool definitions.</p>
+ *
+ * @see org.apache.shenyu.plugin.base.AbstractShenyuPlugin
+ * @see org.apache.shenyu.plugin.mcp.server.manager.ShenyuMcpServerManager
+ * @since 2.7.0.2
*/
public class McpServerPlugin extends AbstractShenyuPlugin {
-
+
private static final Logger LOG = LoggerFactory.getLogger(McpServerPlugin.class);
-
+
+ /**
+ * Standard message endpoint path.
+ */
private static final String MESSAGE_ENDPOINT = "/message";
-
+
+ /**
+ * Streamable HTTP protocol path identifier.
+ */
+ private static final String STREAMABLE_HTTP_PATH = "/streamablehttp";
+
+ /**
+ * SSE protocol path identifier.
+ */
+ private static final String SSE_PATH = "/sse";
+
+ /**
+ * MCP tool call prevention attribute.
+ */
+ private static final String MCP_TOOL_CALL_ATTR = "MCP_TOOL_CALL";
+
+ /**
+ * MCP session ID attribute key.
+ */
+ private static final String MCP_SESSION_ID_ATTR = "MCP_SESSION_ID";
+
+ /**
+ * Session ID query parameter name.
+ */
+ private static final String SESSION_ID_PARAM = "sessionId";
+
+ /**
+ * Session ID header names (in order of preference).
+ */
+ private static final String[] SESSION_ID_HEADERS = {
+ "X-Session-Id", "Mcp-Session-Id",
+ };
+
+ /**
+ * Authorization header name.
+ */
+ private static final String AUTHORIZATION_HEADER = "Authorization";
+
+ /**
+ * Bearer token prefix.
+ */
+ private static final String BEARER_PREFIX = "Bearer ";
+
private final ShenyuMcpServerManager shenyuMcpServerManager;
-
+
private final List<HttpMessageReader<?>> messageReaders;
-
- public McpServerPlugin(final ShenyuMcpServerManager shenyuMcpServerManager, final List<HttpMessageReader<?>> messageReaders) {
+
+ /**
+ * Constructs a new MCP server plugin.
+ *
+ * @param shenyuMcpServerManager the MCP server manager for handling transport providers
+ * @param messageReaders the HTTP message readers for request processing
+ */
+ public McpServerPlugin(final ShenyuMcpServerManager shenyuMcpServerManager,
+ final List<HttpMessageReader<?>> messageReaders) {
this.shenyuMcpServerManager = shenyuMcpServerManager;
this.messageReaders = messageReaders;
}
-
+
@Override
protected String getRawPath(final ServerWebExchange exchange) {
return RequestUrlUtils.getRewrittenRawPath(exchange);
}
-
+
@Override
- protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {
- ShenyuContext shenyuContext = exchange.getAttribute(Constants.CONTEXT);
+ protected Mono<Void> doExecute(final ServerWebExchange exchange,
+ final ShenyuPluginChain chain,
+ final SelectorData selector,
+ final RuleData rule) {
+
+ final ShenyuContext shenyuContext = exchange.getAttribute(Constants.CONTEXT);
Objects.requireNonNull(shenyuContext, "ShenyuContext must not be null");
-
- ShenyuMcpServer server = McpServerPluginDataHandler.CACHED_SERVER.get().obtainHandle(selector.getId());
-
- if (Objects.isNull(server)) {
- return chain.execute(exchange);
- }
-
- String uri = exchange.getRequest().getURI().getRawPath();
- LOG.info("Processing request with URI: {}", uri);
-
+
+ final String uri = exchange.getRequest().getURI().getRawPath();
+ LOG.debug("Processing MCP request with URI: {}", uri);
+
if (!shenyuMcpServerManager.canRoute(uri)) {
- // Continue the chain - the WebFlux infrastructure will route to the appropriate
- // handler
+ LOG.debug("URI not handled by MCP server, continuing chain: {}", uri);
return chain.execute(exchange);
}
-
- LOG.info("Handling MCP request for URI: {}", uri);
-
- // Extract sessionId from request if available
- ServerRequest request = ServerRequest.create(exchange, messageReaders);
-
- // Handle MCP SSE/message requests by delegating to the transport provider
- // directly
- String messageEndpoint = server.getMessageEndpoint();
-
- ShenyuSseServerTransportProvider transportProvider
- = shenyuMcpServerManager.getOrCreateMcpServerTransport(uri, messageEndpoint);
-
- // Mark exchange to prevent further plugin processing
- shenyuContext.setRpcType(RpcTypeEnum.AI.getName());
- exchange.getAttributes().put(Constants.CONTEXT, shenyuContext);
-
- if (uri.endsWith(messageEndpoint)) {
- String sessionId = extractSessionId(exchange);
- if (Objects.nonNull(sessionId)) {
- exchange.getAttributes().put("MCP_SESSION_ID", sessionId);
- exchange.getAttributes().put(Constants.CHAIN, chain);
- // Store sessionId in exchange attributes for later use
- ShenyuMcpExchangeHolder.put(sessionId, exchange);
- LOG.debug("Extracted sessionId: {}", sessionId);
- }
- // Handle JSON-RPC message endpoint
- LOG.debug("Handling MCP message endpoint for URI: {} with sessionId: {}", uri, sessionId);
- return handleMessageEndpoint(exchange, transportProvider, request)
- .doFinally(signalType -> ShenyuMcpExchangeHolder.remove(sessionId));
- } else {
- // Handle SSE connection endpoint
- LOG.debug("Handling MCP SSE connection for URI: {} ", uri);
-
- // Create a custom SSE handler that bypasses the ServerResponse complexity
- return handleSseEndpoint(exchange, transportProvider, request);
- }
-
+
+ LOG.debug("Handling MCP request for URI: {}", uri);
+
+ // Create server request for processing
+ final ServerRequest request = ServerRequest.create(exchange, messageReaders);
+
+ // Route based on protocol type
+ return routeByProtocol(exchange, chain, request, selector, uri);
}
-
+
@Override
public String named() {
return PluginEnum.MCP_SERVER.getName();
}
-
+
@Override
public boolean skip(final ServerWebExchange exchange) {
+ // Skip MCP plugin for MCP tool calls to prevent infinite loops
+ final Boolean isMcpToolCall = exchange.getAttribute(MCP_TOOL_CALL_ATTR);
+ if (Boolean.TRUE.equals(isMcpToolCall)) {
+ LOG.debug("Skipping MCP plugin for tool call to prevent infinite loop");
+ return true;
+ }
+
return skipExcept(exchange, RpcTypeEnum.HTTP);
}
-
+
@Override
public int getOrder() {
return PluginEnum.MCP_SERVER.getCode();
}
-
+
+
+ /**
+ * Routes the request based on detected protocol type.
+ *
+ * @param exchange the server web exchange
+ * @param chain the plugin chain
+ * @param request the server request
+ * @param uri the request URI
+ * @return a Mono representing the processing result
+ */
+ private Mono<Void> routeByProtocol(final ServerWebExchange exchange,
+ final ShenyuPluginChain chain,
+ final ServerRequest request,
+ final SelectorData selector,
+ final String uri) {
+
+ if (isStreamableHttpProtocol(uri)) {
+ return handleStreamableHttpRequest(exchange, chain, request, uri);
+ } else if (isSseProtocol(uri)) {
+ return handleSseRequest(exchange, chain, request, selector, uri);
+ } else {
+ // Default to SSE for backward compatibility
+ LOG.debug("Using default SSE protocol for URI: {}", uri);
+ return handleSseRequest(exchange, chain, request, selector, uri);
+ }
+ }
+
/**
- * Extract sessionId from request parameters or headers.
+ * Extracts session ID from request parameters or headers.
+ * Searches in the following order:
+ * <ol>
+ * <li>Query parameter "sessionId"</li>
+ * <li>Header "X-Session-Id"</li>
+ * <li>Header "Mcp-Session-Id"</li>
+ * <li>Authorization header (Bearer token)</li>
+ * </ol>
*
- * @param exchange The ServerWebExchange containing the request
- * @return The sessionId if found, null otherwise
+ * @param exchange the ServerWebExchange containing the request
+ * @return the sessionId if found, null otherwise
*/
private String extractSessionId(final ServerWebExchange exchange) {
- // Try query parameters
- String sessionId = exchange.getRequest().getQueryParams().getFirst("sessionId");
+ // Try query parameters first
+ String sessionId = exchange.getRequest().getQueryParams().getFirst(SESSION_ID_PARAM);
if (Objects.nonNull(sessionId)) {
LOG.debug("Found sessionId in query parameters: {}", sessionId);
return sessionId;
}
-
- // Try X-Session-Id header
- sessionId = exchange.getRequest().getHeaders().getFirst("X-Session-Id");
- if (Objects.nonNull(sessionId)) {
- LOG.debug("Found sessionId in X-Session-Id header: {}", sessionId);
- return sessionId;
+
+ // Try session ID headers
+ for (String headerName : SESSION_ID_HEADERS) {
+ sessionId = exchange.getRequest().getHeaders().getFirst(headerName);
+ if (Objects.nonNull(sessionId)) {
+ LOG.debug("Found sessionId in {} header: {}", headerName, sessionId);
+ return sessionId;
+ }
}
-
- // Try Authorization header
- String authHeader = exchange.getRequest().getHeaders().getFirst("Authorization");
- if (Objects.nonNull(authHeader) && authHeader.startsWith("Bearer ")) {
- sessionId = authHeader.substring(7);
+
+ // Try Authorization header as fallback
+ final String authHeader = exchange.getRequest().getHeaders().getFirst(AUTHORIZATION_HEADER);
+ if (Objects.nonNull(authHeader) && authHeader.startsWith(BEARER_PREFIX)) {
+ sessionId = authHeader.substring(BEARER_PREFIX.length());
LOG.debug("Found sessionId in Authorization header: {}", sessionId);
return sessionId;
}
-
+
LOG.debug("No sessionId found in request for path: {}", exchange.getRequest().getPath().value());
return null;
}
-
+
+ /**
+ * Checks if the URI indicates Streamable HTTP protocol.
+ *
+ * @param uri the URI to check
+ * @return true if Streamable HTTP protocol should be used
+ */
+ private boolean isStreamableHttpProtocol(final String uri) {
+ return uri.contains(STREAMABLE_HTTP_PATH) || uri.endsWith(STREAMABLE_HTTP_PATH);
+ }
+
+ /**
+ * Checks if the URI indicates SSE protocol.
+ *
+ * @param uri the URI to check
+ * @return true if SSE protocol should be used
+ */
+ private boolean isSseProtocol(final String uri) {
+ return uri.contains(SSE_PATH) || uri.endsWith(SSE_PATH) || uri.endsWith(MESSAGE_ENDPOINT);
+ }
+
/**
- * Create a direct SSE response by manually setting up the SSE stream.
- * This bypasses the ServerResponse complexity and writes directly to the
- * exchange.
+ * Handles Streamable HTTP MCP requests with unified endpoint processing.
+ *
+ * @param exchange the server web exchange
+ * @param chain the plugin chain
+ * @param request the server request
+ * @param uri the request URI
+ * @return a Mono representing the processing result
*/
- private Mono<Void> handleSseEndpoint(final ServerWebExchange exchange, final ShenyuSseServerTransportProvider transportProvider, final ServerRequest request) {
- LOG.debug("Handling SSE endpoint for request: {}", request.path());
-
- // Set SSE headers
+ private Mono<Void> handleStreamableHttpRequest(final ServerWebExchange exchange,
+ final ShenyuPluginChain chain,
+ final ServerRequest request,
+ final String uri) {
+
+ LOG.debug("Handling Streamable HTTP MCP request for URI: {}", uri);
+
+ final ShenyuStreamableHttpServerTransportProvider transportProvider =
+ shenyuMcpServerManager.getOrCreateStreamableHttpTransport(uri);
+
+ setupSessionContext(exchange, chain);
+
+ return processStreamableHttpEndpoint(exchange, transportProvider, request);
+ }
+
+ /**
+ * Handles SSE MCP requests with endpoint-specific routing.
+ *
+ * @param exchange the server web exchange
+ * @param chain the plugin chain
+ * @param request the server request
+ * @param uri the request URI
+ * @return a Mono representing the processing result
+ */
+ private Mono<Void> handleSseRequest(final ServerWebExchange exchange,
+ final ShenyuPluginChain chain,
+ final ServerRequest request,
+ final SelectorData selector,
+ final String uri) {
+
+ LOG.debug("Handling SSE MCP request for URI: {}", uri);
+ ShenyuMcpServer server = McpServerPluginDataHandler.CACHED_SERVER.get().obtainHandle(selector.getId());
+
+ if (Objects.isNull(server)) {
+ return chain.execute(exchange);
+ }
+ // Handle MCP SSE/message requests by delegating to the transport provider
+ // directly
+ String messageEndpoint = server.getMessageEndpoint();
+
+ ShenyuSseServerTransportProvider transportProvider
+ = shenyuMcpServerManager.getOrCreateMcpServerTransport(uri, messageEndpoint);
+
+ if (uri.endsWith(messageEndpoint)) {
+ setupSessionContext(exchange, chain);
+ return handleMessageEndpoint(exchange, transportProvider, request);
+ } else {
+ return handleSseEndpoint(exchange, transportProvider, request);
+ }
+ }
+
+ /**
+ * Sets up session context and exchange correlation.
+ *
+ * @param exchange the server web exchange
+ * @param chain the plugin chain
+ */
+ private void setupSessionContext(final ServerWebExchange exchange, final ShenyuPluginChain chain) {
+ final String sessionId = extractSessionId(exchange);
+ if (Objects.nonNull(sessionId)) {
+ exchange.getAttributes().put(MCP_SESSION_ID_ATTR, sessionId);
+ exchange.getAttributes().put(Constants.CHAIN, chain);
+ ShenyuMcpExchangeHolder.put(sessionId, exchange);
+ LOG.debug("Set up session context for sessionId: {}", sessionId);
+ }
+ }
+
+ /**
+ * Processes Streamable HTTP unified endpoint requests.
+ * Handles both GET (stream establishment) and POST (message processing) requests
+ * according to the Streamable HTTP protocol specification.
+ *
+ * @param exchange the server web exchange
+ * @param transportProvider the Streamable HTTP transport provider
+ * @param request the server request
+ * @return a Mono representing the processing result
+ */
+ private Mono<Void> processStreamableHttpEndpoint(final ServerWebExchange exchange,
+ final ShenyuStreamableHttpServerTransportProvider transportProvider,
+ final ServerRequest request) {
+
+ LOG.debug("Processing Streamable HTTP endpoint for request: {}", request.path());
+
+ final String method = exchange.getRequest().getMethod().name();
+
+ if ("GET".equalsIgnoreCase(method)) {
+ return handleStreamableHttpGetRequest(exchange);
+ } else if ("POST".equalsIgnoreCase(method)) {
+ return handleStreamableHttpPostRequest(exchange, transportProvider, request);
+ } else {
+ return handleUnsupportedMethod(exchange);
+ }
+ }
+
+ /**
+ * Handles Streamable HTTP GET requests (not supported by protocol).
+ *
+ * @param exchange the server web exchange
+ * @return a Mono representing the error response
+ */
+ private Mono<Void> handleStreamableHttpGetRequest(final ServerWebExchange exchange) {
+ LOG.debug("Rejecting Streamable HTTP GET request (protocol does not support GET)");
+
+ setErrorResponse(exchange, HttpStatus.METHOD_NOT_ALLOWED,
+ "POST, OPTIONS",
+ createJsonError(-32601, "Streamable HTTP does not support GET requests. Please use POST requests for all MCP operations."));
+
+ return writeJsonResponse(exchange);
+ }
+
+ /**
+ * Handles Streamable HTTP POST requests for message processing.
+ *
+ * @param exchange the server web exchange
+ * @param transportProvider the transport provider
+ * @param request the server request
+ * @return a Mono representing the processing result
+ */
+ private Mono<Void> handleStreamableHttpPostRequest(final ServerWebExchange exchange,
+ final ShenyuStreamableHttpServerTransportProvider transportProvider,
+ final ServerRequest request) {
+
+ LOG.debug("Processing Streamable HTTP POST request for message handling");
+
+ return transportProvider.handleMessageEndpoint(exchange, request)
+ .flatMap(result -> processStreamableHttpResult(exchange, result))
+ .doOnSuccess(aVoid -> LOG.debug("Streamable HTTP message processing completed"))
+ .doOnError(error -> LOG.error("Error in Streamable HTTP message processing: {}", error.getMessage(), error));
+ }
+
+ /**
+ * Processes the result of Streamable HTTP message handling.
+ *
+ * @param exchange the server web exchange
+ * @param result the message handling result
+ * @return a Mono representing the response writing
+ */
+ private Mono<Void> processStreamableHttpResult(final ServerWebExchange exchange,
+ final MessageHandlingResult result) {
+
+ LOG.debug("Processing Streamable HTTP result - Status: {}, SessionId: {}",
+ result.getStatusCode(), result.getSessionId());
+
+ // Configure response
+ configureStreamableHttpResponse(exchange, result);
+
+ // Write response body
+ final String responseBodyJson = result.getResponseBodyAsJson();
+ final byte[] responseBytes = responseBodyJson.getBytes(StandardCharsets.UTF_8);
+
+ LOG.debug("Writing response body with {} bytes", responseBytes.length);
+
+ return exchange.getResponse()
+ .writeWith(Mono.just(exchange.getResponse().bufferFactory().wrap(responseBytes)))
+ .doOnSuccess(aVoid -> LOG.debug("Response transmission completed successfully"))
+ .doOnError(error -> LOG.error("Error writing response: {}", error.getMessage(), error));
+ }
+
+ /**
+ * Configures response headers for Streamable HTTP.
+ *
+ * @param exchange the server web exchange
+ * @param result the message handling result
+ */
+ private void configureStreamableHttpResponse(final ServerWebExchange exchange,
+ final MessageHandlingResult result) {
+
+ // Set response status
+ exchange.getResponse().setStatusCode(HttpStatus.valueOf(result.getStatusCode()));
+
+ // Add session ID header if available
+ if (Objects.nonNull(result.getSessionId())) {
+ exchange.getResponse().getHeaders().set("Mcp-Session-Id", result.getSessionId());
+ } else {
+ // Fallback to extracting from exchange
+ final String sessionId = extractSessionId(exchange);
+ if (Objects.nonNull(sessionId)) {
+ exchange.getResponse().getHeaders().set("Mcp-Session-Id", sessionId);
+ }
+ }
+
+ // Set standard headers
+ setCorsHeaders(exchange);
+ exchange.getResponse().getHeaders().set("Content-Type", "application/json");
+
+ // Clean up potentially conflicting headers
+ exchange.getResponse().getHeaders().remove("Transfer-Encoding");
+ exchange.getResponse().getHeaders().remove("Content-Length");
+
+ LOG.debug("Configured Streamable HTTP response headers");
+ }
+
+ /**
+ * Handles unsupported HTTP methods.
+ *
+ * @param exchange the server web exchange
+ * @return a Mono representing the error response
+ */
+ private Mono<Void> handleUnsupportedMethod(final ServerWebExchange exchange) {
+ LOG.debug("Unsupported HTTP method: {}", exchange.getRequest().getMethod());
+
+ setErrorResponse(exchange, HttpStatus.BAD_REQUEST,
+ null,
+ createJsonError(-32600, "Unsupported HTTP method"));
+
+ return writeJsonResponse(exchange);
+ }
+
+ /**
+ * Creates SSE response by setting up the SSE stream.
+ *
+ * @param exchange the server web exchange
+ * @param transportProvider the SSE transport provider
+ * @param request the server request
+ * @return a Mono representing the SSE stream
+ */
+ private Mono<Void> handleSseEndpoint(final ServerWebExchange exchange,
+ final ShenyuSseServerTransportProvider transportProvider,
+ final ServerRequest request) {
+
+ LOG.debug("Setting up SSE endpoint for request: {}", request.path());
+
+ // Configure SSE headers
configureSseHeaders(exchange);
-
- // Create SSE flux and write each event immediately to the exchange response
+
+ // Create and write SSE stream
return exchange.getResponse()
.writeWith(transportProvider
.createSseFlux(request)
- .doOnNext(event -> LOG.debug("SSE Event - Type: {}, Data: {}", event.event(), event.data()))
+ .doOnNext(event -> {
+ String eventType = event.event();
+ LOG.debug("SSE Event - Type: {}", Objects.isNull(eventType) ? "data" : eventType);
+ })
.map(event -> SseEventFormatter.formatEvent(event, exchange))
.doOnSubscribe(subscription -> LOG.debug("SSE stream subscribed"))
.doOnComplete(() -> LOG.debug("SSE stream completed"))
- .doOnError(error -> LOG.error("SSE stream error: {}", error.getMessage())));
+ .doOnError(error -> LOG.error("SSE stream error: {}", error.getMessage(), error)));
}
-
+
+ /**
+ * Configures SSE response headers.
+ *
+ * @param exchange the server web exchange
+ */
private void configureSseHeaders(final ServerWebExchange exchange) {
- exchange.getResponse().getHeaders().add("Content-Type", "text/event-stream");
- exchange.getResponse().getHeaders().add("Cache-Control", "no-cache");
- exchange.getResponse().getHeaders().add("Connection", "keep-alive");
- exchange.getResponse().getHeaders().add("Access-Control-Allow-Origin", "*");
- LOG.debug("Configured SSE headers on exchange response");
- }
-
- private Mono<Void> handleMessageEndpoint(final ServerWebExchange exchange, final ShenyuSseServerTransportProvider transportProvider, final ServerRequest request) {
-
- return transportProvider.handleMessageEndpoint(request).flatMap(result -> {
- LOG.debug("Message handling result - Status: {}, Body: {}", result.getStatusCode(), result.getResponseBody());
-
- // Set response status
- exchange.getResponse().setStatusCode(HttpStatus.valueOf(result.getStatusCode()));
- // Set appropriate headers
- exchange.getResponse().getHeaders().add("Content-Type", "application/json");
- exchange.getResponse().getHeaders().add("Access-Control-Allow-Origin", "*");
-
- // Write response body
- String responseBody = String.format("{\"message\":\"%s\"}", result.getResponseBody());
- LOG.debug("Sending response body: {}", responseBody);
-
- return exchange.getResponse().writeWith(Mono.just(exchange.getResponse().bufferFactory().wrap(responseBody.getBytes())));
- }).doOnSuccess(aVoid -> LOG.debug("Message response writing completed")).doOnError(error -> LOG.error("Error writing message response: {}", error.getMessage()));
+ exchange.getResponse().getHeaders().set("Content-Type", "text/event-stream");
+ exchange.getResponse().getHeaders().set("Cache-Control", "no-cache");
+ exchange.getResponse().getHeaders().set("Connection", "keep-alive");
+ setCorsHeaders(exchange);
+ LOG.debug("Configured SSE headers");
+ }
+
+ /**
+ * Handles message endpoint requests for SSE protocol.
+ *
+ * @param exchange the server web exchange
+ * @param transportProvider the SSE transport provider
+ * @param request the server request
+ * @return a Mono representing the processing result
+ */
+ private Mono<Void> handleMessageEndpoint(final ServerWebExchange exchange,
+ final ShenyuSseServerTransportProvider transportProvider,
+ final ServerRequest request) {
+
+ LOG.debug("Processing message endpoint request");
+
+ return transportProvider.handleMessageEndpoint(request)
+ .flatMap(result -> {
+ LOG.debug("Message handling result - Status: {}, Body length: {} chars",
+ result.getStatusCode(),
+ Objects.nonNull(result.getResponseBody()) ? result.getResponseBody().length() : 0);
+
+ // Configure response
+ exchange.getResponse().setStatusCode(HttpStatus.valueOf(result.getStatusCode()));
+ exchange.getResponse().getHeaders().add("Content-Type", "application/json");
+ setCorsHeaders(exchange);
+
+ // Create response body
+ final String responseBody = String.format("{\"message\":\"%s\"}", result.getResponseBody());
+ LOG.debug("Sending message response with length: {} chars", responseBody.length());
+
+ return exchange.getResponse()
+ .writeWith(Mono.just(exchange.getResponse().bufferFactory().wrap(responseBody.getBytes())));
+ })
+ .doOnSuccess(aVoid -> LOG.debug("Message response completed"))
+ .doOnError(error -> LOG.error("Error in message response: {}", error.getMessage(), error));
}
+
+ /**
+ * Sets CORS headers for cross-origin requests.
+ *
+ * @param exchange the server web exchange
+ */
+ private void setCorsHeaders(final ServerWebExchange exchange) {
+ exchange.getResponse().getHeaders().set("Access-Control-Allow-Origin", "*");
+ exchange.getResponse().getHeaders().set("Access-Control-Allow-Headers",
+ "Content-Type, Mcp-Session-Id, Authorization, Last-Event-ID");
+ exchange.getResponse().getHeaders().set("Access-Control-Allow-Methods",
+ "GET, POST, OPTIONS");
+ }
+
+ /**
+ * Sets up error response configuration.
+ *
+ * @param exchange the server web exchange
+ * @param status the HTTP status
+ * @param allowHeader the Allow header value (nullable)
+ * @param errorBody the error response body
+ */
+ private void setErrorResponse(final ServerWebExchange exchange,
+ final HttpStatus status,
+ final String allowHeader,
+ final Map<String, Object> errorBody) {
+
+ exchange.getResponse().setStatusCode(status);
+ exchange.getResponse().getHeaders().add("Content-Type", "application/json");
+ setCorsHeaders(exchange);
+
+ if (Objects.nonNull(allowHeader)) {
+ exchange.getResponse().getHeaders().add("Allow", allowHeader);
+ }
+
+ exchange.getAttributes().put("errorBody", errorBody);
+ }
+
+ /**
+ * Writes JSON response to the exchange.
+ *
+ * @param exchange the server web exchange
+ * @return a Mono representing the write operation
+ */
+ private Mono<Void> writeJsonResponse(final ServerWebExchange exchange) {
+ @SuppressWarnings("unchecked") final Map<String, Object> errorBody = (Map<String, Object>) exchange.getAttributes().get("errorBody");
+
+ if (Objects.isNull(errorBody)) {
+ return Mono.empty();
+ }
+
+ try {
+ final String errorResponse = new ObjectMapper()
+ .writeValueAsString(errorBody);
+ return exchange.getResponse()
+ .writeWith(Mono.just(exchange.getResponse().bufferFactory().wrap(errorResponse.getBytes())));
+ } catch (Exception e) {
+ LOG.error("Error writing JSON response: {}", e.getMessage(), e);
+ return Mono.empty();
+ }
+ }
+
+ /**
+ * Creates a standard JSON-RPC error object.
+ *
+ * @param code the error code
+ * @param message the error message
+ * @return the error object as a Map
+ */
+ private Map<String, Object> createJsonError(final int code, final String message) {
+ return Map.of(
+ "error", Map.of(
+ "code", code,
+ "message", message
+ )
+ );
+ }
+
}
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/callback/ShenyuToolCallback.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/callback/ShenyuToolCallback.java
index c5099af0d1..c1d3583047 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/callback/ShenyuToolCallback.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/callback/ShenyuToolCallback.java
@@ -18,11 +18,11 @@
package org.apache.shenyu.plugin.mcp.server.callback;
import com.google.common.collect.Maps;
+import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import io.modelcontextprotocol.server.McpSyncServerExchange;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.MetaData;
-import org.apache.shenyu.common.enums.RpcTypeEnum;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.plugin.api.ShenyuPluginChain;
import org.apache.shenyu.plugin.api.context.ShenyuContext;
@@ -30,9 +30,11 @@ import org.apache.shenyu.plugin.base.cache.MetaDataCache;
import org.apache.shenyu.plugin.mcp.server.definition.ShenyuToolDefinition;
import org.apache.shenyu.plugin.mcp.server.holder.ShenyuMcpExchangeHolder;
import org.apache.shenyu.plugin.mcp.server.request.BodyWriterExchange;
+import org.apache.shenyu.plugin.mcp.server.request.ParameterFormatter;
import org.apache.shenyu.plugin.mcp.server.request.RequestConfig;
import org.apache.shenyu.plugin.mcp.server.request.RequestConfigHelper;
import org.apache.shenyu.plugin.mcp.server.response.ShenyuMcpResponseDecorator;
+import org.apache.shenyu.plugin.mcp.server.response.NonCommittingMcpResponseDecorator;
import org.apache.shenyu.plugin.mcp.server.session.McpSessionHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,199 +55,628 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+/**
+ * Tool Callback Implementation for Shenyu Gateway MCP Integration.
+ *
+ * <p>Handles tool invocations within the MCP framework, enabling AI models to interact
+ * with Shenyu Gateway services through defined tools. Requires pre-established session
+ * context and ServerWebExchange correlation.</p>
+ *
+ * @see ToolCallback
+ * @see ToolDefinition
+ * @since 2.7.0.2
+ */
public class ShenyuToolCallback implements ToolCallback {
-
+
private static final Logger LOG = LoggerFactory.getLogger(ShenyuToolCallback.class);
-
+
+ /**
+ * Default timeout for tool execution in seconds.
+ */
private static final int DEFAULT_TIMEOUT_SECONDS = 30;
-
+
+ /**
+ * MCP tool call attribute marker to prevent infinite loops.
+ */
+ private static final String MCP_TOOL_CALL_ATTR = "MCP_TOOL_CALL";
+
+ /**
+ * MCP session ID attribute key.
+ */
+ private static final String MCP_SESSION_ID_ATTR = "MCP_TOOL_SESSION_ID";
+
+ /**
+ * Streamable HTTP protocol path indicator.
+ */
+ private static final String STREAMABLE_HTTP_PATH = "/streamablehttp";
+
private final ToolDefinition toolDefinition;
-
+
+ /**
+ * Constructs a new Shenyu tool callback with the specified tool definition.
+ *
+ * @param toolDefinition the tool definition that this callback will handle
+ * @throws NullPointerException if toolDefinition is null
+ */
public ShenyuToolCallback(final ToolDefinition toolDefinition) {
this.toolDefinition = Objects.requireNonNull(toolDefinition, "ToolDefinition cannot be null");
}
-
+
@NonNull
@Override
public ToolDefinition getToolDefinition() {
return this.toolDefinition;
}
-
+
@NonNull
@Override
public String call(@NonNull final String input) {
return call(input, new ToolContext(Maps.newHashMap()));
}
-
+
@NonNull
@Override
public String call(@NonNull final String input, final ToolContext toolContext) {
Objects.requireNonNull(input, "Input cannot be null");
Objects.requireNonNull(toolContext, "ToolContext cannot be null");
-
+
+ LOG.debug("Executing tool call for definition '{}' with input length: {} chars",
+ toolDefinition.name(), input.length());
+
try {
- // Get MCP session information
- final McpSyncServerExchange mcpSyncServerExchange = getMcpSyncServerExchange(toolContext);
- final String sessionId = getSessionId(mcpSyncServerExchange);
-
- // Get tool definition and request configuration
- final ShenyuToolDefinition shenyuToolDefinition = validateAndGetToolDefinition();
- final String configStr = validateAndGetRequestConfig(shenyuToolDefinition);
-
- // Get original exchange object
- final ServerWebExchange originExchange = ShenyuMcpExchangeHolder.get(sessionId);
- if (Objects.isNull(originExchange)) {
- throw new IllegalStateException("No exchange found for sessionId: " + sessionId);
- }
- ShenyuPluginChain chain = originExchange.getAttribute(Constants.CHAIN);
- Assert.notNull(chain, "ShenyuPluginChain cannot be null");
-
- final CompletableFuture<String> future = new CompletableFuture<>();
-
- // Build new exchange object
- final ServerWebExchange decoratedExchange = buildExchange(originExchange, future, sessionId, configStr, input);
-
- LOG.info("[MCP] Executing plugin chain for session: {}", sessionId);
- chain.execute(decoratedExchange).doOnSubscribe(s -> LOG.info("[MCP] Subscribed to plugin chain for session: {}", sessionId)).doOnError(e -> {
- LOG.error("[MCP] Error processing request for session: {}", sessionId, e);
- future.completeExceptionally(e);
- }).doOnSuccess(v -> LOG.info("[MCP] Plugin chain completed successfully for session: {}", sessionId)).doOnCancel(() -> {
- LOG.warn("[MCP] Plugin chain was cancelled for session: {}", sessionId);
- future.completeExceptionally(new RuntimeException("Request was cancelled"));
- }).subscribe();
-
- try {
- return future.get(DEFAULT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- } catch (Exception e) {
- LOG.error("[MCP] Error waiting for response for session: {}", sessionId, e);
- throw new RuntimeException("Error waiting for response: " + e.getMessage(), e);
- }
+ // Extract MCP session context (must be pre-established)
+ final McpSyncServerExchange mcpExchange = extractMcpExchange(toolContext);
+ final String sessionId = extractSessionId(mcpExchange);
+
+ // Validate and extract tool configuration
+ final ShenyuToolDefinition shenyuTool = validateToolDefinition();
+ final String configStr = extractRequestConfig(shenyuTool);
+
+ // Get pre-stored exchange and plugin chain
+ final ServerWebExchange originExchange = getOriginExchange(sessionId);
+ final ShenyuPluginChain chain = getPluginChain(originExchange);
+
+ // Execute the tool call through the plugin chain
+ return executeToolCall(originExchange, chain, sessionId, configStr, input);
+
} catch (Exception e) {
- LOG.error("Failed to process request: {}", e.getMessage(), e);
- throw new RuntimeException("Failed to process request: " + e.getMessage(), e);
- }
- }
-
- private McpSyncServerExchange getMcpSyncServerExchange(final ToolContext toolContext) {
- final McpSyncServerExchange exchange = McpSessionHelper.getMcpSyncServerExchange(toolContext);
- if (Objects.isNull(exchange)) {
- throw new IllegalStateException("Failed to get MCP sync server exchange");
- }
- return exchange;
- }
-
- private String getSessionId(final McpSyncServerExchange exchange) throws NoSuchFieldException, IllegalAccessException {
- final String sessionId = McpSessionHelper.getSessionId(exchange);
- if (!StringUtils.hasText(sessionId)) {
- throw new IllegalStateException("Session ID cannot be empty");
+ LOG.error("Failed to process tool call for '{}': {}", toolDefinition.name(), e.getMessage(), e);
+
+ throw new RuntimeException("Tool execution failed: " + e.getMessage(), e);
}
- return sessionId;
}
-
- private ShenyuToolDefinition validateAndGetToolDefinition() {
+
+ /**
+ * Validates and casts the tool definition to Shenyu-specific type.
+ *
+ * @return the Shenyu tool definition
+ * @throws IllegalStateException if tool definition is not of expected type
+ */
+ private ShenyuToolDefinition validateToolDefinition() {
if (!(this.toolDefinition instanceof ShenyuToolDefinition)) {
- throw new IllegalStateException("Invalid tool definition type");
+ throw new IllegalStateException("Tool definition must be of type ShenyuToolDefinition, got: "
+ + this.toolDefinition.getClass().getSimpleName());
}
return (ShenyuToolDefinition) this.toolDefinition;
}
-
- private String validateAndGetRequestConfig(final ShenyuToolDefinition definition) {
+
+ /**
+ * Extracts and validates the request configuration from the tool definition.
+ *
+ * @param definition the Shenyu tool definition
+ * @return the request configuration string
+ * @throws IllegalStateException if configuration is missing or invalid
+ */
+ private String extractRequestConfig(final ShenyuToolDefinition definition) {
final String config = definition.requestConfig();
if (!StringUtils.hasText(config)) {
throw new IllegalStateException("Request configuration cannot be empty");
}
- LOG.debug("Request configuration: {}", config);
+ LOG.debug("Using request configuration with length: {} chars", config.length());
return config;
}
-
- private ServerWebExchange buildExchange(final ServerWebExchange originExchange, final CompletableFuture<String> future, final String sessionId, final String configStr, final String input) {
+
+ /**
+ * Extracts the plugin chain from the exchange.
+ *
+ * @param exchange the server web exchange
+ * @return the plugin chain
+ * @throws IllegalStateException if chain is not found
+ */
+ private ShenyuPluginChain getPluginChain(final ServerWebExchange exchange) {
+ final ShenyuPluginChain chain = exchange.getAttribute(Constants.CHAIN);
+ Assert.notNull(chain, "ShenyuPluginChain cannot be null");
+ return chain;
+ }
+
+ /**
+ * Executes the tool call through the Shenyu plugin chain.
+ *
+ * @param originExchange the original server web exchange
+ * @param chain the plugin chain to execute
+ * @param sessionId the MCP session identifier
+ * @param configStr the request configuration
+ * @param input the tool input parameters
+ * @return the execution result
+ */
+ private String executeToolCall(final ServerWebExchange originExchange,
+ final ShenyuPluginChain chain,
+ final String sessionId,
+ final String configStr,
+ final String input) {
+
+ final CompletableFuture<String> responseFuture = new CompletableFuture<>();
+ final ServerWebExchange decoratedExchange = buildDecoratedExchange(
+ originExchange, responseFuture, sessionId, configStr, input);
+
+ LOG.debug("Executing plugin chain for session: {}", sessionId);
+
+ // Check if this is a temporary session that needs cleanup
+ final boolean isTemporarySession = sessionId.startsWith("temp_");
+
+ // Execute the plugin chain asynchronously
+ chain.execute(decoratedExchange)
+ .doOnSubscribe(s -> LOG.debug("Plugin chain subscribed for session: {}", sessionId))
+ .doOnError(e -> {
+ LOG.error("Plugin chain execution failed for session {}: {}", sessionId, e.getMessage(), e);
+ responseFuture.completeExceptionally(e);
+ })
+ .doOnSuccess(v -> LOG.debug("Plugin chain completed successfully for session: {}", sessionId))
+ .doOnCancel(() -> {
+ LOG.warn("Plugin chain execution cancelled for session: {}", sessionId);
+ responseFuture.completeExceptionally(new RuntimeException("Execution was cancelled"));
+ })
+ .doFinally(signalType -> {
+ // Clean up temporary sessions after execution
+ if (isTemporarySession) {
+ LOG.debug("Cleaning up temporary session: {} (signal: {})", sessionId, signalType);
+ ShenyuMcpExchangeHolder.remove(sessionId);
+ }
+ })
+ .subscribe();
+
+ // Wait for the response with timeout
+ try {
+ final String result = responseFuture.get(DEFAULT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ LOG.debug("Tool call completed successfully for session: {}", sessionId);
+ return result;
+ } catch (Exception e) {
+ LOG.error("Timeout or error waiting for response for session {}: {}", sessionId, e.getMessage(), e);
+
+ // Ensure cleanup on error for temporary sessions
+ if (isTemporarySession) {
+ LOG.debug("Emergency cleanup of temporary session on error: {}", sessionId);
+ ShenyuMcpExchangeHolder.remove(sessionId);
+ }
+
+ throw new RuntimeException("Tool execution timeout or error: " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Builds a decorated ServerWebExchange for tool execution.
+ * <p>Creates a new exchange with modified request (method, path, headers, body),
+ * response decorator based on protocol type, and updated Shenyu context and metadata.</p>
+ *
+ * @param originExchange the original exchange
+ * @param responseFuture the future for capturing response
+ * @param sessionId the session identifier
+ * @param configStr the request configuration
+ * @param input the tool input parameters
+ * @return the decorated exchange ready for execution
+ */
+ private ServerWebExchange buildDecoratedExchange(final ServerWebExchange originExchange,
+ final CompletableFuture<String> responseFuture,
+ final String sessionId,
+ final String configStr,
+ final String input) {
+
+ // Parse input and configuration
+ final JsonObject inputJson = parseInput(input);
+ final RequestConfig requestConfig = buildRequestConfig(configStr, inputJson);
+
+ // Build decorated request
+ final ServerHttpRequest decoratedRequest = buildDecoratedRequest(
+ originExchange, sessionId, requestConfig);
+
+ // Build response decorator based on protocol
+ final ServerHttpResponseDecorator responseDecorator = createResponseDecorator(
+ originExchange, sessionId, responseFuture, configStr);
+
+ // Create base decorated exchange
+ final ServerWebExchange decoratedExchange = originExchange.mutate()
+ .request(decoratedRequest)
+ .response(responseDecorator)
+ .build();
+ // Handle request body if needed
+ final ServerWebExchange finalExchange = handleRequestBody(decoratedExchange, requestConfig);
+
+ // Configure Shenyu context and metadata
+ configureShenyuContext(finalExchange, sessionId, requestConfig.getPath(), configStr);
+ return finalExchange;
+ }
+
+ /**
+ * Parses the input JSON string into a JsonObject.
+ *
+ * @param input the input JSON string
+ * @return the parsed JsonObject
+ * @throws IllegalArgumentException if input is not valid JSON
+ */
+ private JsonObject parseInput(final String input) {
+ try {
+ final JsonObject inputJson = GsonUtils.getInstance().fromJson(input, JsonObject.class);
+ if (Objects.isNull(inputJson)) {
+ throw new IllegalArgumentException("Invalid input JSON format");
+ }
+ return inputJson;
+ } catch (Exception e) {
+ LOG.error("Failed to parse input JSON: {}", e.getMessage());
+ throw new IllegalArgumentException("Invalid JSON format: " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Builds request configuration from configuration string and input parameters.
+ *
+ * @param configStr the configuration string
+ * @param inputJson the input parameters
+ * @return the built request configuration
+ */
+ private RequestConfig buildRequestConfig(final String configStr, final JsonObject inputJson) {
final RequestConfigHelper configHelper = new RequestConfigHelper(configStr);
- // Parse input parameters
- BuildRequestConfigResult result = getBuildRequestConfig(configHelper, input);
- RequestConfig requestConfig = result.requestConfig();
-
+ final JsonObject requestTemplate = configHelper.getRequestTemplate();
+ final JsonObject argsPosition = configHelper.getArgsPosition();
+ final String urlTemplate = configHelper.getUrlTemplate();
+ final String method = configHelper.getMethod();
+ final boolean argsToJsonBody = configHelper.isArgsToJsonBody();
+
+ // Build path and query parameters (no formatting needed for URLs)
+ final String path = RequestConfigHelper.buildPath(urlTemplate, argsPosition, inputJson);
+
+ // Build body with parameter formatting (only format body parameters that need it)
+ final JsonObject bodyJson = buildFormattedBodyJson(argsToJsonBody, argsPosition, inputJson);
+
+ return new RequestConfig(method, path, bodyJson, requestTemplate, argsToJsonBody);
+ }
+
+ /**
+ * Build body JSON with parameter formatting for specific types.
+ * Only formats parameters that are mapped to body and need type conversion.
+ *
+ * @param argsToJsonBody whether to convert arguments to JSON body
+ * @param argsPosition the argument position mapping
+ * @param inputJson the input JSON object
+ * @return the constructed body JSON object
+ */
+ private JsonObject buildFormattedBodyJson(final boolean argsToJsonBody, final JsonObject argsPosition, final JsonObject inputJson) {
+ JsonObject bodyJson = new JsonObject();
+
+ if (!argsToJsonBody) {
+ return bodyJson;
+ }
+
+ for (String key : argsPosition.keySet()) {
+ String position = argsPosition.get(key).getAsString();
+ if (position.startsWith("body") && inputJson.has(key)) {
+ JsonElement value = inputJson.get(key);
+
+ // Format the value if it's a JSON string that should be parsed
+ JsonElement formattedValue = formatBodyParameterValue(value, key);
+
+ if ("body".equals(position)) {
+ bodyJson.add(key, formattedValue);
+ } else if (position.startsWith("body.")) {
+ String[] pathParts = position.substring(5).split("\\.");
+ setNestedValue(bodyJson, pathParts, formattedValue);
+ }
+ }
+ }
+
+ return bodyJson;
+ }
+
+ /**
+ * Format body parameter value. Only handles JSON string parsing for complex types.
+ *
+ * @param value the parameter value
+ * @param paramName the parameter name for error messages
+ * @return the formatted value
+ */
+ private JsonElement formatBodyParameterValue(final JsonElement value, final String paramName) {
+ // If it's a string, try to parse it as JSON (for array/object types)
+ if (value.isJsonPrimitive() && value.getAsJsonPrimitive().isString()) {
+ String stringValue = value.getAsString();
+ JsonElement parsed = ParameterFormatter.tryParseJsonString(stringValue);
+
+ // If parsing succeeded (returned different object), log it
+ if (!parsed.equals(value)) {
+ LOG.debug("Parsed JSON string parameter '{}' into {}", paramName,
+ parsed.isJsonArray() ? "array" : "object");
+ }
+
+ return parsed;
+ }
+
+ // Return as-is for all other cases
+ return value;
+ }
+
+ /**
+ * Set nested value in JSON object.
+ */
+ private void setNestedValue(final JsonObject jsonObject, final String[] pathParts, final JsonElement value) {
+ JsonObject current = jsonObject;
+
+ for (int i = 0; i < pathParts.length - 1; i++) {
+ String part = pathParts[i];
+ if (!current.has(part)) {
+ current.add(part, new JsonObject());
+ }
+ current = current.getAsJsonObject(part);
+ }
+
+ current.add(pathParts[pathParts.length - 1], value);
+ }
+
+ /**
+ * Builds a decorated HTTP request with modified method, path, and headers.
+ *
+ * @param originExchange the original exchange
+ * @param sessionId the session identifier
+ * @param requestConfig the request configuration
+ * @return the decorated HTTP request
+ */
+ private ServerHttpRequest buildDecoratedRequest(final ServerWebExchange originExchange,
+ final String sessionId,
+ final RequestConfig requestConfig) {
+
final ServerHttpRequest.Builder requestBuilder = originExchange
.getRequest()
.mutate()
.method(HttpMethod.valueOf(requestConfig.getMethod()))
.header("sessionId", sessionId)
.header("Accept", "application/json");
-
- // Add custom headers
+
+ // Add custom headers from template
+ addCustomHeaders(requestBuilder, requestConfig);
+
+ // Set content type for body methods
+ configureContentType(requestBuilder, requestConfig.getMethod());
+
+ // Set the target URI
+ setTargetUri(requestBuilder, originExchange, requestConfig.getPath());
+
+ return requestBuilder.build();
+ }
+
+ /**
+ * Adds custom headers from the request template to the request builder.
+ *
+ * @param requestBuilder the request builder
+ * @param requestConfig the request configuration
+ */
+ private void addCustomHeaders(final ServerHttpRequest.Builder requestBuilder,
+ final RequestConfig requestConfig) {
if (requestConfig.getRequestTemplate().has("headers")) {
for (final var headerElem : requestConfig.getRequestTemplate().getAsJsonArray("headers")) {
final JsonObject headerObj = headerElem.getAsJsonObject();
- requestBuilder.header(headerObj.get("key").getAsString(), headerObj.get("value").getAsString());
+ requestBuilder.header(
+ headerObj.get("key").getAsString(),
+ headerObj.get("value").getAsString()
+ );
}
}
-
- // Set Content-Type
- if (RequestConfigHelper.isRequestBodyMethod(requestConfig.getMethod())) {
+ }
+
+ /**
+ * Configures content type based on HTTP method.
+ *
+ * @param requestBuilder the request builder
+ * @param method the HTTP method
+ */
+ private void configureContentType(final ServerHttpRequest.Builder requestBuilder, final String method) {
+ if (isRequestBodyMethod(method)) {
requestBuilder.header("Content-Type", "application/json");
} else {
requestBuilder.headers(httpHeaders -> httpHeaders.remove("Content-Type"));
}
-
- // Set URI
+ }
+
+ /**
+ * Sets the target URI for the request.
+ *
+ * @param requestBuilder the request builder
+ * @param originExchange the original exchange
+ * @param path the target path
+ */
+ private void setTargetUri(final ServerHttpRequest.Builder requestBuilder,
+ final ServerWebExchange originExchange,
+ final String path) {
try {
final URI oldUri = originExchange.getRequest().getURI();
- final String newUriStr = oldUri.getScheme() + "://" + oldUri.getAuthority() + result.path();
+ final String newUriStr = oldUri.getScheme() + "://" + oldUri.getAuthority() + path;
requestBuilder.uri(new URI(newUriStr));
} catch (URISyntaxException e) {
- throw new RuntimeException("Invalid URI: " + e.getMessage(), e);
- }
-
- // Create response decorator
- final ServerHttpResponseDecorator responseDecorator = new ShenyuMcpResponseDecorator(originExchange.getResponse(), sessionId, future, configHelper.getResponseTemplate());
-
- // Build final exchange object
- final ServerWebExchange decoratedExchange = originExchange.mutate().request(requestBuilder.build()).response(responseDecorator).build();
-
- // Handle request body
- if (RequestConfigHelper.isRequestBodyMethod(requestConfig.getMethod())
- && requestConfig.getBodyJson().size() > 0) {
+ throw new RuntimeException("Invalid URI construction: " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Creates appropriate response decorator based on protocol type.
+ *
+ * @param originExchange the original exchange
+ * @param sessionId the session identifier
+ * @param responseFuture the response future
+ * @param configStr the configuration string (for response template)
+ * @return the appropriate response decorator
+ */
+ private ServerHttpResponseDecorator createResponseDecorator(final ServerWebExchange originExchange,
+ final String sessionId,
+ final CompletableFuture<String> responseFuture,
+ final String configStr) {
+
+ final RequestConfigHelper configHelper = new RequestConfigHelper(configStr);
+ final JsonObject responseTemplate = configHelper.getResponseTemplate();
+
+ if (isStreamableHttpProtocol(originExchange)) {
+ LOG.debug("Using non-committing decorator for Streamable HTTP protocol, session: {}", sessionId);
+ return new NonCommittingMcpResponseDecorator(
+ originExchange.getResponse(), sessionId, responseFuture, responseTemplate);
+ } else {
+ LOG.debug("Using standard decorator for SSE protocol, session: {}", sessionId);
+ return new ShenyuMcpResponseDecorator(
+ originExchange.getResponse(), sessionId, responseFuture, responseTemplate);
+ }
+ }
+
+ /**
+ * Handles request body for methods that support it.
+ *
+ * @param decoratedExchange the decorated exchange
+ * @param requestConfig the request configuration
+ * @return the exchange with body handling applied
+ */
+ private ServerWebExchange handleRequestBody(final ServerWebExchange decoratedExchange,
+ final RequestConfig requestConfig) {
+ if (isRequestBodyMethod(requestConfig.getMethod()) && requestConfig.getBodyJson().size() > 0) {
return new BodyWriterExchange(decoratedExchange, requestConfig.getBodyJson().toString());
}
-
- // Set context and attributes
- final ShenyuContext shenyuContext = originExchange.getAttribute(Constants.CONTEXT);
+ return decoratedExchange;
+ }
+
+ /**
+ * Configures Shenyu context and metadata for the decorated exchange.
+ *
+ * @param decoratedExchange the decorated exchange
+ * @param sessionId the session identifier
+ * @param decoratedPath the decorated request path
+ * @param configStr configStr.
+ */
+ private void configureShenyuContext(final ServerWebExchange decoratedExchange,
+ final String sessionId,
+ final String decoratedPath,
+ final String configStr) {
+
+ final ShenyuContext shenyuContext = decoratedExchange.getAttribute(Constants.CONTEXT);
if (Objects.nonNull(shenyuContext)) {
- String decoratedPath = decoratedExchange.getRequest().getPath().value();
- MetaData metaData = MetaDataCache.getInstance().obtain(decoratedPath);
+ // Set metadata if available
+ configureMetadata(decoratedExchange, decoratedPath, shenyuContext);
+
+ final RequestConfigHelper configHelper = new RequestConfigHelper(configStr);
+ MetaData metaData = MetaDataCache.getInstance().obtain(configHelper.getUrlTemplate());
if (Objects.nonNull(metaData) && Boolean.TRUE.equals(metaData.getEnabled())) {
decoratedExchange.getAttributes().put(Constants.META_DATA, metaData);
shenyuContext.setRpcType(metaData.getRpcType());
- } else {
- shenyuContext.setRpcType(RpcTypeEnum.HTTP.getName());
}
shenyuContext.setPath(decoratedPath);
shenyuContext.setRealUrl(decoratedPath);
+
+ LOG.debug("Configured RpcType to HTTP for tool call, session: {}", sessionId);
+
decoratedExchange.getAttributes().put(Constants.CONTEXT, shenyuContext);
+
+ // Add MCP tool call markers to prevent loops
+ decoratedExchange.getAttributes().put(MCP_TOOL_CALL_ATTR, true);
+ decoratedExchange.getAttributes().put(MCP_SESSION_ID_ATTR, sessionId);
}
-
- return decoratedExchange;
}
-
- private static BuildRequestConfigResult getBuildRequestConfig(final RequestConfigHelper configHelper, final String input) {
- final JsonObject inputJson = GsonUtils.getInstance().fromJson(input, JsonObject.class);
-
- if (Objects.isNull(inputJson)) {
- throw new IllegalArgumentException("Invalid input JSON format");
+
+ /**
+ * Configures metadata for the request if available.
+ *
+ * @param decoratedExchange the decorated exchange
+ * @param decoratedPath the request path
+ * @param shenyuContext the Shenyu context
+ */
+ private void configureMetadata(final ServerWebExchange decoratedExchange,
+ final String decoratedPath,
+ final ShenyuContext shenyuContext) {
+
+ final MetaData metaData = MetaDataCache.getInstance().obtain(decoratedPath);
+ if (Objects.nonNull(metaData) && Boolean.TRUE.equals(metaData.getEnabled())) {
+ decoratedExchange.getAttributes().put(Constants.META_DATA, metaData);
+ // Set metadata RPC type first (will be overridden to HTTP)
+ shenyuContext.setRpcType(metaData.getRpcType());
+ LOG.debug("Applied metadata for path: {}", decoratedPath);
}
- // Parse request configuration
- final JsonObject requestTemplate = configHelper.getRequestTemplate();
- final JsonObject argsPosition = configHelper.getArgsPosition();
- final String urlTemplate = configHelper.getUrlTemplate();
- final String method = configHelper.getMethod();
- final boolean argsToJsonBody = configHelper.isArgsToJsonBody();
-
- final String path = RequestConfigHelper.buildPath(urlTemplate, argsPosition, inputJson);
- final JsonObject bodyJson = RequestConfigHelper.buildBodyJson(argsToJsonBody, argsPosition, inputJson);
-
- final RequestConfig requestConfig = new RequestConfig(method, path, bodyJson, requestTemplate, argsToJsonBody);
- return new BuildRequestConfigResult(path, requestConfig);
}
-
- private record BuildRequestConfigResult(String path, RequestConfig requestConfig) {
+
+ /**
+ * Checks if the protocol is Streamable HTTP based on URI path.
+ *
+ * @param exchange the server web exchange
+ * @return true if Streamable HTTP protocol is detected
+ */
+ private boolean isStreamableHttpProtocol(final ServerWebExchange exchange) {
+ final String uri = exchange.getRequest().getURI().getRawPath();
+ final boolean isStreamable = uri.contains(STREAMABLE_HTTP_PATH) || uri.endsWith(STREAMABLE_HTTP_PATH);
+ LOG.debug("Protocol detection - URI: {}, isStreamableHttp: {}", uri, isStreamable);
+ return isStreamable;
+ }
+
+ /**
+ * Checks if the HTTP method supports request body.
+ *
+ * @param method the HTTP method name
+ * @return true if the method supports request body
+ */
+ private boolean isRequestBodyMethod(final String method) {
+ return "POST".equalsIgnoreCase(method)
+ || "PUT".equalsIgnoreCase(method)
+ || "PATCH".equalsIgnoreCase(method);
+ }
+
+ /**
+ * Extracts the MCP sync server exchange from the tool context.
+ *
+ * @param toolContext the tool context containing MCP session information
+ * @return the MCP sync server exchange
+ * @throws IllegalStateException if exchange cannot be retrieved
+ */
+ private McpSyncServerExchange extractMcpExchange(final ToolContext toolContext) {
+ final McpSyncServerExchange exchange = McpSessionHelper.getMcpSyncServerExchange(toolContext);
+ if (Objects.isNull(exchange)) {
+ throw new IllegalStateException("Failed to retrieve MCP sync server exchange from context");
+ }
+ return exchange;
+ }
+
+ /**
+ * Extracts the session ID from the MCP sync server exchange.
+ *
+ * @param mcpExchange the MCP sync server exchange
+ * @return the session ID
+ * @throws IllegalStateException if session ID cannot be extracted
+ */
+ private String extractSessionId(final McpSyncServerExchange mcpExchange) {
+ final String sessionId;
+ try {
+ sessionId = McpSessionHelper.getSessionId(mcpExchange);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ if (StringUtils.hasText(sessionId)) {
+ LOG.debug("Extracted session ID: {}", sessionId);
+ return sessionId;
+ }
+ throw new IllegalStateException("Session ID is empty – it should have been set earlier by handleMessageEndpoint");
+ }
+
+ /**
+ * Gets the origin ServerWebExchange for the given session ID.
+ *
+ * @param sessionId the session ID
+ * @return the origin ServerWebExchange
+ * @throws IllegalStateException if exchange cannot be retrieved
+ */
+ private ServerWebExchange getOriginExchange(final String sessionId) {
+ final ServerWebExchange exchange = ShenyuMcpExchangeHolder.get(sessionId);
+ if (Objects.nonNull(exchange)) {
+ LOG.debug("Found existing exchange for session: {}", sessionId);
+ return exchange;
+ }
+ throw new IllegalStateException("No ServerWebExchange found for session '" + sessionId
+ + "'. It should have been stored by handleMessageEndpoint before the tool was invoked.");
}
}
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/handler/McpServerPluginDataHandler.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/handler/McpServerPluginDataHandler.java
index d04959bec8..d225f1bd5a 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/handler/McpServerPluginDataHandler.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/handler/McpServerPluginDataHandler.java
@@ -46,30 +46,32 @@ import java.util.function.Supplier;
* The type McpServer plugin data handler.
*/
public class McpServerPluginDataHandler implements PluginDataHandler {
-
- public static final Supplier<CommonHandleCache<String, ShenyuMcpServerTool>> CACHED_TOOL = new BeanHolder<>(
- CommonHandleCache::new);
-
+
public static final Supplier<CommonHandleCache<String, ShenyuMcpServer>> CACHED_SERVER = new BeanHolder<>(
CommonHandleCache::new);
-
+
+ public static final Supplier<CommonHandleCache<String, ShenyuMcpServerTool>> CACHED_TOOL = new BeanHolder<>(
+ CommonHandleCache::new);
+
+ private static final String DEFAULT_MESSAGE_ENDPOINT = "{\"messageEndpoint\":\"/message\"}";
+
private static final String SLASH = "/";
-
+
private static final String STAR = "/**";
-
+
private final ShenyuMcpServerManager shenyuMcpServerManager;
-
+
public McpServerPluginDataHandler(
final ShenyuMcpServerManager shenyuMcpServerManager) {
this.shenyuMcpServerManager = shenyuMcpServerManager;
}
-
+
@Override
public void handlerSelector(final SelectorData selectorData) {
if (Objects.isNull(selectorData) || Objects.isNull(selectorData.getId())) {
return;
}
-
+
if (CollectionUtils.isEmpty(selectorData.getConditionList())) {
return;
}
@@ -80,10 +82,10 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
.map(ConditionData::getParamValue)
.findFirst()
.orElse(null);
-
+
String path = StringUtils.removeEnd(uri, SLASH);
path = StringUtils.removeEnd(path, STAR);
- ShenyuMcpServer shenyuMcpServer = GsonUtils.getInstance().fromJson(selectorData.getHandle(), ShenyuMcpServer.class);
+ ShenyuMcpServer shenyuMcpServer = GsonUtils.getInstance().fromJson(Objects.isNull(selectorData.getHandle()) ? DEFAULT_MESSAGE_ENDPOINT : selectorData.getHandle(), ShenyuMcpServer.class);
shenyuMcpServer.setPath(path);
CACHED_SERVER.get().cachedHandle(
selectorData.getId(),
@@ -93,18 +95,18 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
if (StringUtils.isNotBlank(uri) && !shenyuMcpServerManager.hasMcpServer(uri)) {
shenyuMcpServerManager.getOrCreateMcpServerTransport(uri, messageEndpoint);
}
-
+
// the update is also need to clean, but there is no way to
// distinguish between crate and update, so it is always clean
MetaDataCache.getInstance().clean();
}
-
+
@Override
public void removeSelector(final SelectorData selectorData) {
UpstreamCacheManager.getInstance().removeByKey(selectorData.getId());
MetaDataCache.getInstance().clean();
CACHED_TOOL.get().removeHandle(CacheKeyUtils.INST.getKey(selectorData.getId(), Constants.DEFAULT_RULE));
-
+
// Remove the McpServer for this URI
// First try to get URI from handle, then from condition list
String uri = selectorData.getHandle();
@@ -116,14 +118,14 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
.findFirst()
.orElse(null);
}
-
+
CACHED_SERVER.get().removeHandle(selectorData.getId());
-
+
if (StringUtils.isNotBlank(uri) && shenyuMcpServerManager.hasMcpServer(uri)) {
shenyuMcpServerManager.removeMcpServer(uri);
}
}
-
+
@Override
public void handlerRule(final RuleData ruleData) {
Optional.ofNullable(ruleData.getHandle()).ifPresent(s -> {
@@ -132,9 +134,9 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
// the update is also need to clean, but there is no way to
// distinguish between crate and update, so it is always clean
MetaDataCache.getInstance().clean();
-
+
List<McpServerToolParameter> parameters = mcpServerTool.getParameters();
-
+
// Create JSON schema from parameters
String inputSchema = JsonSchemaUtil.createParameterSchema(parameters);
ShenyuMcpServer server = CACHED_SERVER.get().obtainHandle(ruleData.getSelectorId());
@@ -148,7 +150,7 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
}
});
}
-
+
@Override
public void removeRule(final RuleData ruleData) {
Optional.ofNullable(ruleData.getHandle()).ifPresent(s -> {
@@ -158,10 +160,10 @@ public class McpServerPluginDataHandler implements PluginDataHandler {
});
MetaDataCache.getInstance().clean();
}
-
+
@Override
public String pluginNamed() {
return PluginEnum.MCP_SERVER.getName();
}
-
+
}
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/manager/ShenyuMcpServerManager.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/manager/ShenyuMcpServerManager.java
index 07d337eba1..fa2cb92143 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/manager/ShenyuMcpServerManager.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/manager/ShenyuMcpServerManager.java
@@ -23,10 +23,11 @@ import io.modelcontextprotocol.server.McpAsyncServer;
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpServerFeatures.AsyncToolSpecification;
import io.modelcontextprotocol.spec.McpSchema;
-import org.apache.commons.lang3.StringUtils;
+import io.modelcontextprotocol.spec.McpServerSession;
import org.apache.shenyu.plugin.mcp.server.callback.ShenyuToolCallback;
import org.apache.shenyu.plugin.mcp.server.definition.ShenyuToolDefinition;
import org.apache.shenyu.plugin.mcp.server.transport.ShenyuSseServerTransportProvider;
+import org.apache.shenyu.plugin.mcp.server.transport.ShenyuStreamableHttpServerTransportProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.ai.mcp.McpToolUtils;
@@ -38,16 +39,30 @@ import org.springframework.web.reactive.function.server.HandlerFunction;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.Collections;
/**
- * Manager for MCP servers. Maps URIs to McpServer instances.
+ * Enhanced Manager for MCP servers supporting shared server instances across multiple transport protocols.
+ *
+ * <p>This manager implements a unified architecture where SSE and Streamable HTTP protocols share
+ * the same McpAsyncServer instance per path, with shared tool sets and capabilities.
+ * Provides centralized server and tool management with protocol-specific transport layers.</p>
+ *
+ * @since 2.7.0.2
*/
@Component
public class ShenyuMcpServerManager {
private static final Logger LOG = LoggerFactory.getLogger(ShenyuMcpServerManager.class);
- private static final String SLASH = "/";
+ /**
+ * Protocol names for transport identification.
+ */
+ private static final String SSE_PROTOCOL = "SSE";
+
+ private static final String STREAMABLE_HTTP_PROTOCOL = "Streamable HTTP";
/**
* AntPathMatcher for pattern matching.
@@ -55,63 +70,222 @@ public class ShenyuMcpServerManager {
private final AntPathMatcher pathMatcher = new AntPathMatcher();
/**
- * Map to store URI to McpServer mapping.
+ * Shared ObjectMapper instance for JSON processing.
*/
- private final Map<String, McpAsyncServer> mcpServerMap = new ConcurrentHashMap<>();
+ private final ObjectMapper objectMapper = new ObjectMapper();
/**
- * Map to store URI to McpServer mapping.
+ * Map to store normalized path to shared McpAsyncServer mapping.
+ * Key: normalized server path, Value: shared McpAsyncServer instance
*/
- private final Map<String, ShenyuSseServerTransportProvider> mcpServerTransportMap = new ConcurrentHashMap<>();
+ private final Map<String, McpAsyncServer> sharedServerMap = new ConcurrentHashMap<>();
+ /**
+ * Map to store route handlers for different endpoints.
+ */
private final Map<String, HandlerFunction<?>> routeMap = new ConcurrentHashMap<>();
/**
- * Get or create a McpServer instance for the given URI.
- * Uses pattern matching to find the best matching server.
+ * Map to store composite transport providers for shared servers.
+ */
+ private final Map<String, CompositeTransportProvider> compositeTransportMap = new ConcurrentHashMap<>();
+
+ /**
+ * Get or create a shared MCP server for the given path, supporting multiple transport protocols.
+ *
+ * @param uri The URI to create or get a server for
+ * @param messageEndPoint The message endpoint path
+ * @return The SSE transport provider for the URI
+ */
+ public ShenyuSseServerTransportProvider getOrCreateMcpServerTransport(final String uri, final String messageEndPoint) {
+ String normalizedPath = processPath(uri);
+ return getOrCreateTransport(normalizedPath, SSE_PROTOCOL,
+ () -> createSseTransport(normalizedPath, messageEndPoint));
+ }
+
+ /**
+ * Get or create a shared MCP server for Streamable HTTP transport.
+ *
+ * @param uri The URI to create or get a transport provider for
+ * @return The Streamable HTTP transport provider for the URI
+ */
+ public ShenyuStreamableHttpServerTransportProvider getOrCreateStreamableHttpTransport(final String uri) {
+ String normalizedPath = processPath(uri);
+ return getOrCreateTransport(normalizedPath, STREAMABLE_HTTP_PROTOCOL,
+ () -> createStreamableHttpTransport(normalizedPath, uri));
+ }
+
+ /**
+ * Generic method to get or create transport providers.
*
- * @param uri The URI to create or get a McpServer for
- * @param messageEndpoint The message endpoint to use
- * @return The McpServer for the URI
+ * @param normalizedPath the normalized path
+ * @param protocol the protocol name
+ * @param transportFactory the factory function to create transport if not exists
+ * @param <T> the transport type
+ * @return the transport provider
*/
- public ShenyuSseServerTransportProvider getOrCreateMcpServerTransport(final String uri, final String messageEndpoint) {
- // First try exact match
- ShenyuSseServerTransportProvider transport = mcpServerTransportMap.get(uri);
- if (Objects.nonNull(transport)) {
- return transport;
+ @SuppressWarnings("unchecked")
+ private <T> T getOrCreateTransport(final String normalizedPath, final String protocol,
+ final java.util.function.Supplier<T> transportFactory) {
+ CompositeTransportProvider compositeTransport = getOrCreateCompositeTransport(normalizedPath);
+
+ T transport = switch (protocol) {
+ case SSE_PROTOCOL -> (T) compositeTransport.getSseTransport();
+ case STREAMABLE_HTTP_PROTOCOL -> (T) compositeTransport.getStreamableHttpTransport();
+ default -> null;
+ };
+
+ if (Objects.isNull(transport)) {
+ transport = transportFactory.get();
+ addTransportToSharedServer(normalizedPath, protocol, transport);
}
+
+ return transport;
+ }
- // Then try to find existing transport that matches the pattern
- String basePath = extractBasePath(uri, messageEndpoint);
- transport = mcpServerTransportMap.get(basePath);
- if (Objects.nonNull(transport)) {
- LOG.debug("Found existing transport for base path '{}' for URI '{}'", basePath, uri);
- return transport;
+ /**
+ * Process URI to get normalized path.
+ *
+ * @param uri the URI to process
+ * @return normalized path
+ */
+ private String processPath(final String uri) {
+ return normalizeServerPath(extractBasePath(uri));
+ }
+
+ /**
+ * Gets or creates a composite transport provider for the given normalized path.
+ *
+ * @param normalizedPath the normalized server path
+ * @return the composite transport provider
+ */
+ private CompositeTransportProvider getOrCreateCompositeTransport(final String normalizedPath) {
+ return compositeTransportMap.computeIfAbsent(normalizedPath, path -> {
+ LOG.debug("Creating composite transport provider for path: {}", path);
+ return new CompositeTransportProvider();
+ });
+ }
+
+ /**
+ * Adds a transport provider to the shared server infrastructure.
+ *
+ * @param normalizedPath the normalized server path
+ * @param protocol the protocol name
+ * @param transportProvider the transport provider instance
+ */
+ private void addTransportToSharedServer(final String normalizedPath, final String protocol, final Object transportProvider) {
+ // Get or create shared server
+ getOrCreateSharedServer(normalizedPath);
+
+ // Add transport to composite provider
+ CompositeTransportProvider compositeTransport = compositeTransportMap.get(normalizedPath);
+ if (Objects.nonNull(compositeTransport)) {
+ compositeTransport.addTransport(protocol, transportProvider);
}
- // Create new transport for the base path
- return mcpServerTransportMap.putIfAbsent(basePath, createMcpServerTransport(uri, messageEndpoint));
+ LOG.info("Added {} transport to shared server at path: {}", protocol, normalizedPath);
}
/**
- * Extract the base path from a URI by removing the /message suffix and any
- * sub-paths.
+ * Gets or creates a shared McpAsyncServer instance for the given normalized path.
+ *
+ * @param normalizedPath the normalized server path
+ * @return the shared McpAsyncServer instance
+ */
+ private McpAsyncServer getOrCreateSharedServer(final String normalizedPath) {
+ return sharedServerMap.computeIfAbsent(normalizedPath, path -> {
+ LOG.info("Creating shared MCP server for path: {}", path);
+
+ // Get or create composite transport provider
+ CompositeTransportProvider compositeTransport = getOrCreateCompositeTransport(path);
+
+ // Configure server capabilities
+ var capabilities = McpSchema.ServerCapabilities.builder()
+ .tools(true)
+ .logging()
+ .build();
+
+ // Create shared server with composite transport
+ McpAsyncServer server = McpServer
+ .async(compositeTransport)
+ .serverInfo("MCP Shenyu Server (Multi-Protocol)", "1.0.0")
+ .capabilities(capabilities)
+ .tools(Lists.newArrayList())
+ .build();
+
+ LOG.info("Created shared MCP server for path: {} with multi-protocol support", path);
+ return server;
+ });
+ }
+
+ /**
+ * Creates SSE transport provider.
+ */
+ private ShenyuSseServerTransportProvider createSseTransport(final String normalizedPath, final String messageEndPoint) {
+ String messageEndpoint = normalizedPath + messageEndPoint;
+ ShenyuSseServerTransportProvider transportProvider = ShenyuSseServerTransportProvider.builder()
+ .objectMapper(objectMapper)
+ .sseEndpoint(normalizedPath)
+ .messageEndpoint(messageEndpoint)
+ .build();
+
+ // Register routes
+ registerRoutes(normalizedPath, messageEndpoint, transportProvider::handleSseConnection, transportProvider::handleMessage);
+
+ LOG.debug("Created SSE transport for path: {}", normalizedPath);
+ return transportProvider;
+ }
+
+ /**
+ * Creates Streamable HTTP transport provider.
+ */
+ private ShenyuStreamableHttpServerTransportProvider createStreamableHttpTransport(final String normalizedPath, final String originalUri) {
+ ShenyuStreamableHttpServerTransportProvider transportProvider = ShenyuStreamableHttpServerTransportProvider.builder()
+ .objectMapper(objectMapper)
+ .endpoint(originalUri)
+ .build();
+
+ // Register routes for original URI
+ registerRoutes(originalUri, null, transportProvider::handleUnifiedEndpoint, null);
+
+ LOG.debug("Created Streamable HTTP transport for original URI: {} (normalized: {})", originalUri, normalizedPath);
+ return transportProvider;
+ }
+
+ /**
+ * Register routes for transport providers.
+ *
+ * @param primaryPath the primary path
+ * @param secondaryPath the secondary path (can be null)
+ * @param primaryHandler the primary handler
+ * @param secondaryHandler the secondary handler (can be null)
+ */
+ private void registerRoutes(final String primaryPath, final String secondaryPath,
+ final HandlerFunction<?> primaryHandler, final HandlerFunction<?> secondaryHandler) {
+ routeMap.put(primaryPath, primaryHandler);
+ routeMap.put(primaryPath + "/**", primaryHandler);
+
+ if (Objects.nonNull(secondaryPath) && Objects.nonNull(secondaryHandler)) {
+ routeMap.put(secondaryPath, secondaryHandler);
+ routeMap.put(secondaryPath + "/**", secondaryHandler);
+ }
+ }
+
+ /**
+ * Extract the base path from a URI by removing the /message suffix and any sub-paths.
*
* @param uri The URI to extract base path from
- * @param messageEndpoint Endpoint for the MCP Server
* @return The base path
*/
- private String extractBasePath(final String uri, final String messageEndpoint) {
+ private String extractBasePath(final String uri) {
String basePath = uri;
// Remove /message suffix if present
- if (basePath.endsWith(messageEndpoint)) {
- basePath = basePath.substring(0, basePath.length() - messageEndpoint.length());
+ if (basePath.endsWith("/message")) {
+ basePath = basePath.substring(0, basePath.length() - "/message".length());
}
// For sub-paths, extract the main MCP server path
- // This assumes MCP server paths don't contain multiple levels
- // You may need to adjust this logic based on your specific URL structure
String[] pathSegments = basePath.split("/");
if (pathSegments.length > 2) {
// Keep only the first two segments (empty + server-name)
@@ -128,12 +302,12 @@ public class ShenyuMcpServerManager {
* @return true if a McpServer exists, false otherwise
*/
public boolean hasMcpServer(final String uri) {
- return mcpServerMap.containsKey(uri);
+ String normalizedPath = processPath(uri);
+ return sharedServerMap.containsKey(normalizedPath);
}
/**
* Check if a McpServer can route requests for the given URI.
- * Uses AntPathMatcher to support pattern matching.
*
* @param uri The URI to check
* @return true if the URI is routable, false otherwise
@@ -155,97 +329,301 @@ public class ShenyuMcpServerManager {
return false;
}
-
/**
- * Remove a McpServer for the given URI.
+ * Remove a MCP server and all its associated transports.
*
- * @param uri The URI to remove the McpServer for
+ * @param uri The URI to remove the server for
*/
public void removeMcpServer(final String uri) {
- LOG.info("Removed McpServer for URI: {}", uri);
- ShenyuSseServerTransportProvider transportProvider = mcpServerTransportMap.remove(uri);
- if (Objects.nonNull(transportProvider)) {
- transportProvider
- .closeGracefully()
- .doOnSuccess(aVoid -> LOG.info("Successfully closed McpServer for URI: {}", uri))
- .doOnError(e -> LOG.error("Error closing McpServer for URI: {}", uri, e))
+ String normalizedPath = processPath(uri);
+ LOG.info("Removing MCP server for URI: {} (normalized: {})", uri, normalizedPath);
+
+ // Close composite transport gracefully
+ CompositeTransportProvider compositeTransport = compositeTransportMap.remove(normalizedPath);
+ if (Objects.nonNull(compositeTransport)) {
+ compositeTransport.closeGracefully()
+ .doOnSuccess(aVoid -> LOG.info("Successfully closed composite transport for path: {}", normalizedPath))
+ .doOnError(e -> LOG.error("Error closing composite transport for path: {}", normalizedPath, e))
.subscribe();
}
- mcpServerMap.remove(uri);
+
+ // Remove server
+ sharedServerMap.remove(normalizedPath);
+
+ // Clean up routes
+ routeMap.entrySet().removeIf(entry -> entry.getKey().startsWith(Objects.requireNonNull(normalizedPath)) || entry.getKey().startsWith(uri));
+
+ LOG.info("Removed MCP server for path: {}", normalizedPath);
}
/**
- * Create a new McpServer for the given URI.
+ * Adds a tool to the shared server instance, automatically available across all protocols.
*
- * @param uri The URI to create a McpServer for
- * @param messageEndpoint The message endpoint for the McpServer
- * @return The McpAsyncServer for the new McpServer
+ * @param serverPath the server path
+ * @param name the tool name
+ * @param description the tool description
+ * @param requestTemplate the request template
+ * @param inputSchema the input schema
*/
- private ShenyuSseServerTransportProvider createMcpServerTransport(final String uri, final String messageEndpoint) {
- String path = StringUtils.removeEnd(uri, SLASH);
- LOG.info("Creating new McpServer for URI: {}", path);
- String messagePath = path + messageEndpoint;
- ShenyuSseServerTransportProvider transportProvider = ShenyuSseServerTransportProvider.builder()
- .objectMapper(new ObjectMapper()).sseEndpoint(path).messageEndpoint(messagePath).build();
-
- LOG.debug("Registered RouterFunction for URI: {}", path);
-
- // Configure server capabilities with resource support
- var capabilities = McpSchema.ServerCapabilities.builder()
- // Tool support with list changes notifications
- .tools(true)
- // Logging support
- .logging().build();
-
- // Create the server with both tool and resource capabilities
- // Note: McpServer.async() automatically calls
- // transportProvider.setSessionFactory()
- McpAsyncServer server = McpServer
- .async(transportProvider)
- .serverInfo("MCP Shenyu Server", "1.0.0")
- .capabilities(capabilities)
- .tools(Lists.newArrayList())
+ public void addTool(final String serverPath, final String name, final String description,
+ final String requestTemplate, final String inputSchema) {
+ String normalizedPath = normalizeServerPath(serverPath);
+
+ // Remove existing tool first
+ try {
+ removeTool(serverPath, name);
+ } catch (Exception ignored) {
+ // ignore
+ }
+
+ ToolDefinition shenyuToolDefinition = ShenyuToolDefinition.builder()
+ .name(name)
+ .description(description)
+ .requestConfig(requestTemplate)
+ .inputSchema(inputSchema)
.build();
- // Verify that the session factory was properly set
- LOG.debug("MCP server created successfully for URI: {}, session factory should be available", path);
+ LOG.debug("Adding tool to shared server - name: {}, description: {}, path: {}", name, description, normalizedPath);
- mcpServerMap.put(path, server);
+ ShenyuToolCallback shenyuToolCallback = new ShenyuToolCallback(shenyuToolDefinition);
- // Register both exact paths and wildcard patterns for flexible matching
- routeMap.put(path, transportProvider::handleSseConnection);
- routeMap.put(messagePath, transportProvider::handleMessage);
+ // Add tool to shared server (automatically available across all protocols)
+ McpAsyncServer sharedServer = sharedServerMap.get(normalizedPath);
+ if (Objects.nonNull(sharedServer)) {
+ try {
+ for (AsyncToolSpecification asyncToolSpecification : McpToolUtils.toAsyncToolSpecifications(shenyuToolCallback)) {
+ sharedServer.addTool(asyncToolSpecification).block();
+ }
+
+ Set<String> protocols = getSupportedProtocols(normalizedPath);
+ LOG.info("Added tool '{}' to shared server for path: {} (available across protocols: {})",
+ name, normalizedPath, protocols);
+ } catch (Exception e) {
+ LOG.error("Failed to add tool '{}' to shared server for path: {}", name, normalizedPath, e);
+ }
+ } else {
+ LOG.warn("No shared server found for path: {}", normalizedPath);
+ }
+ }
- // Add wildcard patterns to support sub-paths
- routeMap.put(path + "/**", transportProvider::handleSseConnection);
- routeMap.put(messagePath + "/**", transportProvider::handleMessage);
+ /**
+ * Removes a tool from the shared server instance.
+ *
+ * @param serverPath the server path
+ * @param name the tool name
+ */
+ public void removeTool(final String serverPath, final String name) {
+ String normalizedPath = normalizeServerPath(serverPath);
+ LOG.debug("Removing tool from shared server - name: {}, path: {}", name, normalizedPath);
+
+ McpAsyncServer sharedServer = sharedServerMap.get(normalizedPath);
+ if (Objects.nonNull(sharedServer)) {
+ try {
+ sharedServer.removeTool(name).block();
+
+ Set<String> protocols = getSupportedProtocols(normalizedPath);
+ LOG.info("Removed tool '{}' from shared server for path: {} (removed from protocols: {})",
+ name, normalizedPath, protocols);
+ } catch (Exception e) {
+ LOG.error("Failed to remove tool '{}' from shared server for path: {}", name, normalizedPath, e);
+ }
+ } else {
+ LOG.warn("No shared server found for path: {}", normalizedPath);
+ }
+ }
- LOG.info("Created new McpServer for URI: {} with patterns", path);
- return transportProvider;
+ /**
+ * Get supported protocols for a server path.
+ *
+ * @param serverPath The server path
+ * @return Set of supported protocols
+ */
+ public Set<String> getSupportedProtocols(final String serverPath) {
+ String normalizedPath = normalizeServerPath(serverPath);
+ CompositeTransportProvider compositeTransport = compositeTransportMap.get(normalizedPath);
+ return Objects.nonNull(compositeTransport) ? compositeTransport.getSupportedProtocols() : new HashSet<>();
}
- public void addTool(final String serverPath, final String name, final String description,
- final String requestTemplate, final String inputSchema) {
- // remove first for overwrite
- try {
- removeTool(serverPath, name);
- } catch (Exception ignored) {
- // ignore
+ /**
+ * Normalize the server path by removing protocol-specific suffixes.
+ *
+ * @param path The original path
+ * @return The normalized path for shared server usage
+ */
+ private String normalizeServerPath(final String path) {
+ if (Objects.isNull(path)) {
+ return null;
}
- ToolDefinition shenyuToolDefinition = ShenyuToolDefinition.builder().name(name).description(description)
- .requestConfig(requestTemplate).inputSchema(inputSchema).build();
- LOG.debug("Adding tool, name: {}, description: {}, requestTemplate: {}, inputSchema: {}", name, description,
- requestTemplate, inputSchema);
- ShenyuToolCallback shenyuToolCallback = new ShenyuToolCallback(shenyuToolDefinition);
- for (AsyncToolSpecification asyncToolSpecification : McpToolUtils
- .toAsyncToolSpecifications(shenyuToolCallback)) {
- mcpServerMap.get(serverPath).addTool(asyncToolSpecification).block();
+ String normalizedPath = path;
+
+ // Remove /streamablehttp suffix
+ if (normalizedPath.endsWith("/streamablehttp")) {
+ normalizedPath = normalizedPath.substring(0, normalizedPath.length() - "/streamablehttp".length());
+ LOG.debug("Normalized Streamable HTTP path from '{}' to '{}' for shared server", path, normalizedPath);
}
+
+ return normalizedPath;
}
- public void removeTool(final String serverPath, final String name) {
- LOG.debug("Removing tool, name: {}", name);
- mcpServerMap.get(serverPath).removeTool(name).block();
+ /**
+ * Composite transport provider that delegates to multiple transport implementations.
+ * Enhanced with protocol-aware session management and improved error handling.
+ * Provides unified access to different transport protocols.
+ */
+ private static class CompositeTransportProvider implements io.modelcontextprotocol.spec.McpServerTransportProvider {
+
+ private final Map<String, Object> transports = new ConcurrentHashMap<>();
+
+ private volatile McpServerSession.Factory sessionFactory;
+
+ // Track active sessions per protocol for isolation
+ private final Map<String, Set<String>> protocolSessions = new ConcurrentHashMap<>();
+
+ public void addTransport(final String protocol, final Object transportProvider) {
+ transports.put(protocol, transportProvider);
+ protocolSessions.put(protocol, Collections.synchronizedSet(new HashSet<>()));
+
+ // Set session factory on the new transport if available
+ if (Objects.nonNull(sessionFactory) && transportProvider instanceof io.modelcontextprotocol.spec.McpServerTransportProvider) {
+ ((io.modelcontextprotocol.spec.McpServerTransportProvider) transportProvider).setSessionFactory(sessionFactory);
+ }
+
+ LOG.debug("Added transport '{}' to composite provider", protocol);
+ }
+
+ /**
+ * Gets the SSE transport provider.
+ *
+ * @return the SSE transport provider, or null if not available
+ */
+ public ShenyuSseServerTransportProvider getSseTransport() {
+ Object transport = transports.get(SSE_PROTOCOL);
+ return transport instanceof ShenyuSseServerTransportProvider
+ ? (ShenyuSseServerTransportProvider) transport : null;
+ }
+
+ /**
+ * Gets the Streamable HTTP transport provider.
+ *
+ * @return the Streamable HTTP transport provider, or null if not available
+ */
+ public ShenyuStreamableHttpServerTransportProvider getStreamableHttpTransport() {
+ Object transport = transports.get(STREAMABLE_HTTP_PROTOCOL);
+ return transport instanceof ShenyuStreamableHttpServerTransportProvider
+ ? (ShenyuStreamableHttpServerTransportProvider) transport : null;
+ }
+
+ /**
+ * Gets a transport provider by protocol name.
+ *
+ * @param protocol the protocol name
+ * @return the transport provider, or null if not found
+ */
+ public Object getTransport(final String protocol) {
+ return transports.get(protocol);
+ }
+
+ /**
+ * Checks if a specific protocol is supported.
+ *
+ * @param protocol the protocol name
+ * @return true if the protocol is supported, false otherwise
+ */
+ public boolean hasProtocol(final String protocol) {
+ return transports.containsKey(protocol);
+ }
+
+ /**
+ * Gets all supported protocols.
+ *
+ * @return a set of supported protocol names
+ */
+ public Set<String> getSupportedProtocols() {
+ return new HashSet<>(transports.keySet());
+ }
+
+ @Override
+ public void setSessionFactory(final McpServerSession.Factory sessionFactory) {
+ this.sessionFactory = sessionFactory;
+ // Set session factory on all existing transports atomically
+ synchronized (transports) {
+ for (Object transport : transports.values()) {
+ if (transport instanceof io.modelcontextprotocol.spec.McpServerTransportProvider) {
+ try {
+ ((io.modelcontextprotocol.spec.McpServerTransportProvider) transport).setSessionFactory(sessionFactory);
+ } catch (Exception e) {
+ LOG.error("Failed to set session factory on transport: {}", transport.getClass().getSimpleName(), e);
+ }
+ }
+ }
+ }
+ LOG.debug("Session factory set on composite transport with {} transports", transports.size());
+ }
+
+ @Override
+ public reactor.core.publisher.Mono<Void> notifyClients(final String method, final Object params) {
+ if (transports.isEmpty()) {
+ LOG.debug("No transports available for client notification");
+ return reactor.core.publisher.Mono.empty();
+ }
+
+ LOG.debug("Broadcasting notification '{}' to {} transports", method, transports.size());
+
+ return reactor.core.publisher.Flux.fromIterable(transports.entrySet())
+ .flatMap(entry -> {
+ String protocol = entry.getKey();
+ Object transport = entry.getValue();
+
+ if (transport instanceof io.modelcontextprotocol.spec.McpServerTransportProvider) {
+ return ((io.modelcontextprotocol.spec.McpServerTransportProvider) transport)
+ .notifyClients(method, params)
+ .doOnSuccess(aVoid -> LOG.debug("Successfully notified {} clients", protocol))
+ .doOnError(e -> LOG.warn("Failed to notify {} clients: {}", protocol, e.getMessage()))
+ // Continue with other transports even if one fails
+ .onErrorComplete();
+ } else {
+ LOG.warn("Transport '{}' does not implement McpServerTransportProvider", protocol);
+ return reactor.core.publisher.Mono.empty();
+ }
+ })
+ .then()
+ .doOnSuccess(aVoid -> LOG.debug("Client notification broadcast completed"));
+ }
+
+ @Override
+ public reactor.core.publisher.Mono<Void> closeGracefully() {
+ if (transports.isEmpty()) {
+ return reactor.core.publisher.Mono.empty();
+ }
+
+ LOG.info("Initiating graceful shutdown of {} transports", transports.size());
+
+ return reactor.core.publisher.Flux.fromIterable(transports.entrySet())
+ .flatMap(entry -> {
+ String protocol = entry.getKey();
+ Object transport = entry.getValue();
+
+ if (transport instanceof io.modelcontextprotocol.spec.McpServerTransportProvider) {
+ return ((io.modelcontextprotocol.spec.McpServerTransportProvider) transport)
+ .closeGracefully()
+ .doOnSuccess(aVoid -> LOG.info("Successfully closed {} transport", protocol))
+ .doOnError(e -> LOG.error("Error closing {} transport: {}", protocol, e.getMessage()))
+ // Don't fail entire shutdown if one transport fails
+ .onErrorComplete();
+ } else {
+ LOG.warn("Transport '{}' does not implement graceful shutdown", protocol);
+ return reactor.core.publisher.Mono.empty();
+ }
+ })
+ .then()
+ .doOnSuccess(aVoid -> {
+ // Clear all tracking data after successful shutdown
+ transports.clear();
+ protocolSessions.clear();
+ LOG.info("Graceful shutdown completed - all transports and sessions cleared");
+ });
+ }
}
+
}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/BodyWriterExchange.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/BodyWriterExchange.java
index 85f850620d..c4b4e43f46 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/BodyWriterExchange.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/BodyWriterExchange.java
@@ -17,74 +17,98 @@
package org.apache.shenyu.plugin.mcp.server.request;
+import org.apache.shenyu.plugin.base.support.BodyInserterContext;
+import org.apache.shenyu.plugin.base.support.CachedBodyOutputMessage;
+import org.apache.shenyu.plugin.base.utils.ResponseUtils;
import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.http.HttpHeaders;
+import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.web.reactive.function.BodyInserter;
+import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.ServerWebExchangeDecorator;
import reactor.core.publisher.Flux;
-import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
-import java.util.Map;
import java.util.Objects;
-import java.util.stream.Collectors;
+/**
+ * Body Writer Exchange for MCP tool requests.
+ *
+ * <p>Decorates ServerWebExchange to properly handle request body writing for different content types.
+ * Supports JSON, form-urlencoded, and multipart form data formats.</p>
+ *
+ * @since 2.7.0.2
+ */
public class BodyWriterExchange extends ServerWebExchangeDecorator {
private final String body;
private final String contentType;
+ private final CachedBodyOutputMessage cachedBodyOutputMessage;
+
public BodyWriterExchange(final ServerWebExchange delegate, final String body) {
super(delegate);
this.body = body;
HttpHeaders headers = delegate.getRequest().getHeaders();
this.contentType = headers.getFirst(HttpHeaders.CONTENT_TYPE);
+
+ this.cachedBodyOutputMessage = ResponseUtils.newCachedBodyOutputMessage(delegate);
+
+ initializeBody();
}
- @Override
- public ServerHttpRequest getRequest() {
- return new ServerHttpRequestDecorator(super.getRequest()) {
- @Override
- public Flux<DataBuffer> getBody() {
- DataBufferFactory bufferFactory = BodyWriterExchange.this.getResponse().bufferFactory();
- byte[] bytes = getBodyBytes();
- DataBuffer buffer = bufferFactory.wrap(bytes);
- return Flux.just(buffer);
- }
- };
+ private void initializeBody() {
+ String bodyContent = getFormattedBody();
+ BodyInserter<String, ReactiveHttpOutputMessage> bodyInserter = BodyInserters.fromValue(bodyContent);
+
+ bodyInserter.insert(cachedBodyOutputMessage, new BodyInserterContext()).subscribe();
}
- private byte[] getBodyBytes() {
+ private String getFormattedBody() {
if (Objects.isNull(contentType)) {
- return body.getBytes(StandardCharsets.UTF_8);
+ return body;
}
if (contentType.contains("application/json")) {
- return body.getBytes(StandardCharsets.UTF_8);
+ return body;
} else if (contentType.contains("application/x-www-form-urlencoded")) {
- // body must be in key1=val1&key2=val2 format
- return body.getBytes(StandardCharsets.UTF_8);
+ return body;
} else if (contentType.contains("multipart/form-data")) {
- // Simple handling here, just return the string. For complex scenarios, use
- // MultipartBodyBuilder.
- return body.getBytes(StandardCharsets.UTF_8);
+ return body;
} else {
- return body.getBytes(StandardCharsets.UTF_8);
+ return body;
}
}
- // Optional: Helper method to convert Map<String, Object> to
- // x-www-form-urlencoded string
- public static String toFormUrlEncoded(final Map<String, Object> map) {
- return map.entrySet().stream()
- .map(e -> e.getKey() + "=" + urlEncode(e.getValue().toString()))
- .collect(Collectors.joining("&"));
- }
+ /**
+ * Returns a decorated ServerHttpRequest with the correct body and headers for downstream processing.
+ *
+ * @return the decorated ServerHttpRequest
+ */
+ @Override
+ public ServerHttpRequest getRequest() {
+ return new ServerHttpRequestDecorator(super.getRequest()) {
+ @Override
+ public Flux<DataBuffer> getBody() {
+ return cachedBodyOutputMessage.getBody();
+ }
+
+ @Override
+ public HttpHeaders getHeaders() {
+ HttpHeaders originalHeaders = super.getHeaders();
+ HttpHeaders newHeaders = new HttpHeaders();
+ newHeaders.putAll(originalHeaders);
- private static String urlEncode(final String value) {
- return URLEncoder.encode(value, StandardCharsets.UTF_8);
+ if (Objects.nonNull(body)) {
+ byte[] bodyBytes = getFormattedBody().getBytes(StandardCharsets.UTF_8);
+ newHeaders.setContentLength(bodyBytes.length);
+ }
+
+ return newHeaders;
+ }
+ };
}
}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/ParameterFormatter.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/ParameterFormatter.java
new file mode 100644
index 0000000000..41dacf82c4
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/ParameterFormatter.java
@@ -0,0 +1,60 @@
+/*
+ * 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.mcp.server.request;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonPrimitive;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.StringUtils;
+
+import java.util.Objects;
+
+/**
+ * Simple parameter formatting utilities.
+ */
+public class ParameterFormatter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ParameterFormatter.class);
+
+ /**
+ * Try to parse a JSON string into JsonElement.
+ * If parsing fails, returns the original string as JsonPrimitive.
+ *
+ * @param jsonString the JSON string to parse
+ * @return the parsed JsonElement or original string
+ */
+ public static JsonElement tryParseJsonString(final String jsonString) {
+ if (!StringUtils.hasText(jsonString)) {
+ return new JsonPrimitive(Objects.isNull(jsonString) ? "" : jsonString);
+ }
+
+ String trimmed = jsonString.trim();
+ if (trimmed.startsWith("[") && trimmed.endsWith("]")
+ || trimmed.startsWith("{") && trimmed.endsWith("}")) {
+ try {
+ return GsonUtils.getInstance().fromJson(jsonString, JsonElement.class);
+ } catch (Exception e) {
+ LOG.debug("Failed to parse JSON string, keeping as string: {}", e.getMessage());
+ }
+ }
+
+ return new JsonPrimitive(jsonString);
+ }
+}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/RequestConfigHelper.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/RequestConfigHelper.java
index 5e6e2fe92c..4f076e2d51 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/RequestConfigHelper.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/request/RequestConfigHelper.java
@@ -24,15 +24,9 @@ import org.apache.shenyu.common.utils.GsonUtils;
* Helper class for parsing and handling requestConfig.
*/
public class RequestConfigHelper {
-
- public static final String METHOD_POST = "POST";
-
- public static final String METHOD_PUT = "PUT";
-
- public static final String METHOD_PATCH = "PATCH";
-
+
private final JsonObject configJson;
-
+
/**
* Constructor.
*
@@ -41,16 +35,7 @@ public class RequestConfigHelper {
public RequestConfigHelper(final String requestConfig) {
this.configJson = GsonUtils.getInstance().fromJson(requestConfig, JsonObject.class);
}
-
- /**
- * Get the config json object.
- *
- * @return the config json object
- */
- public JsonObject getConfigJson() {
- return configJson;
- }
-
+
/**
* Get the request template json object.
*
@@ -59,7 +44,7 @@ public class RequestConfigHelper {
public JsonObject getRequestTemplate() {
return configJson.getAsJsonObject("requestTemplate");
}
-
+
/**
* Get the argument position mapping.
*
@@ -68,7 +53,7 @@ public class RequestConfigHelper {
public JsonObject getArgsPosition() {
return configJson.has("argsPosition") ? configJson.getAsJsonObject("argsPosition") : new JsonObject();
}
-
+
/**
* Get the response template json object.
*
@@ -77,7 +62,7 @@ public class RequestConfigHelper {
public JsonObject getResponseTemplate() {
return configJson.has("responseTemplate") ? configJson.getAsJsonObject("responseTemplate") : null;
}
-
+
/**
* Get the url template string.
*
@@ -86,7 +71,7 @@ public class RequestConfigHelper {
public String getUrlTemplate() {
return getRequestTemplate().get("url").getAsString();
}
-
+
/**
* Get the HTTP method string.
*
@@ -96,17 +81,12 @@ public class RequestConfigHelper {
JsonObject requestTemplate = getRequestTemplate();
return requestTemplate.has("method") ? requestTemplate.get("method").getAsString() : "GET";
}
-
+
public boolean isArgsToJsonBody() {
JsonObject requestTemplate = getRequestTemplate();
return requestTemplate.has("argsToJsonBody") && requestTemplate.get("argsToJsonBody").getAsBoolean();
}
-
- public boolean isArgsToUrlParam() {
- JsonObject requestTemplate = getRequestTemplate();
- return requestTemplate.has("argsToUrlParam") && requestTemplate.get("argsToUrlParam").getAsBoolean();
- }
-
+
/**
* Build the request path based on the URL template and argument positions.
*
@@ -118,37 +98,98 @@ public class RequestConfigHelper {
public static String buildPath(final String urlTemplate, final JsonObject argsPosition,
final JsonObject inputJson) {
// First, check if the input value is already a complete URL
- for (String key : argsPosition.keySet()) {
- if (inputJson.has(key)) {
- String value = inputJson.get(key).getAsString();
- // If the input value is already a complete URL, return it directly
- if (value.startsWith("http://") || value.startsWith("https://")) {
- return value;
- }
- }
+ if (isCompleteUrl(argsPosition, inputJson)) {
+ return getCompleteUrl(argsPosition, inputJson);
}
-
+
StringBuilder queryBuilder = new StringBuilder();
-
- // Check if the URL template already contains query parameters
+
+ // Check whether the URL template already contains the query parameters
boolean hasExistingQuery = urlTemplate.contains("?");
String basePath = hasExistingQuery ? urlTemplate.substring(0, urlTemplate.indexOf("?")) : urlTemplate;
String existingQuery = hasExistingQuery ? urlTemplate.substring(urlTemplate.indexOf("?") + 1) : "";
-
- // handle new query parameters
+
+ // Handle new query parameters
+ basePath = processArguments(argsPosition, inputJson, basePath, queryBuilder);
+
+ // Clear the template variables that have not been replaced
+ basePath = basePath.replaceAll("\\{\\{\\.[^}]+}}", "");
+
+ // Build the final URL
+ return buildFinalPath(basePath, queryBuilder, hasExistingQuery, existingQuery);
+ }
+
+ /**
+ * Check if input contains a complete URL.
+ *
+ * @param argsPosition the argument position mapping
+ * @param inputJson the input JSON object
+ * @return true if input contains a complete URL
+ */
+ private static boolean isCompleteUrl(final JsonObject argsPosition, final JsonObject inputJson) {
+ for (String key : argsPosition.keySet()) {
+ if (inputJson.has(key)) {
+ try {
+ String value = inputJson.get(key).getAsString();
+ if (value.startsWith("http://") || value.startsWith("https://") || value.contains("?")) {
+ return true;
+ }
+ } catch (Exception exception) {
+ // Ignore exception
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Get the complete URL from input.
+ *
+ * @param argsPosition the argument position mapping
+ * @param inputJson the input JSON object
+ * @return the complete URL
+ */
+ private static String getCompleteUrl(final JsonObject argsPosition, final JsonObject inputJson) {
+ for (String key : argsPosition.keySet()) {
+ if (inputJson.has(key)) {
+ try {
+ String value = inputJson.get(key).getAsString();
+ if (value.startsWith("http://") || value.startsWith("https://") || value.contains("?")) {
+ return value;
+ }
+ } catch (Exception exception) {
+ // Ignore exception
+ }
+ }
+ }
+ return "";
+ }
+
+ /**
+ * Process arguments for path and query parameters.
+ *
+ * @param argsPosition the argument position mapping
+ * @param inputJson the input JSON object
+ * @param basePath the base path to modify
+ * @param queryBuilder the query builder to append to
+ * @return the modified base path
+ */
+ private static String processArguments(final JsonObject argsPosition, final JsonObject inputJson,
+ final String basePath, final StringBuilder queryBuilder) {
+ String modifiedBasePath = basePath;
for (String key : argsPosition.keySet()) {
String position = argsPosition.get(key).getAsString();
if ("path".equals(position) && inputJson.has(key)) {
- // 处理路径参数
+ // Process path parameters
String value = inputJson.get(key).getAsString();
if (value.contains("?")) {
value = value.substring(0, value.indexOf("?"));
}
value = value.replace("\"", "").trim();
- basePath = basePath.replace("{{." + key + "}}", value);
+ modifiedBasePath = modifiedBasePath.replace("{{." + key + "}}", value);
} else if ("query".equals(position) && inputJson.has(key)) {
- // handle query parameters
- if (!existingQuery.contains(key + "=")) {
+ // Handle query parameters
+ if (!modifiedBasePath.contains(key + "=")) {
if (!queryBuilder.isEmpty()) {
queryBuilder.append("&");
}
@@ -161,14 +202,23 @@ public class RequestConfigHelper {
}
}
}
-
- // clean up the base path by removing any placeholders that were not replaced
- basePath = basePath.replaceAll("\\{\\{\\.[^}]+}}", "");
-
- // build the final path
+ return modifiedBasePath;
+ }
+
+ /**
+ * Build the final path with query parameters.
+ *
+ * @param basePath the base path
+ * @param queryBuilder the query builder
+ * @param hasExistingQuery whether there are existing query parameters
+ * @param existingQuery the existing query string
+ * @return the final path
+ */
+ private static String buildFinalPath(final String basePath, final StringBuilder queryBuilder,
+ final boolean hasExistingQuery, final String existingQuery) {
StringBuilder finalPath = new StringBuilder(basePath);
-
- // add new query parameters if any
+
+ // Add query parameters
if (!queryBuilder.isEmpty()) {
if (hasExistingQuery) {
finalPath.append("&").append(queryBuilder);
@@ -176,8 +226,8 @@ public class RequestConfigHelper {
finalPath.append("?").append(queryBuilder);
}
}
-
- // add existing query parameters if any
+
+ // Add existing query parameters
if (hasExistingQuery && !existingQuery.isEmpty()) {
if (!queryBuilder.isEmpty()) {
finalPath.append("&").append(existingQuery);
@@ -185,42 +235,14 @@ public class RequestConfigHelper {
finalPath.append("?").append(existingQuery);
}
}
-
+
String result = finalPath.toString();
- // remove trailing question mark if it exists
+ // Remove the question mark at the end
if (result.endsWith("?")) {
result = result.substring(0, result.length() - 1);
}
-
+
return result;
}
-
- /**
- * Build the request body JSON object based on the argument positions and input
- * JSON.
- *
- * @param argsToJsonBody whether to convert arguments to JSON body
- * @param argsPosition the argument position mapping
- * @param inputJson the input JSON object
- * @return the constructed body JSON object
- */
- public static JsonObject buildBodyJson(final boolean argsToJsonBody, final JsonObject argsPosition,
- final JsonObject inputJson) {
- JsonObject bodyJson = new JsonObject();
- if (argsToJsonBody) {
- for (String key : argsPosition.keySet()) {
- String position = argsPosition.get(key).getAsString();
- if ("body".equals(position) && inputJson.has(key)) {
- bodyJson.add(key, inputJson.get(key));
- }
- }
- }
- return bodyJson;
- }
-
- public static boolean isRequestBodyMethod(final String method) {
- return METHOD_POST.equalsIgnoreCase(method)
- || METHOD_PUT.equalsIgnoreCase(method)
- || METHOD_PATCH.equalsIgnoreCase(method);
- }
+
}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
new file mode 100644
index 0000000000..5e76a08fd9
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
@@ -0,0 +1,276 @@
+/*
+ * 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.mcp.server.response;
+
+import com.google.gson.JsonObject;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.Objects;
+
+/**
+ * Non-committing MCP response decorator for Streamable HTTP protocol.
+ * Intercepts response data without writing to the original response, avoiding premature
+ * response commitment issues. Used for capturing response data for processing and transformation
+ * before final output.
+ *
+ * @see org.apache.shenyu.plugin.mcp.server.response.ShenyuMcpResponseDecorator
+ * @since 2.7.0.2
+ */
+public class NonCommittingMcpResponseDecorator extends ServerHttpResponseDecorator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NonCommittingMcpResponseDecorator.class);
+
+ /**
+ * Template placeholder pattern for variable substitution.
+ */
+ private static final String PLACEHOLDER_PREFIX = "${";
+
+ private static final String PLACEHOLDER_SUFFIX = "}";
+
+ private final String sessionId;
+
+ private final CompletableFuture<String> responseFuture;
+
+ private final JsonObject responseTemplate;
+
+ /**
+ * Constructs a new non-committing MCP response decorator.
+ *
+ * @param delegate the underlying HTTP response delegate
+ * @param sessionId the MCP session identifier for correlation
+ * @param responseFuture the future to complete with processed response data
+ * @param responseTemplate the JSON template for response transformation (nullable)
+ */
+ public NonCommittingMcpResponseDecorator(final ServerHttpResponse delegate,
+ final String sessionId,
+ final CompletableFuture<String> responseFuture,
+ final JsonObject responseTemplate) {
+ super(delegate);
+ this.sessionId = sessionId;
+ this.responseFuture = responseFuture;
+ this.responseTemplate = responseTemplate;
+ LOG.debug("Created non-committing MCP response decorator for session: {}", sessionId);
+ }
+
+ @Override
+ public Mono<Void> writeWith(final Publisher<? extends DataBuffer> body) {
+ LOG.debug("Processing writeWith for session: {}", sessionId);
+
+ return Flux.from(body)
+ .collectList()
+ .doOnNext(this::processResponseData)
+ .then()
+ .doOnSuccess(aVoid -> LOG.debug("Successfully completed writeWith for session: {}", sessionId))
+ .doOnError(error -> handleProcessingError("writeWith", error));
+ }
+
+ @Override
+ public Mono<Void> writeAndFlushWith(final Publisher<? extends Publisher<? extends DataBuffer>> body) {
+ LOG.debug("Processing writeAndFlushWith for session: {}", sessionId);
+
+ return Flux.from(body)
+ .flatMap(Flux::from)
+ .collectList()
+ .doOnNext(this::processResponseData)
+ .then()
+ .doOnSuccess(aVoid -> LOG.debug("Successfully completed writeAndFlushWith for session: {}", sessionId))
+ .doOnError(error -> handleProcessingError("writeAndFlushWith", error));
+ }
+
+ /**
+ * Processes the collected response data buffers and completes the response future.
+ *
+ * <p>Aggregates all data buffers into a single response string, applies response template
+ * transformations if configured, and completes the response future with the processed result.
+ *
+ * @param dataBuffers the collected response data buffers
+ */
+ private void processResponseData(final java.util.List<? extends DataBuffer> dataBuffers) {
+ try {
+ final String responseBody = aggregateDataBuffers(dataBuffers);
+ LOG.debug("Captured response data for session {}, length: {} chars", sessionId, responseBody.length());
+
+ final String processedResponse = processResponse(responseBody);
+ LOG.debug("Processed response for session {}, final length: {} chars", sessionId, processedResponse.length());
+
+ // Complete the future with processed response without writing to original response
+ responseFuture.complete(processedResponse);
+
+ } catch (Exception e) {
+ LOG.error("Error processing response data for session {}: {}", sessionId, e.getMessage(), e);
+ responseFuture.completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Aggregates multiple data buffers into a single response string.
+ *
+ * @param dataBuffers the list of data buffers to aggregate
+ * @return the aggregated response string
+ */
+ private String aggregateDataBuffers(final java.util.List<? extends DataBuffer> dataBuffers) {
+ final StringBuilder responseBuilder = new StringBuilder();
+
+ for (DataBuffer buffer : dataBuffers) {
+ final byte[] bytes = new byte[buffer.readableByteCount()];
+ buffer.read(bytes);
+ responseBuilder.append(new String(bytes, StandardCharsets.UTF_8));
+ }
+
+ return responseBuilder.toString();
+ }
+
+ /**
+ * Handles processing errors by logging and completing the future exceptionally.
+ *
+ * @param operation the operation that failed
+ * @param error the error that occurred
+ */
+ private void handleProcessingError(final String operation, final Throwable error) {
+ LOG.error("Error in {} operation for session {}: {}", operation, sessionId, error.getMessage(), error);
+ responseFuture.completeExceptionally(error);
+ }
+
+ /**
+ * Processes response data by applying response template transformations.
+ * If no response template is configured, returns the original response body unchanged.
+ * If a template is provided, attempts to parse the response as JSON and apply template
+ * transformations including placeholder substitution.
+ *
+ * @param responseBody the original response body to process
+ * @return the processed response body
+ */
+ private String processResponse(final String responseBody) {
+ try {
+ if (Objects.isNull(responseTemplate) || responseTemplate.size() == 0) {
+ return responseBody;
+ }
+
+ // Attempt to parse response as JSON for template processing
+ final JsonObject responseJson = parseResponseAsJson(responseBody);
+ if (Objects.isNull(responseJson)) {
+ LOG.debug("Response is not valid JSON for session: {}, returning unchanged", sessionId);
+ return responseBody;
+ }
+
+ // Apply response template transformations
+ return applyResponseTemplate(responseJson);
+
+ } catch (Exception e) {
+ LOG.error("Error applying response template for session {}: {}", sessionId, e.getMessage());
+ return responseBody;
+ }
+ }
+
+ /**
+ * Attempts to parse the response body as JSON.
+ *
+ * @param responseBody the response body to parse
+ * @return the parsed JSON object, or null if parsing fails
+ */
+ private JsonObject parseResponseAsJson(final String responseBody) {
+ try {
+ return GsonUtils.getInstance().fromJson(responseBody, JsonObject.class);
+ } catch (Exception e) {
+ LOG.debug("Failed to parse response as JSON for session: {}", sessionId);
+ return null;
+ }
+ }
+
+ /**
+ * Applies response template transformations to the parsed JSON response.
+ *
+ * @param responseJson the parsed JSON response data
+ * @return the transformed response as JSON string
+ */
+ private String applyResponseTemplate(final JsonObject responseJson) {
+ final JsonObject processedResponse = new JsonObject();
+
+ if (responseTemplate.has("content")) {
+ final JsonObject contentTemplate = responseTemplate.getAsJsonObject("content");
+ final JsonObject content = buildContentFromTemplate(contentTemplate, responseJson);
+ processedResponse.add("content", content);
+ } else {
+ // If no specific template structure, return original response
+ return GsonUtils.getInstance().toJson(responseJson);
+ }
+
+ return GsonUtils.getInstance().toJson(processedResponse);
+ }
+
+ /**
+ * Builds content object from template and response data.
+ *
+ * @param contentTemplate the content template configuration
+ * @param responseData the response data for placeholder substitution
+ * @return the built content JSON object
+ */
+ private JsonObject buildContentFromTemplate(final JsonObject contentTemplate, final JsonObject responseData) {
+ final JsonObject content = new JsonObject();
+
+ if (contentTemplate.has("type")) {
+ content.addProperty("type", contentTemplate.get("type").getAsString());
+ }
+
+ if (contentTemplate.has("text")) {
+ final String textTemplate = contentTemplate.get("text").getAsString();
+ final String processedText = applyPlaceholderSubstitution(textTemplate, responseData);
+ content.addProperty("text", processedText);
+ }
+
+ return content;
+ }
+
+ /**
+ * Applies placeholder substitution to a text template using response data.
+ * Supports simple placeholder format: ${fieldName}
+ *
+ * @param template the text template containing placeholders
+ * @param responseData the JSON object containing substitution values
+ * @return the text with placeholders replaced by actual values
+ */
+ private String applyPlaceholderSubstitution(final String template, final JsonObject responseData) {
+ if (Objects.isNull(responseData)) {
+ return template;
+ }
+
+ String result = template;
+
+ for (String key : responseData.keySet()) {
+ final String placeholder = PLACEHOLDER_PREFIX + key + PLACEHOLDER_SUFFIX;
+ if (result.contains(placeholder)) {
+ final String value = responseData.get(key).isJsonPrimitive()
+ ? responseData.get(key).getAsString()
+ : responseData.get(key).toString();
+ result = result.replace(placeholder, value);
+ }
+ }
+
+ return result;
+ }
+}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/MessageHandlingResult.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/MessageHandlingResult.java
new file mode 100644
index 0000000000..be971c7a43
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/MessageHandlingResult.java
@@ -0,0 +1,99 @@
+/*
+ * 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.mcp.server.transport;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Result object for message handling operations.
+ * This class encapsulates the result of processing a Streamable HTTP message,
+ * including the HTTP status code, response body, and session correlation information.
+ */
+public class MessageHandlingResult {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MessageHandlingResult.class);
+
+ private final int statusCode;
+
+ private final Object responseBody;
+
+ private final String sessionId;
+
+ /**
+ * Creates a new message handling result.
+ *
+ * @param statusCode the HTTP status code for the response
+ * @param responseBody the response body object
+ * @param sessionId the session identifier for correlation (nullable)
+ */
+ public MessageHandlingResult(final int statusCode, final Object responseBody, final String sessionId) {
+ this.statusCode = statusCode;
+ this.responseBody = responseBody;
+ this.sessionId = sessionId;
+ }
+
+ /**
+ * Gets the HTTP status code for this result.
+ *
+ * @return the status code
+ */
+ public int getStatusCode() {
+ return statusCode;
+ }
+
+ /**
+ * Gets the response body object.
+ *
+ * @return the response body
+ */
+ public Object getResponseBody() {
+ return responseBody;
+ }
+
+ /**
+ * Gets the session identifier associated with this result.
+ *
+ * @return the session ID, or null if not available
+ */
+ public String getSessionId() {
+ return sessionId;
+ }
+
+ /**
+ * Converts the response body to a JSON string for HTTP response transmission.
+ * If the response body is already a string, returns it as-is. Otherwise,
+ * attempts to serialize it to JSON using ObjectMapper. On serialization failure,
+ * returns a standard JSON-RPC error response.
+ *
+ * @return the response body as JSON string
+ */
+ public String getResponseBodyAsJson() {
+ if (responseBody instanceof String) {
+ return (String) responseBody;
+ }
+
+ try {
+ return new ObjectMapper().writeValueAsString(responseBody);
+ } catch (Exception e) {
+ LOGGER.error("Failed to serialize response body to JSON: {}", e.getMessage());
+ return "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Internal error\"}}";
+ }
+ }
+}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/ShenyuStreamableHttpServerTransportProvider.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/ShenyuStreamableHttpServerTransportProvider.java
new file mode 100644
index 0000000000..fd31765b66
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/ShenyuStreamableHttpServerTransportProvider.java
@@ -0,0 +1,996 @@
+/*
+ * 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.mcp.server.transport;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.spec.McpError;
+import io.modelcontextprotocol.spec.McpSchema;
+import io.modelcontextprotocol.spec.McpServerSession;
+import io.modelcontextprotocol.spec.McpServerTransport;
+import io.modelcontextprotocol.spec.McpServerTransportProvider;
+import io.modelcontextprotocol.util.Assert;
+import org.apache.shenyu.plugin.mcp.server.holder.ShenyuMcpExchangeHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.Exceptions;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Streamable HTTP Server Transport Provider for MCP (Model Context Protocol).
+ * Implements the Streamable HTTP MCP transport protocol with unified endpoint for all MCP operations.
+ * Provides advanced session management with automatic recovery and reconnection capabilities.
+ *
+ * @see McpServerTransportProvider
+ * @see McpServerSession
+ * @see org.apache.shenyu.plugin.mcp.server.holder.ShenyuMcpExchangeHolder
+ * @since 2.7.0.2
+ */
+public class ShenyuStreamableHttpServerTransportProvider implements McpServerTransportProvider {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ShenyuStreamableHttpServerTransportProvider.class);
+
+ /**
+ * Header name for MCP session identification.
+ */
+ private static final String SESSION_ID_HEADER = "Mcp-Session-Id";
+
+ /**
+ * JSON-RPC 2.0 version identifier.
+ */
+ private static final String JSONRPC_VERSION = "2.0";
+
+ /**
+ * MCP initialize method name.
+ */
+ private static final String INITIALIZE_METHOD = "initialize";
+
+ /**
+ * Default MCP protocol version.
+ */
+ private static final String DEFAULT_PROTOCOL_VERSION = "2025-03-26";
+
+ /**
+ * Server information constants.
+ */
+ private static final String SERVER_NAME = "ShenyuMcpServer";
+
+ private static final String SERVER_VERSION = "1.0.0";
+
+ private final ObjectMapper objectMapper;
+
+ private McpServerSession.Factory sessionFactory;
+
+ /**
+ * Map of active client sessions, keyed by session ID.
+ * This enables session reuse for subsequent requests with the same session ID.
+ */
+ private final ConcurrentHashMap<String, McpServerSession> sessions = new ConcurrentHashMap<>();
+
+ /**
+ * Map of session transports, keyed by session ID.
+ * This enables access to transport-specific functionality and response correlation.
+ */
+ private final ConcurrentHashMap<String, StreamableHttpSessionTransport> sessionTransports = new ConcurrentHashMap<>();
+
+ /**
+ * Flag indicating if the transport provider is shutting down.
+ */
+ private volatile boolean isClosing;
+
+ /**
+ * Constructs a new Streamable HTTP server transport provider instance.
+ *
+ * @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
+ * @param endpoint The endpoint path for the Streamable HTTP MCP transport
+ * @throws IllegalArgumentException if objectMapper or endpoint is null
+ */
+ public ShenyuStreamableHttpServerTransportProvider(final ObjectMapper objectMapper, final String endpoint) {
+ Assert.notNull(objectMapper, "ObjectMapper must not be null");
+ Assert.notNull(endpoint, "Endpoint must not be null");
+ this.objectMapper = objectMapper;
+ LOGGER.debug("Created Streamable HTTP transport provider for endpoint: {}", endpoint);
+ }
+
+ /**
+ * Creates a new builder instance for constructing transport providers.
+ *
+ * @return a new Builder instance
+ */
+ public static StreamableHttpProviderBuilder builder() {
+ return new StreamableHttpProviderBuilder();
+ }
+
+ @Override
+ public void setSessionFactory(final McpServerSession.Factory sessionFactory) {
+ this.sessionFactory = sessionFactory;
+ LOGGER.debug("Session factory configured for Streamable HTTP transport");
+ }
+
+ @Override
+ public Mono<Void> notifyClients(final String method, final Object params) {
+ if (sessions.isEmpty()) {
+ LOGGER.debug("No active sessions available for client notification");
+ return Mono.empty();
+ }
+ LOGGER.debug("Broadcasting notification '{}' to {} active sessions", method, sessions.size());
+ return Flux.fromIterable(sessions.values())
+ .flatMap(session -> session.sendNotification(method, params)
+ .doOnError(e -> LOGGER.warn("Failed to send notification to session {}: {}",
+ session.getId(), e.getMessage()))
+ .onErrorComplete())
+ .then()
+ .doOnSuccess(aVoid -> LOGGER.debug("Client notification broadcast completed"));
+ }
+
+ @Override
+ public Mono<Void> closeGracefully() {
+ isClosing = true;
+ if (sessions.isEmpty()) {
+ LOGGER.debug("No active sessions to close during graceful shutdown");
+ return Mono.empty();
+ }
+ LOGGER.debug("Initiating graceful shutdown of {} active sessions", sessions.size());
+ return Flux.fromIterable(sessions.values())
+ .flatMap(McpServerSession::closeGracefully)
+ .doOnComplete(() -> {
+ sessions.clear();
+ sessionTransports.clear();
+ LOGGER.debug("Graceful shutdown completed - all sessions and transports cleared");
+ })
+ .then();
+ }
+
+ /**
+ * Unified endpoint for Streamable HTTP protocol. Handles GET (stream) and POST (message).
+ *
+ * @param request the server request
+ * @return a Mono containing the server response
+ */
+ public Mono<ServerResponse> handleUnifiedEndpoint(final ServerRequest request) {
+ if (isClosing) {
+ return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down");
+ }
+ if (Objects.isNull(sessionFactory)) {
+ LOGGER.error("SessionFactory is null - MCP server not properly initialized");
+ return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).bodyValue("MCP server not properly initialized");
+ }
+ if ("OPTIONS".equalsIgnoreCase(request.methodName())) {
+ // Handle CORS preflight requests
+ return ServerResponse.ok()
+ .header("Access-Control-Allow-Origin", "*")
+ .header("Access-Control-Allow-Headers", "Content-Type, Mcp-Session-Id, Authorization")
+ .header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+ .header("Access-Control-Max-Age", "3600")
+ .build();
+ } else if ("GET".equalsIgnoreCase(request.methodName())) {
+ // Streamable HTTP protocol does not support GET requests, return 405 error
+ return ServerResponse.status(HttpStatus.METHOD_NOT_ALLOWED)
+ .header("Access-Control-Allow-Origin", "*")
+ .header("Access-Control-Allow-Headers", "Content-Type, Mcp-Session-Id, Authorization")
+ .header("Access-Control-Allow-Methods", "POST, OPTIONS")
+ .header("Allow", "POST, OPTIONS")
+ .contentType(MediaType.APPLICATION_JSON)
+ .bodyValue(new java.util.HashMap<String, Object>() {{
+ put("error", new java.util.HashMap<String, Object>() {{
+ put("code", -32601);
+ put("message", "Streamable HTTP does not support GET requests. Please use POST requests for all MCP operations.");
+ }});
+ }});
+ } else if ("POST".equalsIgnoreCase(request.methodName())) {
+ // Extract ServerWebExchange from ServerRequest
+ final ServerWebExchange exchange = request.exchange();
+ return handleMessageEndpoint(exchange, request).flatMap(result -> {
+ ServerResponse.BodyBuilder builder = ServerResponse.status(HttpStatus.valueOf(result.getStatusCode()))
+ .header("Access-Control-Allow-Origin", "*")
+ .header("Access-Control-Allow-Headers", "Content-Type, Mcp-Session-Id, Authorization")
+ .header("Access-Control-Allow-Methods", "GET, POST, OPTIONS");
+ if (Objects.nonNull(result.getSessionId())) {
+ builder.header(SESSION_ID_HEADER, result.getSessionId());
+ }
+ builder.contentType(MediaType.APPLICATION_JSON);
+ return builder.bodyValue(result.getResponseBodyAsJson());
+ });
+ }
+ return ServerResponse.badRequest().bodyValue(new McpError("Unsupported HTTP method"));
+ }
+
+ /**
+ * Handles POST requests for message processing in Streamable HTTP protocol.
+ * This method processes all MCP operations including initialize, tools/list, tools/call,
+ * and other custom operations. Enhanced to support requests without sessionId (creates
+ * temporary sessions) and invalid sessionId (creates new session and re-stores it).
+ * The method distinguishes between initialize requests (which create new sessions)
+ * and regular requests (which require existing session correlation or temporary session creation).
+ *
+ * @param exchange the server web exchange
+ * @param request the server request containing the message
+ * @return a Mono containing the message handling result
+ */
+ public Mono<MessageHandlingResult> handleMessageEndpoint(final ServerWebExchange exchange, final ServerRequest request) {
+ LOGGER.debug("Processing Streamable HTTP message for path: {}", request.path());
+ return request.bodyToMono(String.class)
+ .flatMap(body -> {
+ LOGGER.debug("Received request body with length: {} chars", body.length());
+ try {
+ // Deserialize JSON-RPC request
+ final McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper, body);
+ LOGGER.debug("Parsed JSON-RPC message of type: {}", message.getClass().getSimpleName());
+ // Handle initialize requests specially
+ if (isInitializeRequest(message)) {
+ return handleInitializeRequest(exchange, message);
+ }
+ // Handle regular requests with session management enhancement
+ return handleRegularRequestWithEnhancement(exchange, message, request);
+ } catch (IOException e) {
+ LOGGER.warn("Failed to parse JSON-RPC message: {}", e.getMessage());
+ final Object errorResponse = createJsonRpcError(null, -32700, "Parse error: Invalid JSON-RPC message");
+ return Mono.just(new MessageHandlingResult(400, errorResponse, null));
+ } catch (Exception e) {
+ LOGGER.error("Unexpected error handling message: {}", e.getMessage(), e);
+ final Object errorResponse = createJsonRpcError(null, -32603, "Internal error: " + e.getMessage());
+ return Mono.just(new MessageHandlingResult(500, errorResponse, null));
+ }
+ });
+ }
+
+ /**
+ * Checks if the given message is an initialize request.
+ *
+ * @param message the JSON-RPC message to check
+ * @return true if this is an initialize request
+ */
+ private boolean isInitializeRequest(final McpSchema.JSONRPCMessage message) {
+ if (message instanceof io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) {
+ final String method = ((io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) message).method();
+ return INITIALIZE_METHOD.equals(method);
+ }
+ return false;
+ }
+
+ /**
+ * Handles initialize requests by creating new sessions and transport connections.
+ *
+ * @param exchange the server web exchange
+ * @param message the initialize request message
+ * @return a Mono containing the initialization result
+ */
+ private Mono<MessageHandlingResult> handleInitializeRequest(final ServerWebExchange exchange,
+ final McpSchema.JSONRPCMessage message) {
+ try {
+ // Create new session and transport
+ final StreamableHttpSessionTransport transport = new StreamableHttpSessionTransport();
+ final McpServerSession session = sessionFactory.create(transport);
+ final String newSessionId = session.getId();
+ LOGGER.debug("Created new MCP session: {}", newSessionId);
+ // Store session and transport for reuse
+ sessions.put(newSessionId, session);
+ sessionTransports.put(newSessionId, transport);
+ // Configure exchange attributes
+ configureExchangeForSession(exchange, newSessionId);
+ // Extract and validate protocol version
+ final Object messageId = extractMessageId(message);
+ final String clientProtocolVersion = extractProtocolVersionFromInitialize(message);
+ if (!isSupportedProtocolVersion(clientProtocolVersion)) {
+ LOGGER.warn("Unsupported protocol version requested: {}", clientProtocolVersion);
+ cleanupInvalidSession(newSessionId);
+ final Object errorResponse = createJsonRpcError(messageId, -32600,
+ "Unsupported protocol version. Supported versions: ['" + DEFAULT_PROTOCOL_VERSION + "']");
+ return Mono.just(new MessageHandlingResult(400, errorResponse, null));
+ }
+ // Create initialize response
+ final Object initializeResponse = createInitializeResponse(messageId, clientProtocolVersion, newSessionId);
+ LOGGER.debug("Initialize request processed successfully for session: {}", newSessionId);
+ return Mono.just(new MessageHandlingResult(200, initializeResponse, newSessionId));
+ } catch (Exception e) {
+ LOGGER.error("Error handling initialize request: {}", e.getMessage(), e);
+ final Object errorResponse = createJsonRpcError(extractMessageId(message), -32603,
+ "Internal error during initialization: " + e.getMessage());
+ return Mono.just(new MessageHandlingResult(500, errorResponse, null));
+ }
+ }
+
+ /**
+ * Handles regular (non-initialize) requests with comprehensive session management.
+ * Implements session management strategy for missing session ID (creates temporary session),
+ * invalid session ID (creates new session), and valid session ID (processes normally).
+ * Provides automatic session recovery and ServerWebExchange correlation for tool callbacks.
+ *
+ * @param exchange the server web exchange
+ * @param message the JSON-RPC message
+ * @param request the server request
+ * @return a Mono containing the processing result
+ */
+ private Mono<MessageHandlingResult> handleRegularRequestWithEnhancement(final ServerWebExchange exchange,
+ final McpSchema.JSONRPCMessage message,
+ final ServerRequest request) {
+ final String requestedSessionId = extractSessionId(request);
+ final Object messageId = extractMessageId(message);
+ if (Objects.isNull(requestedSessionId)) {
+ LOGGER.info("No sessionId provided, creating temporary session for request");
+ return createTemporarySessionAndProcess(exchange, message, messageId);
+ }
+ final McpServerSession existingSession = sessions.get(requestedSessionId);
+ if (Objects.isNull(existingSession)) {
+ LOGGER.info("SessionId {} not found, creating new session and re-storing", requestedSessionId);
+ return createSessionAndRestoreId(exchange, message, requestedSessionId, messageId);
+ }
+ LOGGER.debug("Processing request for existing session: {}", requestedSessionId);
+ final ServerWebExchange existingExchange = ShenyuMcpExchangeHolder.get(requestedSessionId);
+ LOGGER.debug("Checking exchange mapping for session {}: existing={}", requestedSessionId,
+ Objects.nonNull(existingExchange) ? "present (ID: " + System.identityHashCode(existingExchange) + ")" : "null");
+ if (Objects.isNull(ShenyuMcpExchangeHolder.get(requestedSessionId))) {
+ LOGGER.info("Exchange mapping lost for session {}, re-binding new exchange (ID: {})",
+ requestedSessionId, System.identityHashCode(exchange));
+ configureExchangeForSession(exchange, requestedSessionId);
+ LOGGER.info("Re-bound ServerWebExchange to session {} after reconnect", requestedSessionId);
+ } else {
+ LOGGER.debug("Exchange mapping already exists for session {}, using existing exchange", requestedSessionId);
+ }
+ return processWithExistingSession(existingSession, requestedSessionId, message, messageId)
+ .map(result -> {
+ if (!requestedSessionId.equals(result.getSessionId())) {
+ LOGGER.info("Returning actual session ID {} instead of requested ID {}", result.getSessionId(), requestedSessionId);
+ return new MessageHandlingResult(result.getStatusCode(), result.getResponseBody(), result.getSessionId());
+ }
+ return result;
+ });
+ }
+
+ /**
+ * Creates a temporary session for stateless requests.
+ * This method handles requests that arrive without a session ID by creating a temporary
+ * session that is automatically cleaned up after the request completes. The session is
+ * fully initialized using async methods to avoid blocking the Netty event loop.
+ * Process Flow:
+ * - Create session with auto-generated ID from MCP framework
+ * - Bind ServerWebExchange to session ID before initialization
+ * - Perform async session initialization (initialize + notification)
+ * - Process the business request
+ * - Clean up session and exchange binding after completion
+ *
+ * @param exchange the server web exchange
+ * @param message the JSON-RPC message
+ * @param messageId the message ID for correlation
+ * @return a Mono containing the processing result
+ */
+ private Mono<MessageHandlingResult> createTemporarySessionAndProcess(final ServerWebExchange exchange,
+ final McpSchema.JSONRPCMessage message,
+ final Object messageId) {
+ try {
+ final StreamableHttpSessionTransport tempTransport = new StreamableHttpSessionTransport();
+ final McpServerSession tempSession = sessionFactory.create(tempTransport);
+ final String actualSessionId = tempSession.getId();
+ LOGGER.info("Created temporary session: {}", actualSessionId);
+ sessions.put(actualSessionId, tempSession);
+ sessionTransports.put(actualSessionId, tempTransport);
+ configureExchangeForSession(exchange, actualSessionId);
+ LOGGER.debug("Bound exchange to temporary session: {} before handshake", actualSessionId);
+ initializeSessionDirectly(tempSession, actualSessionId);
+ tempTransport.resetCapturedMessage();
+ return processWithExistingSession(tempSession, actualSessionId, message, messageId)
+ .doFinally(signalType -> {
+ LOGGER.debug("Cleaning up temporary session: {} (signal: {})", actualSessionId, signalType);
+ removeSession(actualSessionId);
+ ShenyuMcpExchangeHolder.remove(actualSessionId);
+ })
+ .map(result -> new MessageHandlingResult(result.getStatusCode(), result.getResponseBody(), null));
+ } catch (Exception e) {
+ LOGGER.error("Error creating temporary session: {}", e.getMessage(), e);
+ final Object errorResponse = createJsonRpcError(messageId, -32603,
+ "Internal error creating temporary session: " + e.getMessage());
+ return Mono.just(new MessageHandlingResult(500, errorResponse, null));
+ }
+ }
+
+ /**
+ * Creates a new session for session restoration scenarios.
+ * This method handles scenarios where a client provides a session ID that no longer
+ * exists on the server (e.g., server restart, session timeout, network disconnection).
+ * A new session is created using the MCP framework, which generates its own session ID.
+ * The client receives the new session ID for subsequent requests.
+ * Important: The MCP framework generates its own session IDs, so the
+ * client's requested session ID may differ from the actual session ID returned.
+ * The response includes the actual session ID that should be used for future requests.
+ *
+ * @param exchange the server web exchange
+ * @param message the JSON-RPC message
+ * @param requestedSessionId the sessionId that the client requested
+ * @param messageId the message ID for correlation
+ * @return a Mono containing the processing result
+ */
+ private Mono<MessageHandlingResult> createSessionAndRestoreId(final ServerWebExchange exchange,
+ final McpSchema.JSONRPCMessage message,
+ final String requestedSessionId,
+ final Object messageId) {
+ try {
+ final StreamableHttpSessionTransport newTransport = new StreamableHttpSessionTransport(requestedSessionId);
+ final McpServerSession newSession = sessionFactory.create(newTransport);
+ final String actualSessionId = newSession.getId();
+ LOGGER.info("Created new session - requested ID: {}, actual ID: {}", requestedSessionId, actualSessionId);
+ sessions.put(actualSessionId, newSession);
+ sessionTransports.put(actualSessionId, newTransport);
+ configureExchangeForSession(exchange, actualSessionId);
+ LOGGER.debug("Bound exchange to restored session: {} before handshake", actualSessionId);
+ initializeSessionDirectly(newSession, actualSessionId);
+ newTransport.resetCapturedMessage();
+ return processWithExistingSession(newSession, actualSessionId, message, messageId)
+ .map(result -> {
+ if (!actualSessionId.equals(requestedSessionId)) {
+ LOGGER.info("Returning actual session ID {} instead of requested ID {}", actualSessionId, requestedSessionId);
+ return new MessageHandlingResult(result.getStatusCode(), result.getResponseBody(), actualSessionId);
+ }
+ return result;
+ });
+ } catch (Exception e) {
+ LOGGER.error("Error creating session with restored ID {}: {}", requestedSessionId, e.getMessage(), e);
+ final Object errorResponse = createJsonRpcError(messageId, -32603,
+ "Internal error restoring session: " + e.getMessage());
+ return Mono.just(new MessageHandlingResult(500, errorResponse, requestedSessionId));
+ }
+ }
+
+ /**
+ * Processes a message with an existing session.
+ * This method contains the core message processing logic that is shared
+ * between regular requests, temporary sessions, and restored sessions.
+ *
+ * @param session the MCP server session
+ * @param sessionId the session identifier
+ * @param message the JSON-RPC message
+ * @param messageId the message ID for correlation
+ * @return a Mono containing the processing result
+ */
+ private Mono<MessageHandlingResult> processWithExistingSession(final McpServerSession session,
+ final String sessionId,
+ final McpSchema.JSONRPCMessage message,
+ final Object messageId) {
+ // Verify exchange is available before processing
+ final ServerWebExchange verifyExchange = ShenyuMcpExchangeHolder.get(sessionId);
+ if (Objects.isNull(verifyExchange)) {
+ LOGGER.error("CRITICAL: No exchange found in ShenyuMcpExchangeHolder for session {} when processing business request. This will cause ToolCallback to fail.", sessionId);
+ } else {
+ LOGGER.debug("Exchange verification passed for session {} (exchange ID: {})",
+ sessionId, System.identityHashCode(verifyExchange));
+ }
+ final StreamableHttpSessionTransport transport = getSessionTransport(sessionId);
+ // Let MCP framework handle the message - framework will send response through transport
+ return session.handle(message)
+ .cast(Object.class)
+ .doOnSuccess(result -> LOGGER.debug("Successfully processed message for session: {}", sessionId))
+ .then(waitForTransportResponse(transport, sessionId, messageId))
+ .onErrorResume(error -> {
+ LOGGER.error("Error processing message for session {}: {}", sessionId, error.getMessage(), error);
+ final Object errorResponse = createJsonRpcError(messageId, -32603,
+ "Internal error: " + error.getMessage());
+ return Mono.just(new MessageHandlingResult(500, errorResponse, sessionId));
+ });
+ }
+
+ /**
+ * Configures exchange attributes for the given session.
+ *
+ * @param exchange the server web exchange
+ * @param sessionId the session identifier
+ */
+ private void configureExchangeForSession(final ServerWebExchange exchange, final String sessionId) {
+ if (Objects.isNull(exchange)) {
+ LOGGER.error("Attempted to configure null exchange for session: {}", sessionId);
+ return;
+ }
+ exchange.getAttributes().put("MCP_SESSION_ID", sessionId);
+ ShenyuMcpExchangeHolder.put(sessionId, exchange);
+ // Verify exchange was stored correctly
+ final ServerWebExchange storedExchange = ShenyuMcpExchangeHolder.get(sessionId);
+ if (Objects.isNull(storedExchange)) {
+ LOGGER.error("Failed to store exchange in ShenyuMcpExchangeHolder for session: {}", sessionId);
+ } else {
+ LOGGER.info("Successfully configured and stored exchange for session: {} (exchange ID: {})",
+ sessionId, System.identityHashCode(exchange));
+ }
+ }
+
+ /**
+ * Cleans up an invalid session that failed initialization.
+ *
+ * @param sessionId the session ID to clean up
+ */
+ private void cleanupInvalidSession(final String sessionId) {
+ sessions.remove(sessionId);
+ sessionTransports.remove(sessionId);
+ LOGGER.debug("Cleaned up invalid session: {}", sessionId);
+ }
+
+ /**
+ * Extracts the protocol version from an initialize request.
+ *
+ * @param message the initialize request message
+ * @return the requested protocol version or default version
+ */
+ private String extractProtocolVersionFromInitialize(final McpSchema.JSONRPCMessage message) {
+ if (message instanceof io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) {
+ final Object params = ((io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) message).params();
+ if (params instanceof Map) {
+ final Map<?, ?> paramsMap = (Map<?, ?>) params;
+ final Object protocolVersion = paramsMap.get("protocolVersion");
+ return Objects.nonNull(protocolVersion) ? protocolVersion.toString() : DEFAULT_PROTOCOL_VERSION;
+ }
+ }
+ return DEFAULT_PROTOCOL_VERSION;
+ }
+
+ /**
+ * Checks if the specified protocol version is supported.
+ *
+ * @param version the protocol version to check
+ * @return true if the version is supported
+ */
+ private boolean isSupportedProtocolVersion(final String version) {
+ return DEFAULT_PROTOCOL_VERSION.equals(version);
+ }
+
+ /**
+ * Creates a standard initialize response for successful initialization.
+ *
+ * @param messageId the original message ID for correlation
+ * @param protocolVersion the negotiated protocol version
+ * @param sessionId the created session ID
+ * @return the initialize response object
+ */
+ private Object createInitializeResponse(final Object messageId, final String protocolVersion, final String sessionId) {
+ final Map<String, Object> capabilities = new java.util.HashMap<>();
+ final Map<String, Object> toolsCapability = new java.util.HashMap<>();
+ toolsCapability.put("listChanged", true);
+ capabilities.put("tools", toolsCapability);
+ final Map<String, Object> serverInfo = new java.util.HashMap<>();
+ serverInfo.put("name", SERVER_NAME);
+ serverInfo.put("version", SERVER_VERSION);
+ final Map<String, Object> result = new java.util.HashMap<>();
+ result.put("protocolVersion", protocolVersion);
+ result.put("capabilities", capabilities);
+ result.put("serverInfo", serverInfo);
+ result.put("instructions", "Use available tools to interact with Shenyu gateway services");
+ result.put("sessionId", sessionId);
+ return createJsonRpcResponse(messageId, result);
+ }
+
+ /**
+ * Extracts the message ID from a JSON-RPC message for response correlation.
+ *
+ * @param message the JSON-RPC message
+ * @return the message ID, or null if not available
+ */
+ private Object extractMessageId(final McpSchema.JSONRPCMessage message) {
+ if (message instanceof io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) {
+ return ((io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest) message).id();
+ }
+ return null;
+ }
+
+ /**
+ * Creates a standard JSON-RPC 2.0 success response.
+ *
+ * @param id the message ID for correlation
+ * @param result the result object
+ * @return the response object
+ */
+ private Object createJsonRpcResponse(final Object id, final Object result) {
+ final Map<String, Object> response = new java.util.HashMap<>();
+ response.put("jsonrpc", JSONRPC_VERSION);
+ if (Objects.nonNull(id)) {
+ response.put("id", id);
+ }
+ response.put("result", Objects.nonNull(result) ? result : new java.util.HashMap<>());
+ return response;
+ }
+
+ /**
+ * Creates a standard JSON-RPC 2.0 error response.
+ *
+ * @param id the message ID for correlation
+ * @param code the error code
+ * @param message the error message
+ * @return the error response object
+ */
+ private Object createJsonRpcError(final Object id, final int code, final String message) {
+ final Map<String, Object> error = new java.util.HashMap<>();
+ error.put("code", code);
+ error.put("message", message);
+ final Map<String, Object> response = new java.util.HashMap<>();
+ response.put("jsonrpc", JSONRPC_VERSION);
+ if (Objects.nonNull(id)) {
+ response.put("id", id);
+ }
+ response.put("error", error);
+ return response;
+ }
+
+ /**
+ * Extracts the session ID from the request headers or query parameters.
+ *
+ * @param request the server request
+ * @return the session ID, or null if not found
+ */
+ private String extractSessionId(final ServerRequest request) {
+ String sessionId = request.queryParam("sessionId").orElse(null);
+ if (Objects.nonNull(sessionId)) {
+ return sessionId;
+ }
+ return request.headers().firstHeader(SESSION_ID_HEADER);
+ }
+
+ /**
+ * Removes a session from the active sessions map.
+ * This method should be called when a session is closed or becomes invalid
+ * to prevent memory leaks and ensure proper cleanup.
+ *
+ * @param sessionId the session ID to remove
+ */
+ public void removeSession(final String sessionId) {
+ final McpServerSession removedSession = sessions.remove(sessionId);
+ final StreamableHttpSessionTransport removedTransport = sessionTransports.remove(sessionId);
+ if (Objects.nonNull(removedSession) || Objects.nonNull(removedTransport)) {
+ LOGGER.debug("Removed session and transport: {}", sessionId);
+ }
+ }
+
+ /**
+ * Gets the session transport for a given session ID.
+ *
+ * @param sessionId the session identifier
+ * @return the session transport, or null if not found
+ */
+ private StreamableHttpSessionTransport getSessionTransport(final String sessionId) {
+ return sessionTransports.get(sessionId);
+ }
+
+ /**
+ * Waits for and retrieves the transport response for correlation with HTTP response.
+ * Checks if the transport has captured a response from the MCP framework and returns it as a MessageHandlingResult.
+ * If no response is available, returns a default success response to prevent hanging.
+ *
+ * @param transport the session transport to check for responses
+ * @param sessionId the session identifier for logging
+ * @param messageId the original message ID for correlation
+ * @return a Mono containing the message handling result
+ */
+ private Mono<MessageHandlingResult> waitForTransportResponse(final StreamableHttpSessionTransport transport,
+ final String sessionId,
+ final Object messageId) {
+ return Mono.fromCallable(() -> {
+ if (Objects.nonNull(transport) && transport.isResponseReady() && Objects.nonNull(transport.getLastSentMessage())) {
+ final McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
+ LOGGER.debug("Retrieved captured response from transport for session: {}", sessionId);
+ return new MessageHandlingResult(200, sentMessage, sessionId);
+ } else {
+ LOGGER.debug("No response captured from transport, returning default success for session: {}", sessionId);
+ final Object successResponse = createJsonRpcResponse(messageId, new java.util.HashMap<>());
+ return new MessageHandlingResult(200, successResponse, sessionId);
+ }
+ });
+ }
+
+ /**
+ * Performs backend session initialization using async MCP protocol handshake.
+ * Simulates complete initialize request-response cycle to ensure session's internal
+ * state machine transitions correctly. Uses async operations to avoid blocking Netty threads.
+ *
+ * @param session the MCP server session
+ * @param sessionId the session identifier
+ */
+ private void initializeSessionDirectly(final McpServerSession session, final String sessionId) {
+ try {
+ // Core strategy: Simulate complete initialize request-response cycle
+ // This ensures the session's internal state machine transitions correctly
+ LOGGER.debug("Starting backend initialization for session: {}", sessionId);
+ // Create a proper initialize request
+ final String initRequestJson = createInitializeRequest();
+ final McpSchema.JSONRPCMessage initRequest = McpSchema.deserializeJsonRpcMessage(objectMapper, initRequestJson);
+ LOGGER.debug("Created initialize request for session: {}", sessionId);
+ // Use subscribe instead of block to avoid blocking in Netty thread
+ session.handle(initRequest)
+ .doOnSuccess(v -> {
+ LOGGER.debug("Initialize request processed successfully for session: {}", sessionId);
+ // Complete the handshake by sending a notification to finalize session state
+ try {
+ final StreamableHttpSessionTransport transport = sessionTransports.get(sessionId);
+ if (Objects.nonNull(transport) && transport.isResponseReady()) {
+ LOGGER.debug("Initialize response captured, session {} should be ready", sessionId);
+ // Complete the handshake by sending a notification to finalize session state
+ completeInitializationHandshakeAsync(session, sessionId);
+ } else {
+ LOGGER.warn("No initialize response captured for session: {}", sessionId);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error checking initialize response for session {}: {}", sessionId, e.getMessage());
+ }
+ })
+ .doOnError(e -> {
+ LOGGER.error("Failed to process initialize request for session {}: {}", sessionId, e.getMessage(), e);
+ })
+ // Use subscribe instead of block to avoid blocking
+ .subscribe();
+ LOGGER.debug("Backend initialization completed for session: {}", sessionId);
+ // Verify initialization success
+ if (verifySessionInitialized(session, sessionId)) {
+ LOGGER.info("Session {} successfully initialized and ready for business requests", sessionId);
+ } else {
+ LOGGER.warn("Session {} initialization may be incomplete - proceeding anyway", sessionId);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Unexpected error during session initialization for {}: {}", sessionId, e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Creates a proper initialize request JSON for backend session initialization.
+ *
+ * @return the initialize request JSON string
+ */
+ private String createInitializeRequest() {
+ final Map<String, Object> params = new java.util.HashMap<>();
+ params.put("protocolVersion", DEFAULT_PROTOCOL_VERSION);
+ params.put("capabilities", new java.util.HashMap<String, Object>() {{
+ put("roots", new java.util.ArrayList<>());
+ }});
+ params.put("clientInfo", new java.util.HashMap<String, Object>() {{
+ put("name", "ShenyuMcpClient");
+ put("version", "1.0.0");
+ }});
+ final Map<String, Object> request = new java.util.HashMap<>();
+ request.put("jsonrpc", JSONRPC_VERSION);
+ request.put("id", "__backend_init");
+ request.put("method", INITIALIZE_METHOD);
+ request.put("params", params);
+ try {
+ return objectMapper.writeValueAsString(request);
+ } catch (Exception e) {
+ LOGGER.error("Failed to create initialize request JSON: {}", e.getMessage());
+ return "{\"jsonrpc\":\"2.0\",\"id\":\"__backend_init\",\"method\":\"initialize\",\"params\":{\"protocolVersion\":\"" + DEFAULT_PROTOCOL_VERSION + "\"}}";
+ }
+ }
+
+ /**
+ * Verifies that a session has been properly initialized.
+ *
+ * @param session the MCP server session to verify
+ * @param sessionId the session identifier for logging
+ * @return true if the session appears to be initialized
+ */
+ private boolean verifySessionInitialized(final McpServerSession session, final String sessionId) {
+ try {
+ // Strategy 1: Check for 'initialized' field
+ try {
+ final java.lang.reflect.Field initializedField = session.getClass().getDeclaredField("initialized");
+ initializedField.setAccessible(true);
+ final Object value = initializedField.get(session);
+ final boolean isInitialized = Boolean.TRUE.equals(value);
+ LOGGER.debug("Session {} initialized field value: {}", sessionId, isInitialized);
+ return isInitialized;
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ LOGGER.debug("No 'initialized' field found or accessible in session: {}", e.getMessage());
+ }
+ // Strategy 2: Check for 'state' field
+ try {
+ final java.lang.reflect.Field stateField = session.getClass().getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object stateValue = stateField.get(session);
+ if (Objects.nonNull(stateValue)) {
+ final String stateStr = stateValue.toString().toLowerCase();
+ final boolean isInitialized = stateStr.contains("init") && !stateStr.contains("uninit");
+ LOGGER.debug("Session {} state value: {} (initialized: {})", sessionId, stateValue, isInitialized);
+ return isInitialized;
+ }
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ LOGGER.debug("No 'state' field found or accessible in session: {}", e.getMessage());
+ }
+ // Strategy 3: Check transport response
+ final StreamableHttpSessionTransport transport = sessionTransports.get(sessionId);
+ if (Objects.nonNull(transport)) {
+ final boolean hasResponse = transport.isResponseReady();
+ LOGGER.debug("Session {} transport has response ready: {}", sessionId, hasResponse);
+ // If we have a response, the session likely processed the initialize request
+ return hasResponse;
+ }
+ LOGGER.debug("Unable to verify initialization state for session: {}", sessionId);
+ // Unable to verify, assume not initialized
+ return false;
+ } catch (Exception e) {
+ LOGGER.error("Error verifying session initialization for {}: {}", sessionId, e.getMessage());
+ return false;
+ }
+ }
+
+ /**
+ * Completes the handshake by sending a notification to finalize session state asynchronously.
+ * Uses multiple strategies to ensure session state is properly set, including sending
+ * initialized notifications and using reflection as fallback.
+ *
+ * @param session the MCP server session
+ * @param sessionId the session identifier
+ */
+ private void completeInitializationHandshakeAsync(final McpServerSession session, final String sessionId) {
+ try {
+ // Strategy 1: Send a "initialized" notification to complete the handshake
+ try {
+ final Map<String, Object> notification = new java.util.HashMap<>();
+ notification.put("jsonrpc", JSONRPC_VERSION);
+ notification.put("method", "notifications/initialized");
+ notification.put("params", new java.util.HashMap<>());
+ final String notificationJson = objectMapper.writeValueAsString(notification);
+ final McpSchema.JSONRPCMessage notificationMessage = McpSchema.deserializeJsonRpcMessage(objectMapper, notificationJson);
+ session.handle(notificationMessage)
+ .doOnSuccess(v -> {
+ LOGGER.debug("Initialized notification sent successfully for session: {}", sessionId);
+ })
+ .doOnError(e -> {
+ LOGGER.debug("Initialized notification failed for session {}: {}", sessionId, e.getMessage());
+ })
+ .onErrorComplete()
+ .subscribe();
+ } catch (Exception e) {
+ LOGGER.debug("Strategy 1 failed: {}", e.getMessage());
+ }
+ // Strategy 2: Force state transition using reflection as a fallback
+ try {
+ final java.lang.reflect.Field initializedField = session.getClass().getDeclaredField("initialized");
+ initializedField.setAccessible(true);
+ initializedField.set(session, true);
+ LOGGER.debug("Successfully set session {} to initialized state via reflection", sessionId);
+ } catch (Exception e) {
+ LOGGER.debug("Strategy 2 failed: {}", e.getMessage());
+ }
+ // Strategy 3: Try to set state field
+ try {
+ final java.lang.reflect.Field stateField = session.getClass().getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object stateValue = stateField.get(session);
+ if (Objects.nonNull(stateValue) && stateValue.getClass().isEnum()) {
+ // Try to find INITIALIZED enum value
+ for (Object enumConstant : stateValue.getClass().getEnumConstants()) {
+ if (enumConstant.toString().toLowerCase().contains("init") && !enumConstant.toString().toLowerCase().contains("uninit")) {
+ stateField.set(session, enumConstant);
+ LOGGER.debug("Successfully set session {} state to {} via reflection", sessionId, enumConstant);
+ break;
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.debug("Strategy 3 failed: {}", e.getMessage());
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error completing handshake for session {}: {}", sessionId, e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Session transport implementation for Streamable HTTP with proper lifecycle management.
+ * This transport handles the communication between the MCP framework and the Streamable HTTP
+ * protocol. It captures responses from the MCP session and makes them available for
+ * HTTP response correlation.
+ */
+ private class StreamableHttpSessionTransport implements McpServerTransport {
+
+ private final String sessionId;
+
+ private volatile boolean closed;
+
+ private volatile McpSchema.JSONRPCMessage lastSentMessage;
+
+ private volatile boolean responseReady;
+
+ /**
+ * Creates a new session transport with auto-generated session ID.
+ */
+ StreamableHttpSessionTransport() {
+ this.sessionId = java.util.UUID.randomUUID().toString();
+ LOGGER.debug("Created StreamableHttpSessionTransport with auto-generated sessionId: {}", this.sessionId);
+ }
+
+ /**
+ * Creates a new session transport with the specified session ID.
+ *
+ * @param sessionId the session identifier, or null to auto-generate
+ */
+ StreamableHttpSessionTransport(final String sessionId) {
+ this.sessionId = Objects.nonNull(sessionId) ? sessionId : java.util.UUID.randomUUID().toString();
+ LOGGER.debug("Created StreamableHttpSessionTransport with sessionId: {}", this.sessionId);
+ }
+
+ /**
+ * Gets the last message sent through this transport.
+ * This is used for Streamable HTTP response correlation, allowing the
+ * transport provider to capture and return responses sent by the MCP framework.
+ *
+ * @return the last sent message, or null if no message has been sent
+ */
+ public McpSchema.JSONRPCMessage getLastSentMessage() {
+ return lastSentMessage;
+ }
+
+ /**
+ * Checks if a response is ready for retrieval.
+ *
+ * @return true if response is available
+ */
+ public boolean isResponseReady() {
+ return responseReady;
+ }
+
+ @Override
+ public Mono<Void> sendMessage(final McpSchema.JSONRPCMessage message) {
+ if (!closed) {
+ this.lastSentMessage = message;
+ this.responseReady = true;
+ LOGGER.debug("Captured response message for session: {}", sessionId);
+ }
+ return Mono.empty();
+ }
+
+ @Override
+ public <T> T unmarshalFrom(final Object data, final TypeReference<T> typeRef) {
+ try {
+ return new ObjectMapper().convertValue(data, typeRef);
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ }
+
+ @Override
+ public Mono<Void> closeGracefully() {
+ return Mono.fromRunnable(() -> {
+ if (!closed) {
+ closed = true;
+ removeSession(sessionId);
+ LOGGER.debug("Session transport closed gracefully: {}", sessionId);
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ closed = true;
+ removeSession(sessionId);
+ LOGGER.debug("Session transport closed immediately: {}", sessionId);
+ }
+ }
+
+ /**
+ * Clears the captured message and resets the response flag so that a subsequent
+ * business request is not confused with the internal initialize handshake that
+ * is executed during short-reconnect or temporary session creation.
+ */
+ public void resetCapturedMessage() {
+ this.lastSentMessage = null;
+ this.responseReady = false;
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/StreamableHttpProviderBuilder.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/StreamableHttpProviderBuilder.java
new file mode 100644
index 0000000000..ff59ace50c
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/StreamableHttpProviderBuilder.java
@@ -0,0 +1,81 @@
+/*
+ * 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.mcp.server.transport;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.util.Assert;
+
+/**
+ * Builder class for constructing instances of ShenyuStreamableHttpServerTransportProvider.
+ * This builder provides a fluent interface for configuring transport provider instances
+ * with proper validation and default values.
+ * Usage Example:
+ * ShenyuStreamableHttpServerTransportProvider provider =
+ * ShenyuStreamableHttpServerTransportProvider.builder()
+ * .objectMapper(new ObjectMapper())
+ * .endpoint("/mcp")
+ * .build();
+ */
+public class StreamableHttpProviderBuilder {
+
+ /**
+ * Default unified endpoint path for Streamable HTTP protocol.
+ */
+ private static final String DEFAULT_ENDPOINT = "/mcp";
+
+ private ObjectMapper objectMapper;
+
+ private String endpoint = DEFAULT_ENDPOINT;
+
+ /**
+ * Sets the ObjectMapper for JSON serialization/deserialization.
+ *
+ * @param objectMapper the ObjectMapper instance (required)
+ * @return this builder for method chaining
+ * @throws IllegalArgumentException if objectMapper is null
+ */
+ public StreamableHttpProviderBuilder objectMapper(final ObjectMapper objectMapper) {
+ Assert.notNull(objectMapper, "ObjectMapper must not be null");
+ this.objectMapper = objectMapper;
+ return this;
+ }
+
+ /**
+ * Sets the endpoint path for the transport provider.
+ *
+ * @param endpoint the endpoint path (defaults to "/mcp" if not specified)
+ * @return this builder for method chaining
+ * @throws IllegalArgumentException if endpoint is null
+ */
+ public StreamableHttpProviderBuilder endpoint(final String endpoint) {
+ Assert.notNull(endpoint, "Endpoint must not be null");
+ this.endpoint = endpoint;
+ return this;
+ }
+
+ /**
+ * Builds a new ShenyuStreamableHttpServerTransportProvider instance.
+ *
+ * @return the configured transport provider
+ * @throws IllegalStateException if required configuration is missing
+ */
+ public ShenyuStreamableHttpServerTransportProvider build() {
+ Assert.notNull(objectMapper, "ObjectMapper must be configured");
+ return new ShenyuStreamableHttpServerTransportProvider(objectMapper, endpoint);
+ }
+}
\ No newline at end of file
diff --git a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/utils/JsonSchemaUtil.java b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/utils/JsonSchemaUtil.java
index 4c62efdd39..fee135e937 100644
--- a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/utils/JsonSchemaUtil.java
+++ b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/utils/JsonSchemaUtil.java
@@ -46,7 +46,7 @@ public final class JsonSchemaUtil {
* @return the JSON schema as a string
*/
public static String createParameterSchema(final List<McpServerToolParameter> parameters,
- final SchemaOption... schemaOptions) {
+ final SchemaOption... schemaOptions) {
if (CollectionUtils.isEmpty(parameters)) {
return emptySchema(schemaOptions);
}
@@ -58,7 +58,7 @@ public final class JsonSchemaUtil {
ObjectNode properties = schema.putObject("properties");
for (McpServerToolParameter parameter : parameters) {
ObjectNode property = properties.putObject(parameter.getName());
- property.put("type", "string");
+ property.put("type", parameter.getType());
property.put("description", parameter.getDescription());
}