增加 Sampling（委托生成）功能

samples/DotNetCampus.SampleMcpServer/McpTools/SampleTool.cs → samples/DotNetCampus.SampleMcpServer/McpTools/EchoDelayTool.cs

@@ -3,7 +3,7 @@
 
 namespace DotNetCampus.SampleMcpServer.McpTools;
 
-public class SampleTool
+public class EchoDelayTool
 {
     /// <summary>
     /// 用于给 AI 调试使用的工具，原样返回一些信息

samples/DotNetCampus.SampleMcpServer/McpTools/SamplingTool.cs

@@ -0,0 +1,57 @@
+using DotNetCampus.ModelContextProtocol.CompilerServices;
+using DotNetCampus.ModelContextProtocol.Exceptions;
+using DotNetCampus.ModelContextProtocol.Protocol.Messages;
+using DotNetCampus.ModelContextProtocol.Servers;
+
+namespace DotNetCampus.SampleMcpServer.McpTools;
+
+public class SamplingTool
+{
+    /// <summary>
+    /// 通过客户端的 LLM 进行采样，将 prompt 发送给客户端，获取 LLM 响应并返回。
+    /// 用于人工验证 sampling/createMessage 协议流程是否正常。
+    /// </summary>
+    /// <param name="prompt">发送给 LLM 的提示词</param>
+    /// <param name="maxTokens">最大生成令牌数</param>
+    /// <param name="systemPrompt">可选的系统提示词</param>
+    /// <param name="context">MCP 工具上下文</param>
+    [McpServerTool]
+    public async Task<CallToolResult> AskLlm(
+        string prompt,
+        int maxTokens = 1024,
+        string? systemPrompt = null,
+        IMcpServerCallToolContext context = null!)
+    {
+        if (!context.Sampling.IsSupported)
+        {
+            return CallToolResult.FromError(
+                "当前客户端未声明 Sampling 能力。请确保客户端支持 sampling/createMessage 请求。\n" +
+                "The connected client has not declared Sampling capability.");
+        }
+
+        try
+        {
+            var result = await context.Sampling.CreateMessageAsync(prompt, maxTokens, systemPrompt, context.CancellationToken);
+
+            var responseText = result.Content switch
+            {
+                TextContentBlock text => text.Text,
+                _ => $"[Non-text content: {result.Content?.GetType().Name}]",
+            };
+
+            return $"""
+                Model: {result.Model}
+                StopReason: {result.StopReason ?? "unknown"}
+                Role: {result.Role}
+                ---
+                {responseText}
+                """;
+        }
+        catch (McpSamplingRejectedException ex)
+        {
+            return CallToolResult.FromError(
+                $"采样请求被用户拒绝。Sampling request was rejected by the user.\n" +
+                $"Code: {ex.ErrorCode}, Message: {ex.RejectionMessage}");
+        }
+    }
+}

samples/DotNetCampus.SampleMcpServer/Program.cs

@@ -32,11 +32,12 @@ private static async Task Main(string[] args)
             .WithRequestHandlers(s => new CustomRequestHandlers(s))
             .WithJsonSerializer(McpToolJsonContext.Default)
             .WithTools(t => t
-                .WithTool(() => new SampleTool())
+                .WithTool(() => new EchoDelayTool())
                 .WithTool(() => new InputTool())
                 .WithTool(() => new OutputTool())
                 .WithTool(() => new PolymorphicTool())
                 .WithTool(() => new ResourceTool())
+                .WithTool(() => new SamplingTool())
             )
             .WithResources(r => r
                 .WithResource(() => new SampleResource())

src/DotNetCampus.ModelContextProtocol.Ipc/Transports/Ipc/IpcServerTransportSession.cs

@@ -1,4 +1,6 @@
-using dotnetCampus.Ipc.Pipes;
+using System.Collections.Concurrent;
+using dotnetCampus.Ipc.Pipes;
+using DotNetCampus.ModelContextProtocol.Protocol.Messages;
 using DotNetCampus.ModelContextProtocol.Protocol.Messages.JsonRpc;
 
 namespace DotNetCampus.ModelContextProtocol.Transports.Ipc;
@@ -8,6 +10,9 @@ namespace DotNetCampus.ModelContextProtocol.Transports.Ipc;
 /// </summary>
 public class IpcServerTransportSession : IServerTransportSession
 {
+    private readonly ConcurrentDictionary<string, TaskCompletionSource<JsonRpcResponse>> _pendingRequests = [];
+    private PeerProxy? _peer;
+
     /// <summary>
     /// 创建 DotNetCampus.Ipc 传输层的一个会话。
     /// </summary>
@@ -22,15 +27,75 @@ public IpcServerTransportSession(string sessionId)
     /// </summary>
     public string SessionId { get; }
 
+    /// <inheritdoc />
+    public ClientCapabilities? ConnectedClientCapabilities { get; set; }
+
+    /// <summary>
+    /// 设置与此会话关联的 IPC 对端代理，用于 SendRequestAsync 发送消息。
+    /// </summary>
+    internal void SetPeer(PeerProxy peer)
+    {
+        _peer = peer;
+    }
+
     /// <inheritdoc />
     public Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
     {
         throw new NotImplementedException();
     }
 
+    /// <inheritdoc />
+    public async Task<JsonRpcResponse> SendRequestAsync(JsonRpcRequest request, CancellationToken cancellationToken = default)
+    {
+        if (request.Id?.ToString() is not { } id)
+        {
+            throw new InvalidOperationException("请求 ID 不能为 null。Request ID must not be null.");
+        }
+
+        var tcs = new TaskCompletionSource<JsonRpcResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+        _pendingRequests[id] = tcs;
+
+        using var registration = cancellationToken.Register(() =>
+        {
+            if (_pendingRequests.TryRemove(id, out var removed))
+            {
+                removed.TrySetCanceled(cancellationToken);
+            }
+        });
+
+        try
+        {
+            await SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
+            return await tcs.Task.ConfigureAwait(false);
+        }
+        finally
+        {
+            _pendingRequests.TryRemove(id, out _);
+        }
+    }
+
+    /// <inheritdoc />
+    public void HandleResponseAsync(JsonRpcResponse response)
+    {
+        if (response.Id?.ToString() is not { } id)
+        {
+            return;
+        }
+
+        if (_pendingRequests.TryRemove(id, out var tcs))
+        {
+            tcs.TrySetResult(response);
+        }
+    }
+
     /// <inheritdoc />
     public ValueTask DisposeAsync()
     {
+        foreach (var (_, tcs) in _pendingRequests)
+        {
+            tcs.TrySetCanceled();
+        }
+        _pendingRequests.Clear();
         return ValueTask.CompletedTask;
     }
 }

src/DotNetCampus.ModelContextProtocol.TouchSocket.Http/Transports/TouchSocket/TouchSocketHttpServerTransport.cs

@@ -6,6 +6,7 @@
 using DotNetCampus.ModelContextProtocol.Protocol;
 using DotNetCampus.ModelContextProtocol.Protocol.Messages.JsonRpc;
 using DotNetCampus.ModelContextProtocol.Servers;
+using DotNetCampus.ModelContextProtocol.Transports;
 using DotNetCampus.ModelContextProtocol.Transports.Http;
 using TouchSocket.Core;
 using TouchSocket.Http;
@@ -257,11 +258,15 @@ private async ValueTask HandleStreamableHttpMessageAsync(HttpContext context, Ca
             }
         }
 
-        JsonRpcRequest? jsonRpcRequest;
+        var sessionIdStr = request.Headers.Get(SessionIdHeader).First;
+
+        // 将 body 读取并解析为 JsonDocument，通过 JsonElement 检测消息类型。
+        ReadOnlyMemory<byte> bodyBytes;
+        JsonDocument bodyDoc;
         try
         {
-            var bodyBytes = await request.GetContentAsync();
-            jsonRpcRequest = await _manager.ReadRequestAsync(bodyBytes);
+            bodyBytes = await request.GetContentAsync();
+            bodyDoc = JsonDocument.Parse(bodyBytes);
         }
         catch (JsonException)
         {
@@ -270,71 +275,125 @@ private async ValueTask HandleStreamableHttpMessageAsync(HttpContext context, Ca
             return;
         }
 
-        if (jsonRpcRequest == null)
+        using (bodyDoc)
         {
-            Log.Warn($"[McpServer][TouchSocket] POST request rejected: Empty body.");
-            await context.RespondHttpError(HttpStatusCode.BadRequest, "Empty body");
-            return;
-        }
+            var bodyElement = bodyDoc.RootElement;
 
-        var isInitialize = jsonRpcRequest.Method == RequestMethods.Initialize;
-        var sessionIdStr = request.Headers.Get(SessionIdHeader).First;
-        TouchSocketHttpServerTransportSession? session;
+            // 检测是 JSON-RPC 请求（有 method）还是响应（无 method，有 result 或 error）。
+            var isResponse = !bodyElement.TryGetProperty("method", out _)
+                && (bodyElement.TryGetProperty("result", out _) || bodyElement.TryGetProperty("error", out _));
 
-        if (isInitialize)
-        {
-            // 初始化请求，创建新 Session
-            var newSessionId = _manager.MakeNewSessionId();
-            var newSession = new TouchSocketHttpServerTransportSession(_manager, newSessionId.Id);
+            if (isResponse)
+            {
+                // 客户端响应服务器发起的请求（如 sampling/createMessage）。
+                if (string.IsNullOrEmpty(sessionIdStr) || !_sessions.TryGetValue(sessionIdStr, out var responseSession))
+                {
+                    Log.Warn($"[McpServer][TouchSocket] Response routing failed: Session not found. SessionId={sessionIdStr}");
+                    await context.RespondHttpError(HttpStatusCode.NotFound, "Session not found");
+                    return;
+                }
+
+                JsonRpcResponse? jsonRpcResponse;
+                try
+                {
+                    jsonRpcResponse = await _manager.ReadResponseAsync(bodyBytes);
+                }
+                catch (JsonException)
+                {
+                    await context.RespondHttpError(HttpStatusCode.BadRequest, "Invalid JSON response");
+                    return;
+                }
+
+                if (jsonRpcResponse is not null)
+                {
+                    responseSession.HandleResponseAsync(jsonRpcResponse);
+                }
+
+                await context.RespondHttpSuccess(HttpStatusCode.Accepted);
+                return;
+            }
 
-            if (_sessions.TryAdd(newSessionId.Id, newSession))
+            JsonRpcRequest? jsonRpcRequest;
+            try
             {
-                session = newSession;
-                _manager.Add(session);
-                context.Response.Headers.Add(SessionIdHeader, newSessionId.Id);
-                Log.Info($"[McpServer][TouchSocket] Session created. SessionId={newSessionId.Id}");
+                jsonRpcRequest = await _manager.ReadRequestAsync(bodyBytes);
             }
-            else
+            catch (JsonException)
             {
-                Log.Error($"[McpServer][TouchSocket] Session ID collision. SessionId={newSessionId.Id}");
-                await context.RespondHttpError(HttpStatusCode.InternalServerError, "Session ID collision");
+                Log.Warn($"[McpServer][TouchSocket] POST request rejected: Invalid JSON.");
+                await context.RespondHttpError(HttpStatusCode.BadRequest, "Invalid JSON");
                 return;
             }
-        }
-        else
-        {
-            if (string.IsNullOrEmpty(sessionIdStr))
+
+            if (jsonRpcRequest == null)
             {
-                Log.Warn($"[McpServer][TouchSocket] POST request rejected: Missing Mcp-Session-Id header. Method={jsonRpcRequest.Method}");
-                await context.RespondHttpError(HttpStatusCode.BadRequest, "Missing Mcp-Session-Id header");
+                Log.Warn($"[McpServer][TouchSocket] POST request rejected: Empty body.");
+                await context.RespondHttpError(HttpStatusCode.BadRequest, "Empty body");
                 return;
             }
 
-            if (!_sessions.TryGetValue(sessionIdStr, out session))
+            var isInitialize = jsonRpcRequest.Method == RequestMethods.Initialize;
+            TouchSocketHttpServerTransportSession? session;
+
+            if (isInitialize)
             {
-                Log.Warn($"[McpServer][TouchSocket] POST request rejected: Session not found. SessionId={sessionIdStr}, Method={jsonRpcRequest.Method}");
-                await context.RespondHttpError(HttpStatusCode.NotFound, "Session not found");
-                return;
+                // 初始化请求，创建新 Session
+                var newSessionId = _manager.MakeNewSessionId();
+                var newSession = new TouchSocketHttpServerTransportSession(_manager, newSessionId.Id);
+
+                if (_sessions.TryAdd(newSessionId.Id, newSession))
+                {
+                    session = newSession;
+                    _manager.Add(session);
+                    context.Response.Headers.Add(SessionIdHeader, newSessionId.Id);
+                    Log.Info($"[McpServer][TouchSocket] Session created. SessionId={newSessionId.Id}");
+                }
+                else
+                {
+                    Log.Error($"[McpServer][TouchSocket] Session ID collision. SessionId={newSessionId.Id}");
+                    await context.RespondHttpError(HttpStatusCode.InternalServerError, "Session ID collision");
+                    return;
+                }
+            }
+            else
+            {
+                if (string.IsNullOrEmpty(sessionIdStr))
+                {
+                    Log.Warn($"[McpServer][TouchSocket] POST request rejected: Missing Mcp-Session-Id header. Method={jsonRpcRequest.Method}");
+                    await context.RespondHttpError(HttpStatusCode.BadRequest, "Missing Mcp-Session-Id header");
+                    return;
+                }
+
+                if (!_sessions.TryGetValue(sessionIdStr, out session))
+                {
+                    Log.Warn($"[McpServer][TouchSocket] POST request rejected: Session not found. SessionId={sessionIdStr}, Method={jsonRpcRequest.Method}");
+                    await context.RespondHttpError(HttpStatusCode.NotFound, "Session not found");
+                    return;
+                }
             }
-        }
 
-        Log.Debug($"[McpServer][TouchSocket] Handling JSON-RPC request. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
+            Log.Debug($"[McpServer][TouchSocket] Handling JSON-RPC request. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
 
-        var jsonRpcResponse = await _manager.HandleRequestAsync(jsonRpcRequest,
-            s => s.AddHttpTransportServices(session.SessionId, request),
-            cancellationToken: cancellationToken);
+            var jsonRpcResponse2 = await _manager.HandleRequestAsync(jsonRpcRequest,
+                s =>
+                {
+                    s.AddHttpTransportServices(session.SessionId, request);
+                    s.AddTransportSession(session);
+                },
+                cancellationToken: cancellationToken);
 
-        if (jsonRpcResponse != null)
-        {
-            // Request: Success or Failed.
-            Log.Debug($"[McpServer][TouchSocket] Sending JSON-RPC response. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
-            await context.RespondJsonRpcAsync(_manager, HttpStatusCode.OK, jsonRpcResponse);
-        }
-        else
-        {
-            // Notification: No need to respond.
-            Log.Debug($"[McpServer][TouchSocket] No response for notification. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
-            await context.RespondHttpSuccess(HttpStatusCode.Accepted);
+            if (jsonRpcResponse2 != null)
+            {
+                // Request: Success or Failed.
+                Log.Debug($"[McpServer][TouchSocket] Sending JSON-RPC response. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
+                await context.RespondJsonRpcAsync(_manager, HttpStatusCode.OK, jsonRpcResponse2);
+            }
+            else
+            {
+                // Notification: No need to respond.
+                Log.Debug($"[McpServer][TouchSocket] No response for notification. SessionId={session.SessionId}, Method={jsonRpcRequest.Method}, MessageId={jsonRpcRequest.Id}");
+                await context.RespondHttpSuccess(HttpStatusCode.Accepted);
+            }
         }
     }