mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 01:18:08 +00:00
fix(mcp+adapter): 流式 tool_calls 解析修复 + MCP transport 超时保护
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 拨号超时
This commit is contained in:
2
.gitignore
vendored
2
.gitignore
vendored
@ -36,7 +36,7 @@ third_party/homeagent-sdk/example/
|
||||
codegraph.json
|
||||
|
||||
# 运行时产物(不提交)
|
||||
adapters/
|
||||
/adapters/
|
||||
knowledge/
|
||||
memory/
|
||||
scripts/
|
||||
|
||||
@ -99,7 +99,28 @@ function adapter.transform_stream_chunk(raw_chunk)
|
||||
unified.reasoning_content = delta.reasoning_content
|
||||
end
|
||||
if delta.tool_calls then
|
||||
unified.tool_calls = delta.tool_calls
|
||||
local tcs = {}
|
||||
for _, tc in ipairs(delta.tool_calls) do
|
||||
-- OpenAI 流式格式: {function:{name,arguments}, id, type, index}
|
||||
-- homed StreamChunk.ToolCalls 期望扁平格式: {id, type, name, raw_arguments}
|
||||
local fn = tc["function"]
|
||||
local name = (type(fn) == "table" and fn.name) or tc.name or ""
|
||||
local raw_args = ""
|
||||
if type(fn) == "table" and type(fn.arguments) == "string" then
|
||||
raw_args = fn.arguments
|
||||
elseif type(tc.arguments) == "string" then
|
||||
raw_args = tc.arguments
|
||||
end
|
||||
-- 不能按 name 过滤:OpenAI 流式分片中后续块 name 为空但携带 arguments
|
||||
-- accumulateStream 按 index 累积并在 flushToolCall 时校验 name
|
||||
table.insert(tcs, {
|
||||
id = tc.id or "",
|
||||
type = tc.type or "function",
|
||||
name = name,
|
||||
raw_arguments = raw_args
|
||||
})
|
||||
end
|
||||
unified.tool_calls = tcs
|
||||
end
|
||||
return json.encode(unified)
|
||||
end
|
||||
|
||||
@ -5,8 +5,10 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// SSETransport 通过 HTTP POST 进行 JSON-RPC 通信。
|
||||
@ -18,9 +20,19 @@ type SSETransport struct {
|
||||
}
|
||||
|
||||
func NewSSETransport(url string) *SSETransport {
|
||||
// 必须带超时:远程 MCP server 网络抖动/无响应时,
|
||||
// 无超时的 client 会让插件加载永久阻塞(webui 等后续插件全部起不来)
|
||||
return &SSETransport{
|
||||
url: url,
|
||||
client: &http.Client{},
|
||||
url: url,
|
||||
client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
DialContext: (&net.Dialer{
|
||||
Timeout: 10 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
}).DialContext,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -8,6 +8,7 @@ import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// StdioTransport 通过子进程 stdin/stdout 进行 JSON-RPC 通信
|
||||
@ -105,11 +106,22 @@ func (t *StdioTransport) Send(req *rpcRequest) (*rpcResponse, error) {
|
||||
return nil, fmt.Errorf("write newline: %w", err)
|
||||
}
|
||||
|
||||
// 超时保护:MCP server 进程启动慢/卡死不响应时,不能让插件加载永久阻塞
|
||||
// (stdio server 冷启动如 npx 拉包可能数十秒,给 60s)
|
||||
timeout := time.After(60 * time.Second)
|
||||
select {
|
||||
case resp := <-ch:
|
||||
return resp, nil
|
||||
case <-t.done:
|
||||
t.mu.Lock()
|
||||
delete(t.pending, req.ID)
|
||||
t.mu.Unlock()
|
||||
return nil, fmt.Errorf("mcp transport closed")
|
||||
case <-timeout:
|
||||
t.mu.Lock()
|
||||
delete(t.pending, req.ID)
|
||||
t.mu.Unlock()
|
||||
return nil, fmt.Errorf("mcp request timeout (60s), method may be hung")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user