AskaEth
76263d42ed
① light.go 的 2 处 "light judge: " + d.Reason 与 engage.go 同形（model reason → 返回值 → coalesce_prepare 的 Debug）。把上一轮的 engageReason 提成包内共用的 judgeReason（judge_reason.go，RedactTextLimited + strings.Fields，60 rune）， engage.go 7 处 + light.go 2 处 + notice.go 的 "notice review failed: "+err.Error() 统一走它。控制流零改动：只换「进去的那份文本」，判定条件与返回值拼接结构未动。 ② MCP 的 Logger 面先探针后修：go-sdk v1.8.0 的 keepalive 会把对端 JSON-RPC error 原样打进日志（实测泄漏，见 sdklog_probe_test.go 的输出），tools/call 路径不打日志。 新增 sdklog.go：redactingLogger 交给 SDK，wire 错误（链上有 *jsonrpc.Error）只留 code + message_bytes（复用 upstreamError），本地错误保留原因，消息/属性压平+打码+300 rune。 ③ errortoken.go 的口径注释纠正：从「包外不得调用」改为按字段类别判定（协议枚举可守、 自由文本默认不带），并把 cdp.go 这条 M51 用法作为例子写进去。
215 lines
7.3 KiB
Go
215 lines
7.3 KiB
Go
package mcp
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
|
||
sdkjsonrpc "github.com/modelcontextprotocol/go-sdk/jsonrpc"
|
||
mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||
)
|
||
|
||
// Connect 选择传输（未注入时）、建立官方 SDK 会话并完成 initialize 握手。
|
||
// 幂等：已连接时直接返回 nil；已关闭时返回 ErrClosed。
|
||
func (c *Client) Connect(ctx context.Context) error {
|
||
same, err := c.begin()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if same {
|
||
return nil
|
||
}
|
||
|
||
tr := c.tr
|
||
if tr == nil {
|
||
if tr, err = newTransport(c.cfg); err != nil {
|
||
return fmt.Errorf("mcp connect %q: %w", c.cfg.Name, err)
|
||
}
|
||
}
|
||
|
||
session, err := c.newSDKClient().Connect(ctx, tr, c.sessionOptions())
|
||
if err != nil {
|
||
return fmt.Errorf("mcp connect %q: %w", c.cfg.Name, wrapConnectErr(err))
|
||
}
|
||
info := serverInfoFrom(session, c.hostFeatures())
|
||
c.mu.Lock()
|
||
c.session = session
|
||
c.tr = tr
|
||
c.state = stateConnected
|
||
c.info = info
|
||
c.mu.Unlock()
|
||
|
||
c.log.Debug("mcp connected", "name", c.cfg.Name, "server", info.Name,
|
||
"version", info.Version, "protocol", info.ProtocolVersion,
|
||
"tools_declared", info.Features.Tools)
|
||
return nil
|
||
}
|
||
|
||
// Close 关闭 SDK 会话并标记客户端已关闭。幂等。
|
||
func (c *Client) Close() error {
|
||
c.mu.Lock()
|
||
if c.state == stateClosed {
|
||
c.mu.Unlock()
|
||
return nil
|
||
}
|
||
session := c.session
|
||
c.session = nil
|
||
c.state = stateClosed
|
||
c.mu.Unlock()
|
||
|
||
if session != nil {
|
||
if err := session.Close(); err != nil {
|
||
return fmt.Errorf("mcp close %q: %w", c.cfg.Name, upstreamError(err))
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ListTools 返回服务器暴露的工具（SDK 自动翻页）。
|
||
// 服务器在 initialize 里没有声明 tools 能力族时拒绝调用、不发出 RPC：
|
||
// design/09 §1.1「声明了才算可用，未声明不得调用」。
|
||
func (c *Client) ListTools(ctx context.Context) ([]ToolInfo, error) {
|
||
session := c.connected()
|
||
if session == nil {
|
||
return nil, fmt.Errorf("mcp list tools %q: %w", c.cfg.Name, ErrNotConnected)
|
||
}
|
||
if !c.Features().Tools {
|
||
return nil, fmt.Errorf("mcp list tools %q: %w", c.cfg.Name, ErrCapabilityNotDeclared)
|
||
}
|
||
out := make([]ToolInfo, 0, 8)
|
||
for tool, err := range session.Tools(ctx, nil) {
|
||
if err != nil {
|
||
return nil, fmt.Errorf("mcp list tools %q: %w", c.cfg.Name, upstreamError(err))
|
||
}
|
||
out = append(out, toolInfoFrom(tool))
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// CallTool 调用服务器上的工具。args 必须是 JSON 对象（map[string]any）。
|
||
// 调用前用 schema 缓存里的 input schema 校验参数并补默认值。
|
||
func (c *Client) CallTool(ctx context.Context, name string, args any) (ToolResult, error) {
|
||
arguments, err := normalizeArgs(args)
|
||
if err != nil {
|
||
return ToolResult{}, fmt.Errorf("mcp call tool %q: %w", name, err)
|
||
}
|
||
return c.callTool(ctx, name, arguments, nil)
|
||
}
|
||
|
||
// callTool 是 CallTool 的内部实现，fallback 是缓存未命中时可用的 input schema
|
||
// （来自 tools/list 的发现结果）。
|
||
func (c *Client) callTool(ctx context.Context, name string, args map[string]any, fallback map[string]any) (ToolResult, error) {
|
||
session := c.connected()
|
||
if session == nil {
|
||
return ToolResult{}, fmt.Errorf("mcp call tool %q: %w", name, ErrNotConnected)
|
||
}
|
||
prepared, err := c.schemas.Prepare(c.cfg.Name, name, fallback, args)
|
||
if err != nil {
|
||
return ToolResult{}, fmt.Errorf("mcp call tool %q: %w", name, err)
|
||
}
|
||
res, err := session.CallTool(ctx, &mcpsdk.CallToolParams{Name: name, Arguments: prepared})
|
||
if err != nil {
|
||
return ToolResult{}, fmt.Errorf("mcp call tool %q on %q: %w", name, c.cfg.Name, upstreamError(err))
|
||
}
|
||
return ToolResult{
|
||
Content: toContentBlocks(res.Content),
|
||
Structured: res.StructuredContent,
|
||
IsError: res.IsError,
|
||
}, nil
|
||
}
|
||
|
||
// DiscoverTools 列出服务器工具，把每把工具包装成 sdk.Tool 并把 input schema
|
||
// 放进缓存，使 MCP server 看起来就是一个普通工具提供者。
|
||
func (c *Client) DiscoverTools(ctx context.Context) ([]sdk.Tool, error) {
|
||
tools, err := c.ListTools(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out := make([]sdk.Tool, 0, len(tools))
|
||
for _, t := range tools {
|
||
c.cacheSchema(t)
|
||
out = append(out, &mcpTool{client: c, info: t})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// cacheSchema 把一次发现到的 input schema 解析并放进缓存；解析失败只降级为
|
||
// 「调用时不校验」，不影响工具接入。
|
||
func (c *Client) cacheSchema(t ToolInfo) {
|
||
if err := c.schemas.Put(c.cfg.Name, t.Name, t.InputSchema); err != nil {
|
||
c.log.Warn("mcp tool input schema not resolvable; validation skipped",
|
||
"name", c.cfg.Name, "tool", t.Name, "err", err)
|
||
}
|
||
}
|
||
|
||
// handleToolListChanged 是 SDK 的 notifications/tools/list_changed 回调：
|
||
// 先让 schema 缓存失效（design/09 §1.1），再合并触发宿主的增量刷新。
|
||
// 它可能运行在 SDK 的读路径上，因此真正的刷新放到新 goroutine 里。
|
||
func (c *Client) handleToolListChanged(context.Context, *mcpsdk.ToolListChangedRequest) {
|
||
c.schemas.Invalidate(c.cfg.Name)
|
||
if c.onToolListChanged == nil {
|
||
return
|
||
}
|
||
if !c.refreshPending.CompareAndSwap(false, true) {
|
||
return
|
||
}
|
||
go func() {
|
||
defer c.refreshPending.Store(false)
|
||
c.onToolListChanged()
|
||
}()
|
||
}
|
||
|
||
// newSDKClient 按配置构建官方 SDK 客户端：只宣告宿主真正具备的客户端能力。
|
||
func (c *Client) newSDKClient() *mcpsdk.Client {
|
||
client := mcpsdk.NewClient(&mcpsdk.Implementation{Name: clientName, Version: clientVersion}, &mcpsdk.ClientOptions{
|
||
// 🔴 交给 SDK 的 Logger 必须过 redactingLogger（sdklog.go）：SDK 自己会把对端
|
||
// 错误原样打进日志（keepalive 那条已实测），那条路绕开了本包 upstreamError 的出口。
|
||
Logger: redactingLogger(c.log),
|
||
Capabilities: c.clientCapabilities(),
|
||
CreateMessageHandler: c.wrapSampling(),
|
||
ElicitationHandler: c.wrapElicitation(),
|
||
ToolListChangedHandler: c.handleToolListChanged,
|
||
KeepAlive: c.keepAlive,
|
||
KeepAliveFailureThreshold: c.keepAliveFailures,
|
||
})
|
||
if len(c.roots) > 0 {
|
||
client.AddRoots(c.roots...)
|
||
}
|
||
return client
|
||
}
|
||
|
||
// sessionOptions 返回 SDK 会话选项；宿主固定了协议版本时才返回非 nil。
|
||
func (c *Client) sessionOptions() *mcpsdk.ClientSessionOptions {
|
||
if c.preferredProtocolVersion == "" {
|
||
return nil
|
||
}
|
||
return &mcpsdk.ClientSessionOptions{ProtocolVersion: c.preferredProtocolVersion}
|
||
}
|
||
|
||
// wrapConnectErr 把 SDK 报的「协议版本不支持」映射到本包哨兵错误，便于调用方判定。
|
||
//
|
||
// 它同时是握手失败的**脱敏点**：原始 SDK 错误（第三方 server 的自由文本）只经
|
||
// upstreamError 留下 code 与长度，绝不被 %w 带进文案 —— 而哨兵映射仍然成立，因为
|
||
// upstreamFailure 的 cause 保留了整条链，errors.Is 不受影响。
|
||
func wrapConnectErr(err error) error {
|
||
safe := upstreamError(err)
|
||
var wire *sdkjsonrpc.Error
|
||
if errors.As(err, &wire) && wire.Code == mcpsdk.CodeUnsupportedProtocolVersion {
|
||
return fmt.Errorf("%w: %w", ErrUnsupportedProtocolVersion, safe)
|
||
}
|
||
return safe
|
||
}
|
||
|
||
// normalizeArgs 把 agent 传来的参数值规整成 MCP arguments 对象（必须是 JSON 对象）。
|
||
func normalizeArgs(args any) (map[string]any, error) {
|
||
if args == nil {
|
||
return map[string]any{}, nil
|
||
}
|
||
if m, ok := args.(map[string]any); ok {
|
||
return m, nil
|
||
}
|
||
return nil, fmt.Errorf("arguments must be a JSON object, got %T", args)
|
||
}
|