Browse Source
- 新增 McpStreamableController 提供流式HTTP单端点支持 /tenantadmin/mcp - 实现 StreamableHttpServerTransportProvider 支持MCP协议的流式HTTP传输 - 添加 @McpToolClass 和 @McpToolParam 注解用于标记MCP工具方法和参数 - 开发 McpToolScanner 扫描器自动注册带有注解的工具方法 - 集成 mcp-java8-sdk 依赖支持MCP协议服务器功能 - 新增 McpToolArgInfo 和 McpToolParamType 模型类处理工具参数元信息 - 配置 McpToolConfig 自动装配SSE和流式HTTP两种传输方式 - 为 MyPageParam 添加 MCP 参数注解支持分页参数映射pull/1/head
19 changed files with 1193 additions and 21 deletions
@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
|
||||
package apelet.common.core.mcp.annotation; |
||||
|
||||
import java.lang.annotation.Documented; |
||||
import java.lang.annotation.ElementType; |
||||
import java.lang.annotation.Retention; |
||||
import java.lang.annotation.RetentionPolicy; |
||||
import java.lang.annotation.Target; |
||||
|
||||
/** |
||||
* MCP 工具方法注解：标注在业务方法上，启动时由 McpToolScanner 扫描自动生成 MCP Tool 并注册。 |
||||
* <p>name 与 description 即 MCP Tool 的 name / description，inputSchema 由方法参数上的 @McpToolParam 组装。</p> |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-31 |
||||
*/ |
||||
@Documented |
||||
@Target(ElementType.METHOD) |
||||
@Retention(RetentionPolicy.RUNTIME) |
||||
public @interface McpToolClass { |
||||
|
||||
/** |
||||
* MCP Tool 名称，客户端 tools/list 看到的 name。 |
||||
*/ |
||||
String name(); |
||||
|
||||
/** |
||||
* MCP Tool 描述，供大模型理解用途。 |
||||
*/ |
||||
String description(); |
||||
} |
||||
@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
|
||||
package apelet.common.core.mcp.annotation; |
||||
|
||||
import apelet.common.core.mcp.model.McpToolParamType; |
||||
|
||||
import java.lang.annotation.Documented; |
||||
import java.lang.annotation.ElementType; |
||||
import java.lang.annotation.Retention; |
||||
import java.lang.annotation.RetentionPolicy; |
||||
import java.lang.annotation.Target; |
||||
|
||||
/** |
||||
* MCP 工具方法参数注解：标注在 @McpToolClass 方法的参数上，以及 type=OBJECT 的 schemaClass 字段上。 |
||||
* <p>启动扫描时按注解属性组装 inputSchema；调用工具时按 fieldName 从入参取值绑定到方法实参。</p> |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-31 |
||||
*/ |
||||
@Documented |
||||
@Target({ElementType.PARAMETER, ElementType.FIELD}) |
||||
@Retention(RetentionPolicy.RUNTIME) |
||||
public @interface McpToolParam { |
||||
|
||||
/** |
||||
* 入参字段名，即客户端传参时使用的 key。 |
||||
* 未配置时取 Java 参数名（默认编译带调试信息，可通过 DefaultParameterNameDiscoverer 获取）。 |
||||
*/ |
||||
String fieldName() default ""; |
||||
|
||||
/** |
||||
* 字段类型（JSON Schema 的 type），默认字符串。 |
||||
*/ |
||||
McpToolParamType type() default McpToolParamType.STRING; |
||||
|
||||
/** |
||||
* 字段描述，供大模型理解入参含义。 |
||||
*/ |
||||
String description() default ""; |
||||
|
||||
/** |
||||
* 是否必填，默认 false。 |
||||
*/ |
||||
boolean required() default false; |
||||
|
||||
/** |
||||
* type=OBJECT 时必填，指向的对象类型，其 @McpToolParam 标注字段会递归组装成嵌套 schema。 |
||||
*/ |
||||
Class<?> schemaClass() default Void.class; |
||||
} |
||||
@ -0,0 +1,91 @@
@@ -0,0 +1,91 @@
|
||||
package apelet.common.core.mcp.config; |
||||
|
||||
import apelet.common.core.mcp.transport.StreamableHttpServerTransportProvider; |
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import io.modelcontextprotocol.server.McpServer; |
||||
import io.modelcontextprotocol.server.McpServerFeatures; |
||||
import io.modelcontextprotocol.server.McpSyncServer; |
||||
import io.modelcontextprotocol.server.transport.WebMvcSseServerTransportProvider; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.context.annotation.Bean; |
||||
import org.springframework.context.annotation.Configuration; |
||||
import org.springframework.web.servlet.function.RouterFunction; |
||||
import org.springframework.web.servlet.function.ServerResponse; |
||||
|
||||
import java.util.List; |
||||
|
||||
/** |
||||
* MCP Server 装配：把业务工具暴露为 MCP tool，供大模型（客户端）调用。 |
||||
* 基于 mcp-java8-sdk 0.8.1（构造器 API，非新版 builder()）。 |
||||
* 该 SDK 不提供 listTools()，故注册时同步收集一份工具清单供调试接口读取（见 McpToolScanner.getTools()）。 |
||||
* 双传输并存：SSE（/mcp/sse + /mcp/message）+ Streamable HTTP（/mcp 单端点）。 |
||||
* 工具由 McpToolScanner 启动时扫描 @McpToolClass 自动生成并统一注入。 |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-27 |
||||
*/ |
||||
@Configuration |
||||
public class McpToolConfig { |
||||
|
||||
/** |
||||
* MCP 消息专用 ObjectMapper：应用注入的 ObjectMapper 在序列化 @JsonTypeInfo 多态对象 |
||||
* （如 CallToolResult.content 的 TextContent）时丢失 type 判别符，导致客户端反序列化失败， |
||||
* 故 MCP 的序列化统一走此普通 mapper（普通 mapper 会正常输出 "type":"text"）。 |
||||
*/ |
||||
private static final ObjectMapper MCP_OBJECT_MAPPER = new ObjectMapper(); |
||||
|
||||
@Autowired |
||||
private McpToolScanner mcpToolScanner; |
||||
|
||||
/** |
||||
* MCP 传输层（Spring MVC SSE）：SSE 长连接端点 + JSON-RPC message 端点。 |
||||
* 端点路径带 /tenantadmin 前缀（本应用无 servlet context-path，模块 urlPrefix 统一用 /tenantadmin/xxx）。 |
||||
*/ |
||||
@Bean |
||||
public WebMvcSseServerTransportProvider mcpTransport() { |
||||
return new WebMvcSseServerTransportProvider(MCP_OBJECT_MAPPER, "/tenantadmin/mcp/message", "/tenantadmin/mcp/sse"); |
||||
} |
||||
|
||||
/** |
||||
* 把 SSE transport 内部的 RouterFunction（SSE/message 两个端点）注册进 Spring MVC。 |
||||
*/ |
||||
@Bean |
||||
public RouterFunction<ServerResponse> mcpEndpoints(WebMvcSseServerTransportProvider mcpTransport) { |
||||
return mcpTransport.getRouterFunction(); |
||||
} |
||||
|
||||
/** |
||||
* 构建同步 MCP Server（SSE 传输），并注册业务工具。 |
||||
*/ |
||||
@Bean |
||||
public McpSyncServer mcpSyncServer(WebMvcSseServerTransportProvider mcpTransport) { |
||||
return registerTools(McpServer.sync(mcpTransport).serverInfo("business-mcp-server", "1.0.0")).build(); |
||||
} |
||||
|
||||
/** |
||||
* MCP 传输层（Streamable HTTP，单端点 POST /tenantadmin/mcp）。 |
||||
* 端点由 McpStreamableController 提供，不走 RouterFunction（避免应用转换器二次加工响应）。 |
||||
*/ |
||||
@Bean |
||||
public StreamableHttpServerTransportProvider streamableHttpTransport() { |
||||
return new StreamableHttpServerTransportProvider(MCP_OBJECT_MAPPER); |
||||
} |
||||
|
||||
/** |
||||
* 构建同步 MCP Server（Streamable HTTP 传输），注册同一套业务工具。 |
||||
*/ |
||||
@Bean |
||||
public McpSyncServer mcpStreamableSyncServer(StreamableHttpServerTransportProvider streamableHttpTransport) { |
||||
return registerTools(McpServer.sync(streamableHttpTransport).serverInfo("business-mcp-server", "1.0.0")).build(); |
||||
} |
||||
|
||||
/** |
||||
* 注册 MCP 工具（SSE 与 Streamable 两个 server 共用同一套注册逻辑）。 |
||||
* 工具清单由 McpToolScanner 扫描 @McpToolClass 自动组装，这里批量注入。 |
||||
*/ |
||||
private McpServer.SyncSpecification registerTools(McpServer.SyncSpecification spec) { |
||||
List<McpServerFeatures.SyncToolSpecification> specifications = mcpToolScanner.getToolSpecifications(); |
||||
return specifications.isEmpty() ? spec : spec.tools(specifications); |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,499 @@
@@ -0,0 +1,499 @@
|
||||
package apelet.common.core.mcp.config; |
||||
|
||||
import apelet.common.core.mcp.annotation.McpToolClass; |
||||
import apelet.common.core.mcp.annotation.McpToolParam; |
||||
import apelet.common.core.mcp.model.McpToolArgInfo; |
||||
import apelet.common.core.mcp.model.McpToolParamType; |
||||
import apelet.common.core.object.TokenData; |
||||
import cn.hutool.core.util.StrUtil; |
||||
import com.alibaba.fastjson.JSON; |
||||
import com.alibaba.fastjson.JSONArray; |
||||
import com.alibaba.fastjson.JSONObject; |
||||
import io.modelcontextprotocol.server.McpServerFeatures; |
||||
import io.modelcontextprotocol.spec.McpSchema; |
||||
import lombok.Getter; |
||||
import lombok.extern.slf4j.Slf4j; |
||||
import org.springframework.beans.factory.annotation.Value; |
||||
import org.springframework.context.ApplicationContext; |
||||
import org.springframework.core.DefaultParameterNameDiscoverer; |
||||
import org.springframework.stereotype.Component; |
||||
import org.springframework.util.ClassUtils; |
||||
import org.springframework.web.bind.annotation.RestController; |
||||
import org.springframework.web.context.request.RequestAttributes; |
||||
import org.springframework.web.context.request.RequestContextHolder; |
||||
import org.springframework.web.context.request.ServletRequestAttributes; |
||||
|
||||
import javax.annotation.PostConstruct; |
||||
import javax.servlet.http.HttpServletRequest; |
||||
import javax.servlet.http.HttpServletResponse; |
||||
import java.lang.reflect.*; |
||||
import java.math.BigDecimal; |
||||
import java.util.*; |
||||
import java.util.stream.Collectors; |
||||
|
||||
/** |
||||
* MCP 工具扫描注册器：启动时从 Spring 容器取 @RestController/@Controller/@Component 等组件 bean， |
||||
* 扫描其上 @McpToolClass 标注的方法，自动组装 McpSchema.Tool 并生成调用 handler。 |
||||
* <p>直接基于容器 bean 扫描（getBeanNamesForAnnotation），不做 classpath 资源扫描， |
||||
* 避免 fat jar 嵌套 jar 里包名通配（**）扫不到的问题，也无需 scan-packages 等配置。 |
||||
* 组装结果以 SyncToolSpecification 列表暴露给 McpToolConfig 注入 SSE + Streamable 两个 server。</p> |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-31 |
||||
*/ |
||||
@Slf4j |
||||
@Component |
||||
public class McpToolScanner { |
||||
|
||||
/** |
||||
* 方法参数名发现器：优先 -parameters，兜底 LocalVariableTable（默认编译带调试信息）。 |
||||
*/ |
||||
private static final DefaultParameterNameDiscoverer PARAM_NAME_DISCOVERER = new DefaultParameterNameDiscoverer(); |
||||
private final ApplicationContext applicationContext; |
||||
/** |
||||
* MCP 调用使用的默认用户Id（MCP 无 HTTP 登录态，服务内 TokenData.takeFromRequest() 会 NPE，调用前注入）。 |
||||
*/ |
||||
@Value("${apelet.mcp.default-user-id:0}") |
||||
private Long defaultUserId; |
||||
/** |
||||
* 已组装好的工具注册项（供 spec.tools 批量注入）。 |
||||
*/ |
||||
@Getter |
||||
private List<McpServerFeatures.SyncToolSpecification> toolSpecifications = Collections.emptyList(); |
||||
/** |
||||
* 已注册的工具清单（SDK 不提供 listTools()，供调试接口读取）。 |
||||
*/ |
||||
private List<McpSchema.Tool> toolList = Collections.emptyList(); |
||||
|
||||
public McpToolScanner(ApplicationContext applicationContext) { |
||||
this.applicationContext = applicationContext; |
||||
} |
||||
|
||||
/** |
||||
* 按返回类型给动态代理一个安全的默认值（对象 null、字符串空、集合空、数组空、数值 0、布尔 false）。 |
||||
*/ |
||||
private static Object defaultValue(Class<?> returnType) { |
||||
if (returnType == String.class) { |
||||
return ""; |
||||
} |
||||
if (returnType == boolean.class) { |
||||
return false; |
||||
} |
||||
if (returnType == int.class) { |
||||
return 0; |
||||
} |
||||
if (returnType == long.class) { |
||||
return 0L; |
||||
} |
||||
if (returnType == List.class || returnType == Map.class || returnType == Set.class) { |
||||
return Collections.emptyMap(); |
||||
} |
||||
if (returnType.isArray()) { |
||||
return Array.newInstance(returnType.getComponentType(), 0); |
||||
} |
||||
return null; |
||||
} |
||||
|
||||
@PostConstruct |
||||
public void init() { |
||||
List<McpServerFeatures.SyncToolSpecification> specifications = new ArrayList<>(); |
||||
List<McpSchema.Tool> tools = new ArrayList<>(); |
||||
// 直接从 Spring 容器取组件 bean 扫描，不做 classpath 资源扫描。
|
||||
// 原因：ClassPathScanningCandidateComponentProvider 走 PathMatchingResourcePatternResolver 递归匹配 classpath 资源，
|
||||
// 应用打成 fat jar 后 @McpToolClass 类在 BOOT-INF/lib/*.jar 嵌套 jar 里，包名带 `**` 时匹配不上，导致扫不到；
|
||||
// 改为从容器取 bean 则完全不受打包方式影响。@RestController/@Controller 都被 @Component 元注解标注，
|
||||
// getBeanNamesForAnnotation(Component.class) 全部命中，@Service/@Repository 等组件同样覆盖。
|
||||
Set<Class<?>> scanned = new HashSet<>(); |
||||
for (String beanName : applicationContext.getBeanNamesForAnnotation(RestController.class)) { |
||||
Class<?> clazz = applicationContext.getType(beanName); |
||||
if (clazz == null) { |
||||
continue; |
||||
} |
||||
clazz = ClassUtils.getUserClass(clazz); |
||||
if (!scanned.add(clazz)) { |
||||
continue; |
||||
} |
||||
this.scanClassMethods(specifications, tools, clazz, beanName); |
||||
} |
||||
this.toolSpecifications = specifications; |
||||
this.toolList = tools; |
||||
log.info("MCP 工具扫描完成，共注册 {} 个 tool：{}", tools.size(), |
||||
tools.stream().map(McpSchema.Tool::getName).collect(Collectors.toList())); |
||||
} |
||||
|
||||
/** |
||||
* 扫描单个 bean 类的 @McpToolClass 方法并注册。 |
||||
*/ |
||||
private void scanClassMethods(List<McpServerFeatures.SyncToolSpecification> specifications, |
||||
List<McpSchema.Tool> tools, Class<?> clazz, String beanName) { |
||||
for (Method method : clazz.getDeclaredMethods()) { |
||||
// 只暴露 public 方法为 MCP 工具，私有/受保护方法不注册。
|
||||
if (!Modifier.isPublic(method.getModifiers())) { |
||||
continue; |
||||
} |
||||
McpToolClass ann = method.getAnnotation(McpToolClass.class); |
||||
if (ann == null) { |
||||
continue; |
||||
} |
||||
this.registerTool(specifications, tools, clazz, beanName, method, ann); |
||||
} |
||||
} |
||||
|
||||
public List<McpSchema.Tool> getTools() { |
||||
return toolList; |
||||
} |
||||
|
||||
/** |
||||
* 组装单个工具：inputSchema + Tool + 调用 handler。 |
||||
*/ |
||||
private void registerTool(List<McpServerFeatures.SyncToolSpecification> specifications, |
||||
List<McpSchema.Tool> tools, Class<?> clazz, String beanName, Method method, McpToolClass ann) { |
||||
List<McpToolArgInfo> argInfos = this.buildArgInfos(method); |
||||
String inputSchema = this.buildInputSchema(argInfos); |
||||
McpSchema.Tool tool = new McpSchema.Tool(ann.name(), ann.description(), inputSchema); |
||||
// 懒取 bean：只有真正带 @McpToolClass 方法的类才触发初始化；拿到的是代理实例，invoke 时走 AOP。
|
||||
Object bean = applicationContext.getBean(beanName); |
||||
McpServerFeatures.SyncToolSpecification specification = new McpServerFeatures.SyncToolSpecification(); |
||||
specification.setTool(tool); |
||||
specification.setCall((exchange, args) -> this.invoke(bean, method, argInfos, args)); |
||||
specifications.add(specification); |
||||
tools.add(tool); |
||||
log.info("MCP 工具已注册：{} (inputSchema={})", tool.getName(), inputSchema); |
||||
} |
||||
|
||||
/** |
||||
* 组装方法参数绑定信息：@McpToolParam 未配置 fieldName 时取 Java 参数名。 |
||||
*/ |
||||
private List<McpToolArgInfo> buildArgInfos(Method method) { |
||||
Parameter[] parameters = method.getParameters(); |
||||
String[] paramNames = PARAM_NAME_DISCOVERER.getParameterNames(method); |
||||
List<McpToolArgInfo> argInfos = new ArrayList<>(); |
||||
for (int i = 0; i < parameters.length; i++) { |
||||
McpToolParam ann = parameters[i].getAnnotation(McpToolParam.class); |
||||
if (ann == null) { |
||||
log.warn("MCP 工具方法 [{}#{}] 参数 [{}] 缺少 @McpToolParam，跳过该参数", method.getDeclaringClass().getSimpleName(), method.getName(), i); |
||||
continue; |
||||
} |
||||
String fieldName = StrUtil.isNotBlank(ann.fieldName()) ? ann.fieldName() |
||||
: (paramNames != null && paramNames.length > i ? paramNames[i] : "arg" + i); |
||||
argInfos.add(this.toArgInfo(i, parameters[i].getType(), parameters[i].getParameterizedType(), ann, fieldName)); |
||||
} |
||||
return argInfos; |
||||
} |
||||
|
||||
/** |
||||
* 由注解 + Java 类型组装元信息（方法参数与对象字段共用）。 |
||||
*/ |
||||
private McpToolArgInfo toArgInfo(int paramIndex, Class<?> paramType, Type genericType, McpToolParam ann, String fieldName) { |
||||
Class<?> elementType = ann.type() == McpToolParamType.ARRAY ? this.inferElementType(genericType, paramType) : null; |
||||
Class<?> schemaClass = ann.type() == McpToolParamType.OBJECT ? (ann.schemaClass() == Void.class ? paramType : ann.schemaClass()) : null; |
||||
return new McpToolArgInfo(paramIndex, fieldName, ann.type(), ann.description(), paramType, elementType, schemaClass, ann.required()); |
||||
} |
||||
|
||||
/** |
||||
* 组装 inputSchema：{"type":"object","properties":{...},"required":[...]}。 |
||||
*/ |
||||
private String buildInputSchema(List<McpToolArgInfo> argInfos) { |
||||
JSONObject root = new JSONObject(true); |
||||
root.put("type", "object"); |
||||
JSONObject properties = new JSONObject(true); |
||||
JSONArray required = new JSONArray(); |
||||
for (McpToolArgInfo info : argInfos) { |
||||
properties.put(info.getFieldName(), this.buildElementSchema(info)); |
||||
if (info.isRequired()) { |
||||
required.add(info.getFieldName()); |
||||
} |
||||
} |
||||
if (!properties.isEmpty()) { |
||||
root.put("properties", properties); |
||||
} |
||||
if (!required.isEmpty()) { |
||||
root.put("required", required); |
||||
} |
||||
return root.toJSONString(); |
||||
} |
||||
|
||||
/** |
||||
* 组装单个参数/字段的 JSON Schema；OBJECT 递归展开 schemaClass 的 @McpToolParam 字段。 |
||||
*/ |
||||
private JSONObject buildElementSchema(McpToolArgInfo info) { |
||||
JSONObject schema = new JSONObject(true); |
||||
schema.put("type", info.getType().getJsonType()); |
||||
if (StrUtil.isNotBlank(info.getDescription())) { |
||||
schema.put("description", info.getDescription()); |
||||
} |
||||
if (info.getType() == McpToolParamType.OBJECT && info.getSchemaClass() != null) { |
||||
JSONObject properties = new JSONObject(true); |
||||
JSONArray required = new JSONArray(); |
||||
this.collectObjectSchema(info.getSchemaClass(), properties, required); |
||||
if (!properties.isEmpty()) { |
||||
schema.put("properties", properties); |
||||
} |
||||
if (!required.isEmpty()) { |
||||
schema.put("required", required); |
||||
} |
||||
} else if (info.getType() == McpToolParamType.ARRAY && info.getElementType() != null) { |
||||
JSONObject items = new JSONObject(true); |
||||
items.put("type", this.toJsonType(info.getElementType())); |
||||
schema.put("items", items); |
||||
} |
||||
return schema; |
||||
} |
||||
|
||||
/** |
||||
* 收集对象类型中 @McpToolParam 标注字段的嵌套 schema。 |
||||
*/ |
||||
private void collectObjectSchema(Class<?> schemaClass, JSONObject properties, JSONArray required) { |
||||
for (Field field : schemaClass.getDeclaredFields()) { |
||||
McpToolParam ann = field.getAnnotation(McpToolParam.class); |
||||
if (ann == null) { |
||||
continue; |
||||
} |
||||
String fieldName = StrUtil.isNotBlank(ann.fieldName()) ? ann.fieldName() : field.getName(); |
||||
properties.put(fieldName, this.buildElementSchema(this.toArgInfo(-1, field.getType(), field.getGenericType(), ann, fieldName))); |
||||
if (ann.required()) { |
||||
required.add(fieldName); |
||||
} |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 推断数组元素类型：List<E> 泛型实参或数组组件类型，未知默认 String。 |
||||
*/ |
||||
private Class<?> inferElementType(Type genericType, Class<?> rawType) { |
||||
if (genericType instanceof ParameterizedType) { |
||||
Type actual = ((ParameterizedType) genericType).getActualTypeArguments()[0]; |
||||
if (actual instanceof Class) { |
||||
return (Class<?>) actual; |
||||
} |
||||
} |
||||
return rawType.isArray() ? rawType.getComponentType() : String.class; |
||||
} |
||||
|
||||
/** |
||||
* Java 类型 → JSON Schema type。 |
||||
*/ |
||||
private String toJsonType(Class<?> javaType) { |
||||
if (javaType == BigDecimal.class || javaType == Double.class || javaType == Float.class |
||||
|| javaType == double.class || javaType == float.class) { |
||||
return "number"; |
||||
} |
||||
if (javaType == Long.class || javaType == Integer.class || javaType == Short.class || javaType == Byte.class |
||||
|| javaType == long.class || javaType == int.class) { |
||||
return "integer"; |
||||
} |
||||
if (javaType == Boolean.class || javaType == boolean.class) { |
||||
return "boolean"; |
||||
} |
||||
return "string"; |
||||
} |
||||
|
||||
/** |
||||
* 调用工具方法：按元信息从入参 map 取参、转换类型、反射调用，结果包成 CallToolResult。 |
||||
* <p>MCP 调用运行在 reactor 调度线程，无 servlet 请求上下文，需先挂一个动态代理的 request/response |
||||
* （承载 TokenData）再调用，否则 @OperationLog 等切面在 RequestContextHolder 上强转 NPE。</p> |
||||
*/ |
||||
private McpSchema.CallToolResult invoke(Object bean, Method method, List<McpToolArgInfo> argInfos, Map<String, Object> args) { |
||||
try { |
||||
//todo 这里要求带上token MCP 无 HTTP 登录态，注入默认用户 TokenData，避免服务内 takeFromRequest() 拿到 null 抛 NPE。
|
||||
TokenData tokenData = new TokenData(); |
||||
tokenData.setUserId(this.defaultUserId); |
||||
return this.withRequestContext(tokenData, () -> { |
||||
Object[] invokeArgs = this.bindArgs(args, argInfos, method.getParameterCount()); |
||||
Object result = method.invoke(bean, invokeArgs); |
||||
String text = result == null ? "" : (result instanceof String ? (String) result : JSON.toJSONString(result)); |
||||
return new McpSchema.CallToolResult(Collections.singletonList(new McpSchema.TextContent(text)), false); |
||||
}); |
||||
} catch (Exception e) { |
||||
Throwable cause = e instanceof InvocationTargetException ? ((InvocationTargetException) e).getTargetException() : e; |
||||
log.error("MCP 工具 [{}] 调用失败", method.getName(), cause); |
||||
return new McpSchema.CallToolResult(Collections.singletonList(new McpSchema.TextContent(cause.getMessage())), true); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 在指定 TokenData 的请求上下文下执行动作：当前线程没有请求上下文时挂动态代理的 request/response， |
||||
* 执行完恢复原样。 |
||||
*/ |
||||
private <T> T withRequestContext(TokenData tokenData, ThrowingSupplier<T> action) throws Exception { |
||||
RequestAttributes previous = RequestContextHolder.getRequestAttributes(); |
||||
boolean created = previous == null; |
||||
if (created) { |
||||
ServletRequestAttributes attributes = new ServletRequestAttributes(mcpMockRequest(), mcpMockResponse()); |
||||
RequestContextHolder.setRequestAttributes(attributes); |
||||
} |
||||
TokenData.addToRequest(tokenData); |
||||
try { |
||||
return action.get(); |
||||
} finally { |
||||
if (created) { |
||||
RequestContextHolder.resetRequestAttributes(); |
||||
} |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 动态代理的 HttpServletRequest：仅承载 getAttribute/setAttribute（供 TokenData），其余方法返回安全默认值。 |
||||
*/ |
||||
private HttpServletRequest mcpMockRequest() { |
||||
return (HttpServletRequest) Proxy.newProxyInstance(this.getClass().getClassLoader(), |
||||
new Class<?>[]{HttpServletRequest.class}, new MockRequestHandler()); |
||||
} |
||||
|
||||
/** |
||||
* 动态代理的 HttpServletResponse：setHeader 等方法空操作。 |
||||
*/ |
||||
private HttpServletResponse mcpMockResponse() { |
||||
return (HttpServletResponse) Proxy.newProxyInstance(this.getClass().getClassLoader(), |
||||
new Class<?>[]{HttpServletResponse.class}, (proxy, method, args1) -> defaultValue(method.getReturnType())); |
||||
} |
||||
|
||||
/** |
||||
* 按 argInfos 的 paramIndex 填充方法实参数组（数组长度按方法形参总数，未标注解的参数保持 null）。 |
||||
*/ |
||||
private Object[] bindArgs(Map<String, Object> args, List<McpToolArgInfo> argInfos, int parameterCount) { |
||||
Object[] invokeArgs = new Object[parameterCount]; |
||||
for (McpToolArgInfo info : argInfos) { |
||||
invokeArgs[info.getParamIndex()] = this.convertValue(args.get(info.getFieldName()), info); |
||||
} |
||||
return invokeArgs; |
||||
} |
||||
|
||||
/** |
||||
* 按元信息把入参 JSON 值转换为 Java 值（简单类型 / 数组 / 对象递归）。 |
||||
*/ |
||||
private Object convertValue(Object raw, McpToolArgInfo info) { |
||||
if (raw == null) { |
||||
return null; |
||||
} |
||||
switch (info.getType()) { |
||||
case STRING: |
||||
return String.valueOf(raw); |
||||
case BOOLEAN: |
||||
return Boolean.valueOf(raw.toString()); |
||||
case INTEGER: |
||||
return this.toScalar(raw, info.getParamType()); |
||||
case NUMBER: |
||||
return this.toScalar(raw, info.getParamType()); |
||||
case ARRAY: |
||||
return this.toArray(raw, info.getParamType(), info.getElementType()); |
||||
case OBJECT: |
||||
return this.toObject(raw, info.getSchemaClass()); |
||||
default: |
||||
return raw; |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 构造数组实参：支持 List<E> 与 E[]，元素按元素类型转换。 |
||||
*/ |
||||
private Object toArray(Object raw, Class<?> paramType, Class<?> elementType) { |
||||
List<?> source = raw instanceof List ? (List<?>) raw : Arrays.asList((Object[]) raw); |
||||
List<Object> converted = new ArrayList<>(); |
||||
for (Object item : source) { |
||||
converted.add(this.toScalar(item, elementType)); |
||||
} |
||||
if (!paramType.isArray()) { |
||||
return converted; |
||||
} |
||||
Object array = Array.newInstance(elementType, converted.size()); |
||||
for (int i = 0; i < converted.size(); i++) { |
||||
Array.set(array, i, converted.get(i)); |
||||
} |
||||
return array; |
||||
} |
||||
|
||||
/** |
||||
* 构造对象实参：反射 new schemaClass 实例，遍历 @McpToolParam 字段取值转换后 set 入。 |
||||
*/ |
||||
private Object toObject(Object raw, Class<?> schemaClass) { |
||||
if (!(raw instanceof Map) || schemaClass == null) { |
||||
return null; |
||||
} |
||||
Map<?, ?> map = (Map<?, ?>) raw; |
||||
try { |
||||
Object instance = schemaClass.getDeclaredConstructor().newInstance(); |
||||
for (Field field : schemaClass.getDeclaredFields()) { |
||||
McpToolParam ann = field.getAnnotation(McpToolParam.class); |
||||
if (ann == null) { |
||||
continue; |
||||
} |
||||
String fieldName = StrUtil.isNotBlank(ann.fieldName()) ? ann.fieldName() : field.getName(); |
||||
Object value = map.get(fieldName); |
||||
if (value == null) { |
||||
continue; |
||||
} |
||||
McpToolArgInfo fieldInfo = this.toArgInfo(-1, field.getType(), field.getGenericType(), ann, fieldName); |
||||
field.setAccessible(true); |
||||
field.set(instance, this.convertValue(value, fieldInfo)); |
||||
} |
||||
return instance; |
||||
} catch (Exception e) { |
||||
log.error("MCP 对象参数 [{}] 组装失败", schemaClass.getSimpleName(), e); |
||||
return null; |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 标量值按目标 Java 类型转换（String/数值/布尔）。 |
||||
*/ |
||||
private Object toScalar(Object raw, Class<?> type) { |
||||
if (type == String.class) { |
||||
return String.valueOf(raw); |
||||
} |
||||
if (type == Long.class || type == long.class) { |
||||
return Long.valueOf(raw.toString()); |
||||
} |
||||
if (type == Integer.class || type == int.class) { |
||||
return Integer.valueOf(raw.toString()); |
||||
} |
||||
if (type == BigDecimal.class) { |
||||
return new BigDecimal(raw.toString()); |
||||
} |
||||
if (type == Double.class || type == double.class) { |
||||
return Double.valueOf(raw.toString()); |
||||
} |
||||
if (type == Float.class || type == float.class) { |
||||
return Float.valueOf(raw.toString()); |
||||
} |
||||
if (type == Boolean.class || type == boolean.class) { |
||||
return Boolean.valueOf(raw.toString()); |
||||
} |
||||
return raw; |
||||
} |
||||
|
||||
/** |
||||
* 允许抛受检异常的 Supplier（method.invoke 抛 IllegalAccessException/InvocationTargetException）。 |
||||
*/ |
||||
@FunctionalInterface |
||||
private interface ThrowingSupplier<T> { |
||||
T get() throws Exception; |
||||
} |
||||
|
||||
/** |
||||
* 动态代理 request 处理器：getAttribute/setAttribute 落到内存 map，其余方法返回安全默认值。 |
||||
*/ |
||||
private static class MockRequestHandler implements InvocationHandler { |
||||
|
||||
private final Map<String, Object> attributes = new HashMap<>(); |
||||
|
||||
@Override |
||||
public Object invoke(Object proxy, Method method, Object[] args) { |
||||
String name = method.getName(); |
||||
if ("getAttribute".equals(name)) { |
||||
return attributes.get(args[0]); |
||||
} |
||||
if ("setAttribute".equals(name)) { |
||||
attributes.put((String) args[0], args[1]); |
||||
return null; |
||||
} |
||||
if ("getHeader".equals(name)) { |
||||
return attributes.get(args[0]); |
||||
} |
||||
if ("getRequestURI".equals(name)) { |
||||
return "/mcp"; |
||||
} |
||||
return defaultValue(method.getReturnType()); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,47 @@
@@ -0,0 +1,47 @@
|
||||
package apelet.common.core.mcp.controller; |
||||
|
||||
import apelet.common.core.mcp.transport.StreamableHttpServerTransportProvider; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.util.StreamUtils; |
||||
import org.springframework.web.bind.annotation.PostMapping; |
||||
import org.springframework.web.bind.annotation.RequestHeader; |
||||
import org.springframework.web.bind.annotation.RequestMapping; |
||||
import org.springframework.web.bind.annotation.RestController; |
||||
|
||||
import javax.servlet.http.HttpServletRequest; |
||||
import javax.servlet.http.HttpServletResponse; |
||||
import java.io.IOException; |
||||
import java.nio.charset.StandardCharsets; |
||||
|
||||
/** |
||||
* Streamable HTTP 单端点（POST /tenantadmin/mcp）。 |
||||
* 直接读写 HttpServletRequest/Response，序列化由 provider 完成并原样写入， |
||||
* 不经过 Spring 消息转换器（应用 ObjectMapper 会丢 @JsonTypeInfo 的 type 判别符 / 双重编码 JsonNode）。 |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-28 |
||||
*/ |
||||
@RestController |
||||
@RequestMapping("/tenantadmin/mcp") |
||||
public class McpStreamableController { |
||||
|
||||
@Autowired |
||||
private StreamableHttpServerTransportProvider streamableHttpTransport; |
||||
|
||||
@PostMapping |
||||
public void handle(HttpServletRequest request, HttpServletResponse response, |
||||
@RequestHeader(value = "Mcp-Session-Id", required = false) String sessionHeader) throws IOException { |
||||
String body = new String(StreamUtils.copyToByteArray(request.getInputStream()), StandardCharsets.UTF_8); |
||||
StreamableHttpServerTransportProvider.McpProcessResult result = |
||||
streamableHttpTransport.processMcpRequest(body, sessionHeader); |
||||
response.setStatus(result.httpStatus); |
||||
response.setContentType("application/json;charset=UTF-8"); |
||||
if (result.sessionId != null) { |
||||
response.setHeader(StreamableHttpServerTransportProvider.SESSION_HEADER, result.sessionId); |
||||
} |
||||
String json = result.responseJson != null ? result.responseJson : result.errorJson; |
||||
if (json != null) { |
||||
response.getWriter().write(json); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,71 @@
@@ -0,0 +1,71 @@
|
||||
package apelet.common.core.mcp.model; |
||||
|
||||
/** |
||||
* 单个 MCP 工具方法参数/对象字段的元信息：组装 inputSchema 与调用时绑定实参共用。 |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-31 |
||||
*/ |
||||
public class McpToolArgInfo { |
||||
|
||||
/** 参数在方法形参列表中的下标；对象字段为 -1。 */ |
||||
private final int paramIndex; |
||||
/** 入参字段名（注解 fieldName 未配置时为 Java 参数名/属性名）。 */ |
||||
private final String fieldName; |
||||
/** 注解声明的类型（决定 inputSchema 的 type 与转换分支）。 */ |
||||
private final McpToolParamType type; |
||||
/** 字段描述。 */ |
||||
private final String description; |
||||
/** Java 类型（形参或字段类型）。 */ |
||||
private final Class<?> paramType; |
||||
/** ARRAY 类型的元素类型（由泛型实参或数组组件类型推断），非 ARRAY 为 null。 */ |
||||
private final Class<?> elementType; |
||||
/** OBJECT 类型指向的对象类型，非 OBJECT 为 null。 */ |
||||
private final Class<?> schemaClass; |
||||
/** 是否必填（进 inputSchema 的 required 数组）。 */ |
||||
private final boolean required; |
||||
|
||||
public McpToolArgInfo(int paramIndex, String fieldName, McpToolParamType type, String description, |
||||
Class<?> paramType, Class<?> elementType, Class<?> schemaClass, boolean required) { |
||||
this.paramIndex = paramIndex; |
||||
this.fieldName = fieldName; |
||||
this.type = type; |
||||
this.description = description; |
||||
this.paramType = paramType; |
||||
this.elementType = elementType; |
||||
this.schemaClass = schemaClass; |
||||
this.required = required; |
||||
} |
||||
|
||||
public int getParamIndex() { |
||||
return paramIndex; |
||||
} |
||||
|
||||
public String getFieldName() { |
||||
return fieldName; |
||||
} |
||||
|
||||
public McpToolParamType getType() { |
||||
return type; |
||||
} |
||||
|
||||
public String getDescription() { |
||||
return description; |
||||
} |
||||
|
||||
public Class<?> getParamType() { |
||||
return paramType; |
||||
} |
||||
|
||||
public Class<?> getElementType() { |
||||
return elementType; |
||||
} |
||||
|
||||
public Class<?> getSchemaClass() { |
||||
return schemaClass; |
||||
} |
||||
|
||||
public boolean isRequired() { |
||||
return required; |
||||
} |
||||
} |
||||
@ -0,0 +1,35 @@
@@ -0,0 +1,35 @@
|
||||
package apelet.common.core.mcp.model; |
||||
|
||||
import lombok.Getter; |
||||
|
||||
/** |
||||
* MCP 参数类型枚举：value 即 JSON Schema 的 type 值。 |
||||
* <p>Java 侧实际的转换类型由方法参数声明决定，枚举只负责 inputSchema 的 type 与转换分支。</p> |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-31 |
||||
*/ |
||||
@Getter |
||||
public enum McpToolParamType { |
||||
|
||||
/** 字符串 */ |
||||
STRING("string"), |
||||
/** 数字（整数或浮点），对应 BigDecimal/Double/Float */ |
||||
NUMBER("number"), |
||||
/** 整数，对应 Integer/Long */ |
||||
INTEGER("integer"), |
||||
/** 布尔值 */ |
||||
BOOLEAN("boolean"), |
||||
/** 数组，元素类型由 List 泛型实参或数组组件类型推断 */ |
||||
ARRAY("array"), |
||||
/** 对象，需要配合 schemaClass 指定类型 */ |
||||
OBJECT("object"); |
||||
|
||||
/** JSON Schema 的 type 值 */ |
||||
private final String jsonType; |
||||
|
||||
McpToolParamType(String jsonType) { |
||||
this.jsonType = jsonType; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,230 @@
@@ -0,0 +1,230 @@
|
||||
package apelet.common.core.mcp.transport; |
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference; |
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import io.modelcontextprotocol.spec.McpSchema; |
||||
import io.modelcontextprotocol.spec.McpServerSession; |
||||
import io.modelcontextprotocol.spec.McpServerTransport; |
||||
import io.modelcontextprotocol.spec.McpServerTransportProvider; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import reactor.core.publisher.Flux; |
||||
import reactor.core.publisher.Mono; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.Collections; |
||||
import java.util.Map; |
||||
import java.util.UUID; |
||||
import java.util.concurrent.BlockingQueue; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.concurrent.LinkedBlockingQueue; |
||||
import java.util.concurrent.TimeUnit; |
||||
|
||||
/** |
||||
* Streamable HTTP 传输层（MCP spec 2025-03-26 单端点 POST 传输）。 |
||||
* 基于 mcp-java8-sdk 0.8.1 的 SPI（McpServerTransportProvider + McpServerTransport）自研， |
||||
* 因该 fork 不提供 Streamable HTTP Provider（官方 SDK 0.9.0 才引入）。 |
||||
* 会话通过 Mcp-Session-Id 头管理：initialize 创建，后续请求携带。 |
||||
* 响应序列化由调用方（McpStreamableController）直接写入 HTTP 响应， |
||||
* 不经过 Spring 消息转换器，避免应用 ObjectMapper 丢失 @JsonTypeInfo 的 type 判别符 |
||||
* 或对 JsonNode 双重编码。 |
||||
* |
||||
* @author chenchuchuan |
||||
* @date 2026-08-28 |
||||
*/ |
||||
public class StreamableHttpServerTransportProvider implements McpServerTransportProvider { |
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(StreamableHttpServerTransportProvider.class); |
||||
|
||||
public static final String SESSION_HEADER = "Mcp-Session-Id"; |
||||
|
||||
private final ObjectMapper objectMapper; |
||||
private final ConcurrentHashMap<String, McpServerSession> sessions = new ConcurrentHashMap<>(); |
||||
private final ConcurrentHashMap<String, StreamableHttpSessionTransport> transports = new ConcurrentHashMap<>(); |
||||
private volatile boolean isClosing; |
||||
|
||||
private McpServerSession.Factory sessionFactory; |
||||
|
||||
public StreamableHttpServerTransportProvider(ObjectMapper objectMapper) { |
||||
this.objectMapper = objectMapper; |
||||
} |
||||
|
||||
@Override |
||||
public void setSessionFactory(McpServerSession.Factory sessionFactory) { |
||||
this.sessionFactory = sessionFactory; |
||||
} |
||||
|
||||
@Override |
||||
public Mono<Void> notifyClients(String method, Map<String, Object> params) { |
||||
return Flux.fromIterable(sessions.values()) |
||||
.flatMap(session -> session.sendNotification(method, params)) |
||||
.then(); |
||||
} |
||||
|
||||
@Override |
||||
public Mono<Void> closeGracefully() { |
||||
isClosing = true; |
||||
return Flux.fromIterable(sessions.values()) |
||||
.flatMap(McpServerSession::closeGracefully) |
||||
.then(); |
||||
} |
||||
|
||||
/** |
||||
* 处理一条 MCP 请求：JSON-RPC 解析 + Mcp-Session-Id 会话路由 + 业务处理。 |
||||
* 返回序列化好的响应 JSON（含 type 判别符），由调用方原样写入 HTTP 响应。 |
||||
*/ |
||||
public McpProcessResult processMcpRequest(String body, String sessionHeader) { |
||||
if (isClosing) { |
||||
return McpProcessResult.error(503, null, "Server is shutting down"); |
||||
} |
||||
McpSchema.JSONRPCMessage message; |
||||
try { |
||||
message = McpSchema.deserializeJsonRpcMessage(objectMapper, body); |
||||
} catch (Exception e) { |
||||
logger.error("Failed to deserialize MCP message: {}", e.getMessage()); |
||||
return McpProcessResult.error(400, null, "Invalid message format"); |
||||
} |
||||
|
||||
String sessionId = sessionHeader; |
||||
boolean isInitialize = message instanceof McpSchema.JSONRPCRequest |
||||
&& "initialize".equals(((McpSchema.JSONRPCRequest) message).getMethod()); |
||||
|
||||
McpServerSession session; |
||||
StreamableHttpSessionTransport transport; |
||||
if (isInitialize && sessionId == null) { |
||||
sessionId = UUID.randomUUID().toString(); |
||||
transport = new StreamableHttpSessionTransport(sessionId, objectMapper); |
||||
session = sessionFactory.create(transport); |
||||
sessions.put(sessionId, session); |
||||
transports.put(sessionId, transport); |
||||
} else if (sessionId != null) { |
||||
transport = transports.get(sessionId); |
||||
session = sessions.get(sessionId); |
||||
if (session == null || transport == null) { |
||||
return McpProcessResult.error(404, sessionId, "Session not found: " + sessionId); |
||||
} |
||||
} else { |
||||
return McpProcessResult.error(400, null, "Session ID missing in message endpoint"); |
||||
} |
||||
|
||||
logger.info("MCP 请求 method={} sessionId={}", |
||||
message instanceof McpSchema.JSONRPCRequest ? ((McpSchema.JSONRPCRequest) message).getMethod() |
||||
: (message instanceof McpSchema.JSONRPCNotification ? ((McpSchema.JSONRPCNotification) message).getMethod() : "unknown"), |
||||
sessionId); |
||||
|
||||
try { |
||||
session.handle(message).block(Duration.ofSeconds(30)); |
||||
// 协议上客户端应在 initialize 后发送 notifications/initialized 才会进入 INITIALIZED 状态，
|
||||
// 但直接 HTTP 调用（Postman / 平台扫描）往往省略该通知，导致后续 tools/list 等请求
|
||||
// 因 exchange 未就绪而在 exchangeSink.asMono() 上无限阻塞超时，故服务端自动补发。
|
||||
if (isInitialize) { |
||||
session.handle(new McpSchema.JSONRPCNotification("2.0", "notifications/initialized", Collections.emptyMap())) |
||||
.block(Duration.ofSeconds(30)); |
||||
} |
||||
} catch (Exception e) { |
||||
logger.error("Failed to process MCP message: {}", e.getMessage()); |
||||
return McpProcessResult.error(500, sessionId, "Failed to process MCP message: " + e.getMessage()); |
||||
} |
||||
|
||||
McpSchema.JSONRPCMessage response = null; |
||||
if (message instanceof McpSchema.JSONRPCRequest) { |
||||
try { |
||||
response = transport.pollResponse(); |
||||
} catch (InterruptedException e) { |
||||
Thread.currentThread().interrupt(); |
||||
return McpProcessResult.error(500, sessionId, "Interrupted while waiting for response"); |
||||
} |
||||
} |
||||
try { |
||||
return new McpProcessResult(200, sessionId, response != null ? objectMapper.writeValueAsString(response) : null, null); |
||||
} catch (Exception e) { |
||||
logger.error("Failed to serialize MCP response: {}", e.getMessage()); |
||||
return McpProcessResult.error(500, sessionId, "Failed to serialize MCP response: " + e.getMessage()); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* MCP 请求处理结果，供控制器原样写入 HTTP 响应。 |
||||
*/ |
||||
public static class McpProcessResult { |
||||
|
||||
/** HTTP 状态码 */ |
||||
public final int httpStatus; |
||||
/** 会话 id（可能为 null） */ |
||||
public final String sessionId; |
||||
/** 正常响应 JSON（含 type 判别符）；通知类请求为 null */ |
||||
public final String responseJson; |
||||
/** 错误响应 JSON；正常时为 null */ |
||||
public final String errorJson; |
||||
|
||||
McpProcessResult(int httpStatus, String sessionId, String responseJson, String errorJson) { |
||||
this.httpStatus = httpStatus; |
||||
this.sessionId = sessionId; |
||||
this.responseJson = responseJson; |
||||
this.errorJson = errorJson; |
||||
} |
||||
|
||||
static McpProcessResult error(int httpStatus, String sessionId, String message) { |
||||
return new McpProcessResult(httpStatus, sessionId, null, |
||||
"{\"error\":\"" + message.replace("\"", "\\\"") + "\"}"); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* 每会话的传输层：sendMessage 把出站消息放入队列，由 processMcpRequest 取走序列化写回 HTTP 响应。 |
||||
*/ |
||||
public static class StreamableHttpSessionTransport implements McpServerTransport { |
||||
|
||||
private final String sessionId; |
||||
private final ObjectMapper objectMapper; |
||||
private final BlockingQueue<McpSchema.JSONRPCMessage> outboundMessages = new LinkedBlockingQueue<>(); |
||||
private volatile boolean isClosed; |
||||
|
||||
StreamableHttpSessionTransport(String sessionId, ObjectMapper objectMapper) { |
||||
this.sessionId = sessionId; |
||||
this.objectMapper = objectMapper; |
||||
} |
||||
|
||||
@Override |
||||
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) { |
||||
return Mono.fromRunnable(() -> { |
||||
try { |
||||
if (isClosed) { |
||||
logger.warn("Failed to send message to session {}: transport is closed", sessionId); |
||||
return; |
||||
} |
||||
outboundMessages.offer(message); |
||||
logger.debug("Message sent to session {}", sessionId); |
||||
} catch (Exception e) { |
||||
logger.error("Failed to send message to session {}: {}", sessionId, e.getMessage()); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
@Override |
||||
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) { |
||||
return objectMapper.convertValue(data, typeRef); |
||||
} |
||||
|
||||
@Override |
||||
public Mono<Void> closeGracefully() { |
||||
return Mono.fromRunnable(() -> { |
||||
isClosed = true; |
||||
outboundMessages.clear(); |
||||
}); |
||||
} |
||||
|
||||
@Override |
||||
public void close() { |
||||
isClosed = true; |
||||
outboundMessages.clear(); |
||||
} |
||||
|
||||
/** |
||||
* 取出一条出站消息（通常是对当前请求的 JSON-RPC 响应）。 |
||||
*/ |
||||
McpSchema.JSONRPCMessage pollResponse() throws InterruptedException { |
||||
return isClosed ? null : outboundMessages.poll(1, TimeUnit.SECONDS); |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue