.NET：后端管道 – ModelRiver Docs

概述

ASP.NET Core 的 minimal APIs 和 System.Threading.Channels 为处理 ModelRiver 后端管道 Webhooks 提供了一个高性能的管道。使用中间件进行签名验证，使用通道 (Channels) 进行后台处理，并使用 HttpClient 进行回调。

您将构建的内容：

一个用于接收 Webhooks 的 minimal API 端点
HMAC 签名验证中间件
使用 IHostedService 和 Channel<T> 进行后台处理
利用 HttpClient 向 ModelRiver 发起回调

快速开始

创建项目

Bash

dotnet new web -n MyAiApp
cd MyAiApp

配置

JSON

1// appsettings.json
2{
3  "ModelRiver": {
4    "ApiKey": "mr_live_YOUR_API_KEY",
5    "WebhookSecret": "your_webhook_secret",
6    "BaseUrl": "https://api.modelriver.com"
7  }
8}

CSHARP

1// ModelRiverOptions.cs
2public class ModelRiverOptions
3{
4    public string ApiKey { get; set; } = "";
5    public string WebhookSecret { get; set; } = "";
6    public string BaseUrl { get; set; } = "https://api.modelriver.com";
7}

Webhook 端点

CSHARP

1// Program.cs
2using System.Security.Cryptography;
3using System.Text;
4using System.Text.Json;
5using System.Threading.Channels;
6 
7var builder = WebApplication.CreateBuilder(args);
8 
9builder.Services.Configure<ModelRiverOptions>(
10    builder.Configuration.GetSection("ModelRiver"));
11 
12builder.Services.AddHttpClient("ModelRiver");
13 
14// 后台处理通道
15var channel = Channel.CreateUnbounded<WebhookJob>();
16builder.Services.AddSingleton(channel);
17builder.Services.AddHostedService<WebhookProcessor>();
18 
19var app = builder.Build();
20 
21app.MapPost("/webhooks/modelriver", async (
22    HttpContext context,
23    Channel<WebhookJob> jobChannel,
24    IConfiguration config) =>
25{
26    // 1. 读取原始 Body
27    using var reader = new StreamReader(context.Request.Body);
28    var rawBody = await reader.ReadToEndAsync();
29 
30    // 2. 验证签名
31    var signature = context.Request.Headers["x-modelriver-signature"].FirstOrDefault() ?? "";
32    var timestamp = context.Request.Headers["x-modelriver-timestamp"].FirstOrDefault() ?? "";
33    var secret = config["ModelRiver:WebhookSecret"] ?? "";
34 
35    // 签名字符串："<timestamp>.<原始请求体中 "data" 值的 JSON 文本>"（切勿重新编码）
36    var start = rawBody.IndexOf("\"data\":", StringComparison.Ordinal);
37    var end = rawBody.LastIndexOf(",\"timestamp\":", StringComparison.Ordinal);
38    if (start == -1 || end < start + 7)
39        return Results.Unauthorized();
40    var data = rawBody.Substring(start + 7, end - start - 7);
41 
42    using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(secret));
43    var hash = hmac.ComputeHash(Encoding.UTF8.GetBytes($"{timestamp}.{data}"));
44    var expected = Convert.ToHexString(hash).ToLowerInvariant();
45 
46    if (!CryptographicOperations.FixedTimeEquals(
47        Encoding.UTF8.GetBytes(expected),
48        Encoding.UTF8.GetBytes(signature)))
49    {
50        return Results.Unauthorized();
51    }
52 
53    // 3. 解析负载：事件载荷位于 "data" 之下
54    var payload = JsonSerializer.Deserialize<WebhookEnvelope>(rawBody,
55        new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower })?.Data;
56 
57    if (payload is null)
58        return Results.BadRequest();
59 
60    // 4. 处理后端管道的工作流
61    if (payload.Type == "task.ai_generated" && payload.CallbackUrl is not null)
62    {
63        await jobChannel.Writer.WriteAsync(new WebhookJob
64        {
65            Event = payload.Event ?? "",
66            AiResponse = payload.AiResponse,
67            CallbackUrl = payload.CallbackUrl,
68            CustomerData = payload.CustomerData ?? new(),
69        });
70 
71        return Results.Ok(new { received = true });
72    }
73 
74    return Results.Ok(new { received = true });
75});
76 
77app.Run();
78 
79// --- 模型 ---
80 
81public record WebhookEnvelope
82{
83    public string? CallbackUrl { get; init; }
84    public string ChannelId { get; init; } = "";
85    public WebhookPayload? Data { get; init; }
86    public long Timestamp { get; init; }
87}
88 
89public record WebhookPayload
90{
91    public string Type { get; init; } = "";
92    public string? Event { get; init; }
93    public string ChannelId { get; init; } = "";
94    public AiResponseData? AiResponse { get; init; }
95    public string? CallbackUrl { get; init; }
96    public bool? CallbackRequired { get; init; }
97    public Dictionary<string, object>? CustomerData { get; init; }
98}
99 
100public record AiResponseData
101{
102    public Dictionary<string, object> Data { get; init; } = new();
103}
104 
105public record WebhookJob
106{
107    public string Event { get; init; } = "";
108    public AiResponseData? AiResponse { get; init; }
109    public string CallbackUrl { get; init; } = "";
110    public Dictionary<string, object> CustomerData { get; init; } = new();
111}

后台处理器

CSHARP

1// WebhookProcessor.cs
2using System.Text.Json;
3using System.Threading.Channels;
4using Microsoft.Extensions.Options;
5 
6public class WebhookProcessor : BackgroundService
7{
8    private readonly Channel<WebhookJob> _channel;
9    private readonly IHttpClientFactory _httpClientFactory;
10    private readonly ModelRiverOptions _options;
11    private readonly ILogger<WebhookProcessor> _logger;
12 
13    public WebhookProcessor(
14        Channel<WebhookJob> channel,
15        IHttpClientFactory httpClientFactory,
16        IOptions<ModelRiverOptions> options,
17        ILogger<WebhookProcessor> logger)
18    {
19        _channel = channel;
20        _httpClientFactory = httpClientFactory;
21        _options = options.Value;
22        _logger = logger;
23    }
24 
25    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
26    {
27        await foreach (var job in _channel.Reader.ReadAllAsync(stoppingToken))
28        {
29            try
30            {
31                await ProcessJobAsync(job, stoppingToken);
32            }
33            catch (Exception ex)
34            {
35                _logger.LogError(ex, "为事件处理 Webhook 任务失败: {Event}", job.Event);
36            }
37        }
38    }
39 
40    private async Task ProcessJobAsync(WebhookJob job, CancellationToken ct)
41    {
42        var client = _httpClientFactory.CreateClient("ModelRiver");
43        client.DefaultRequestHeaders.Add("Authorization", $"Bearer {_options.ApiKey}");
44 
45        var enrichedData = job.AiResponse?.Data ?? new Dictionary<string, object>();
46 
47        // 您的自定义业务逻辑
48        if (job.Event == "content_ready")
49        {
50            enrichedData["processed"] = true;
51            enrichedData["saved_at"] = DateTime.UtcNow.ToString("O");
52        }
53 
54        // 回调 ModelRiver
55        var callback = new
56        {
57            data = enrichedData,
58            task_id = $"dotnet_{job.Event}_{DateTimeOffset.UtcNow.ToUnixTimeSeconds()}",
59            metadata = new
60            {
61                processed_by = "dotnet",
62                processed_at = DateTime.UtcNow.ToString("O"),
63            },
64        };
65 
66        var content = new StringContent(
67            JsonSerializer.Serialize(callback),
68            System.Text.Encoding.UTF8,
69            "application/json");
70 
71        var response = await client.PostAsync(job.CallbackUrl, content, ct);
72 
73        if (response.IsSuccessStatusCode)
74        {
75            _logger.LogInformation("✅ 已针对事件发送回调: {Event}", job.Event);
76        }
77        else
78        {
79            _logger.LogError("❌ 回调失败 ({Status})", response.StatusCode);
80 
81            // 发送错误回调
82            var errorContent = new StringContent(
83                JsonSerializer.Serialize(new { data = new { processing_status = "failed", processing_error = $"HTTP {response.StatusCode}" } }),
84                System.Text.Encoding.UTF8,
85                "application/json");
86 
87            await client.PostAsync(job.CallbackUrl, errorContent, ct);
88        }
89    }
90}

触发异步请求

CSHARP

1// ModelRiverService.cs
2public class ModelRiverService
3{
4    private readonly HttpClient _client;
5    private readonly ModelRiverOptions _options;
6 
7    public ModelRiverService(IHttpClientFactory factory, IOptions<ModelRiverOptions> options)
8    {
9        _client = factory.CreateClient("ModelRiver");
10        _options = options.Value;
11        _client.DefaultRequestHeaders.Add("Authorization", $"Bearer {_options.ApiKey}");
12    }
13 
14    public async Task<AsyncAiResponse> TriggerAsync(string workflow, string prompt)
15    {
16        var response = await _client.PostAsJsonAsync($"{_options.BaseUrl}/v1/ai/async", new
17        {
18            model = workflow,
19            messages = new[] { new { role = "user", content = prompt } },
20        });
21 
22        response.EnsureSuccessStatusCode();
23        return await response.Content.ReadFromJsonAsync<AsyncAiResponse>()
24            ?? throw new Exception("解析响应失败");
25    }
26}
27 
28public record AsyncAiResponse(
29    string ChannelId,
30    string WsToken,
31    string WebsocketUrl,
32    string WebsocketChannel
33);

最佳实践

使用 Channel<T>：用于后台处理的高性能进程内队列。
使用 CryptographicOperations.FixedTimeEquals：用于签名的恒定时间比较。
使用 IHttpClientFactory：适当的连接池和生命周期管理。
使用 IHostedService：随应用程序启动的长运行后台处理器。
读取原始 Body：在模型绑定之前读取流以进行签名验证。

下一步

返回后端框架：所有框架指南
Webhooks 参考：重试策略和投递监控
后端管道概述：架构和流程

.NET 的后端管道

概述

快速开始

创建项目

配置

Webhook 端点

后台处理器

触发异步请求

最佳实践

下一步