mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 09:28:14 +00:00
51eb98e0ae
openai adapter transform_stream_chunk:
- OpenAI 流式分片是嵌套格式 function.{name,arguments}，原样透传后
json.Unmarshal 到扁平 ToolCall{name,arguments} 时 name 恒为空，
accumulateStream flushToolCall 因 acc.name=="" 静默丢弃整个工具调用
（流式路径自上线起 tool_calls 全部丢失的根因）
- 现在正确解包 function.name → name，function.arguments → raw_arguments
（保留原始 JSON 字符串分片，由 accumulateStream 按 index 拼接）
- 不按 name 过滤分片：OpenAI 流式续传块 name 为空但携带 arguments 分片
mcp stdio/sse transport:
- stdio Send() 无超时：server 进程卡死时插件加载永久阻塞
- sse http.Client 无超时：远程 server 网络抖动/无响应时永久阻塞，
导致 webui 等后续插件全部无法启动（生产实例偶发启动卡死根因）
- stdio 加 60s 请求超时；sse client 加 30s 整体 + 10s 拨号超时
116 lines
3.0 KiB
Go
116 lines
3.0 KiB
Go
package mcp
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// SSETransport 通过 HTTP POST 进行 JSON-RPC 通信。
|
||
// 兼容两种服务端响应：application/json 直连响应，以及 Streamable HTTP
|
||
// 的异步响应（HTTP 202 + text/event-stream 的 SSE data 帧）。
|
||
type SSETransport struct {
|
||
url string
|
||
client *http.Client
|
||
}
|
||
|
||
func NewSSETransport(url string) *SSETransport {
|
||
// 必须带超时：远程 MCP server 网络抖动/无响应时，
|
||
// 无超时的 client 会让插件加载永久阻塞（webui 等后续插件全部起不来）
|
||
return &SSETransport{
|
||
url: url,
|
||
client: &http.Client{
|
||
Timeout: 30 * time.Second,
|
||
Transport: &http.Transport{
|
||
DialContext: (&net.Dialer{
|
||
Timeout: 10 * time.Second,
|
||
KeepAlive: 30 * time.Second,
|
||
}).DialContext,
|
||
},
|
||
},
|
||
}
|
||
}
|
||
|
||
// parseResponseBody 根据 Content-Type 解析 JSON-RPC 响应
|
||
func parseResponseBody(contentType string, body []byte) (*rpcResponse, error) {
|
||
if strings.Contains(contentType, "text/event-stream") {
|
||
// SSE 流：逐行提取 data: 帧
|
||
var lastJSON []byte
|
||
for _, line := range strings.Split(string(body), "\n") {
|
||
line = strings.TrimRight(line, "\r")
|
||
if strings.HasPrefix(line, "data:") {
|
||
data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||
if data == "" || data == "[DONE]" {
|
||
continue
|
||
}
|
||
var frame map[string]json.RawMessage
|
||
if err := json.Unmarshal([]byte(data), &frame); err == nil {
|
||
if _, isResp := frame["id"]; isResp || frame["result"] != nil || frame["error"] != nil {
|
||
lastJSON = []byte(data)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if len(lastJSON) == 0 {
|
||
return nil, fmt.Errorf("SSE 流中未找到 JSON-RPC 响应帧: %s", truncate(string(body), 300))
|
||
}
|
||
body = lastJSON
|
||
}
|
||
|
||
var rpcResp rpcResponse
|
||
if err := json.Unmarshal(body, &rpcResp); err != nil {
|
||
return nil, fmt.Errorf("unmarshal response: %w, body=%s", err, truncate(string(body), 300))
|
||
}
|
||
return &rpcResp, nil
|
||
}
|
||
|
||
func truncate(s string, n int) string {
|
||
if len(s) > n {
|
||
return s[:n] + "..."
|
||
}
|
||
return s
|
||
}
|
||
|
||
func (t *SSETransport) Send(req *rpcRequest) (*rpcResponse, error) {
|
||
data, err := json.Marshal(req)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("marshal: %w", err)
|
||
}
|
||
|
||
httpReq, err := http.NewRequest("POST", t.url, bytes.NewReader(data))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("http request: %w", err)
|
||
}
|
||
httpReq.Header.Set("Content-Type", "application/json")
|
||
httpReq.Header.Set("Accept", "application/json, text/event-stream")
|
||
|
||
resp, err := t.client.Do(httpReq)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("http post: %w", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
body, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("read body: %w", err)
|
||
}
|
||
|
||
ct := resp.Header.Get("Content-Type")
|
||
// 非 2xx：尝试提取错误信息
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
return nil, fmt.Errorf("http status %d: %s", resp.StatusCode, truncate(string(body), 300))
|
||
}
|
||
|
||
return parseResponseBody(ct, body)
|
||
}
|
||
|
||
func (t *SSETransport) Close() error {
|
||
t.client.CloseIdleConnections()
|
||
return nil
|
||
}
|