返回 DeepSeek-Reasonix
protocol_inbound.go
根目录 / internal / plugin / protocol_inbound.go
1 package plugin
2
3 import (
4 "encoding/json"
5 "fmt"
6 "net/url"
7 "path/filepath"
8 "strconv"
9 "strings"
10 "sync"
11
12 "reasonix/internal/tool"
13 )
14
15 // progressTransport is optional so lightweight transports used by embedders and
16 // tests remain valid. Native transports implement it and route a server's
17 // notifications/progress message to the matching tools/call context.
18 type progressTransport interface {
19 registerProgress(token string, sink tool.ProgressFunc) func()
20 }
21
22 // notificationTransport is implemented by supervised SDK transports that can
23 // receive server notifications. The callback must stay non-blocking because the
24 // SDK dispatches notification handlers independently from request completion.
25 type notificationTransport interface {
26 registerNotification(method string, callback func(json.RawMessage)) func()
27 }
28
29 type progressRouter struct {
30 mu sync.Mutex
31 sinks map[string]tool.ProgressFunc
32 }
33
34 type notificationRouter struct {
35 mu sync.Mutex
36 nextID uint64
37 listeners map[string]map[uint64]func(json.RawMessage)
38 }
39
40 func (r *notificationRouter) registerNotification(method string, callback func(json.RawMessage)) func() {
41 method = strings.TrimSpace(method)
42 if method == "" || callback == nil {
43 return func() {}
44 }
45 r.mu.Lock()
46 if r.listeners == nil {
47 r.listeners = map[string]map[uint64]func(json.RawMessage){}
48 }
49 r.nextID++
50 id := r.nextID
51 if r.listeners[method] == nil {
52 r.listeners[method] = map[uint64]func(json.RawMessage){}
53 }
54 r.listeners[method][id] = callback
55 r.mu.Unlock()
56 return func() {
57 r.mu.Lock()
58 delete(r.listeners[method], id)
59 if len(r.listeners[method]) == 0 {
60 delete(r.listeners, method)
61 }
62 r.mu.Unlock()
63 }
64 }
65
66 func (r *notificationRouter) dispatchNotification(method string, params json.RawMessage) {
67 r.mu.Lock()
68 listeners := make([]func(json.RawMessage), 0, len(r.listeners[method]))
69 for _, callback := range r.listeners[method] {
70 listeners = append(listeners, callback)
71 }
72 r.mu.Unlock()
73 for _, callback := range listeners {
74 callback(append(json.RawMessage(nil), params...))
75 }
76 }
77
78 func (r *progressRouter) registerProgress(token string, sink tool.ProgressFunc) func() {
79 if token == "" || sink == nil {
80 return func() {}
81 }
82 r.mu.Lock()
83 if r.sinks == nil {
84 r.sinks = map[string]tool.ProgressFunc{}
85 }
86 r.sinks[token] = sink
87 r.mu.Unlock()
88 return func() {
89 r.mu.Lock()
90 delete(r.sinks, token)
91 r.mu.Unlock()
92 }
93 }
94
95 func (r *progressRouter) clear() {
96 r.mu.Lock()
97 r.sinks = nil
98 r.mu.Unlock()
99 }
100
101 func (r *progressRouter) dispatchProgress(params json.RawMessage) bool {
102 var p struct {
103 ProgressToken any `json:"progressToken"`
104 Progress *float64 `json:"progress"`
105 Total *float64 `json:"total"`
106 Message string `json:"message"`
107 }
108 if err := json.Unmarshal(params, &p); err != nil {
109 return false
110 }
111 token := progressTokenKey(p.ProgressToken)
112 if token == "" {
113 return false
114 }
115 r.mu.Lock()
116 sink := r.sinks[token]
117 r.mu.Unlock()
118 if sink == nil {
119 return false
120 }
121 sink(formatMCPProgress(p.Message, p.Progress, p.Total))
122 return true
123 }
124
125 func progressTokenKey(token any) string {
126 switch value := token.(type) {
127 case string:
128 return value
129 case float64:
130 return strconv.FormatFloat(value, 'f', -1, 64)
131 case json.Number:
132 return value.String()
133 default:
134 return ""
135 }
136 }
137
138 func formatMCPProgress(message string, progress, total *float64) string {
139 label := strings.TrimSpace(message)
140 if label == "" {
141 label = "MCP progress"
142 }
143 formatNumber := func(value float64) string {
144 return strconv.FormatFloat(value, 'f', -1, 64)
145 }
146 switch {
147 case progress != nil && total != nil:
148 return fmt.Sprintf("%s (%s/%s)\n", label, formatNumber(*progress), formatNumber(*total))
149 case progress != nil:
150 return fmt.Sprintf("%s (%s)\n", label, formatNumber(*progress))
151 default:
152 return label + "\n"
153 }
154 }
155
156 type mcpRoot struct {
157 URI string `json:"uri"`
158 Name string `json:"name,omitempty"`
159 }
160
161 func mcpRoots(workspaceRoot string) []mcpRoot {
162 root := strings.TrimSpace(workspaceRoot)
163 if root == "" {
164 return nil
165 }
166 abs, err := filepath.Abs(root)
167 if err != nil {
168 return nil
169 }
170 clean := filepath.Clean(abs)
171 path := filepath.ToSlash(clean)
172 fileURL := &url.URL{Scheme: "file"}
173 if after, ok := strings.CutPrefix(path, "//"); ok {
174 parts := strings.SplitN(after, "/", 2)
175 fileURL.Host = parts[0]
176 if len(parts) == 2 {
177 fileURL.Path = "/" + parts[1]
178 } else {
179 fileURL.Path = "/"
180 }
181 } else {
182 if volume := filepath.VolumeName(clean); volume != "" && !strings.HasPrefix(path, "/") {
183 path = "/" + path
184 }
185 fileURL.Path = path
186 }
187 name := filepath.Base(clean)
188 if name == "." {
189 name = clean
190 }
191 return []mcpRoot{{URI: fileURL.String(), Name: name}}
192 }
193
193 lines GO