1package mcp
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/base64"
8 "encoding/json"
9 "fmt"
10 "net/http"
11 "net/http/httptest"
12 "os"
13 "os/exec"
14 "os/signal"
15 "path/filepath"
16 "strconv"
17 "strings"
18 "sync"
19 "sync/atomic"
20 "syscall"
21 "testing"
22 "time"
23)
24
25// This file holds the fake MCP servers used by the package tests: a
26// newline-delimited JSON-RPC stdio server (the test binary re-executes itself)
27// and a streamable HTTP server built on httptest. Both share fakeBackend.
28
29const fakeServerEnv = "ZUT_MCP_FAKE_SERVER"
30
31// fakeToolCount is the number of tools the fake server lists in total.
32const fakeToolCount = 18
33
34var fakeImageBytes = []byte("\x89PNG\r\n\x1a\nfake-image-bytes")
35
36func TestMain(m *testing.M) {
37 switch os.Getenv(fakeServerEnv) {
38 case "1":
39 runFakeStdioServer()
40 os.Exit(0)
41 case "sleeper", "sleeper-stubborn":
42 // A descendant of the fake server that outlives it.
43 if os.Getenv(fakeServerEnv) == "sleeper-stubborn" {
44 signal.Ignore(syscall.SIGTERM)
45 }
46 time.Sleep(2 * time.Minute)
47 os.Exit(0)
48 }
49 os.Exit(m.Run())
50}
51
52// fakeBackend implements the protocol methods that do not depend on the
53// transport. Behaviour switches come from the environment (stdio) or fields.
54type fakeBackend struct {
55 noResources bool
56 protocol string
57 repeatPage bool
58 extraTool atomic.Bool
59 // manyResources makes resources/list return this many long entries.
60 manyResources int
61 // hugeDescription, when > 0, gives the first resource a description of
62 // that many bytes.
63 hugeDescription int
64 events string
65}
66
67func fakeBackendFromEnv() *fakeBackend {
68 b := &fakeBackend{
69 noResources: os.Getenv("FAKE_NO_RESOURCES") == "1",
70 protocol: os.Getenv("FAKE_PROTO"),
71 repeatPage: os.Getenv("FAKE_BAD_CURSOR") == "1",
72 events: os.Getenv("FAKE_EVENT_FILE"),
73 }
74 if n, err := strconv.Atoi(os.Getenv("FAKE_MANY_RESOURCES")); err == nil {
75 b.manyResources = n
76 }
77 if n, err := strconv.Atoi(os.Getenv("FAKE_HUGE_DESC")); err == nil {
78 b.hugeDescription = n
79 }
80 return b
81}
82
83func fakeToolDef(name, desc string, readOnly bool) map[string]any {
84 def := map[string]any{
85 "name": name,
86 "description": desc,
87 "inputSchema": map[string]any{
88 "$schema": "http://json-schema.org/draft-07/schema#",
89 "type": "object",
90 "anyOf": []any{map[string]any{"required": []string{"text"}}},
91 "properties": map[string]any{
92 "text": map[string]any{"type": "string"},
93 },
94 },
95 }
96 if readOnly {
97 def["annotations"] = map[string]any{"readOnlyHint": true}
98 }
99 return def
100}
101
102func (b *fakeBackend) toolPage(cursor string) map[string]any {
103 if cursor == "" {
104 return map[string]any{
105 "tools": []any{
106 fakeToolDef("echo", "Echo text back", true),
107 fakeToolDef("image", "Return an image", false),
108 fakeToolDef("resource", "Return an embedded resource", false),
109 fakeToolDef("err", "Return an isError result", false),
110 },
111 "nextCursor": "p2",
112 }
113 }
114 tools := []any{
115 fakeToolDef("structured", "Return structuredContent only", false),
116 fakeToolDef("slow", "Block until cancelled", false),
117 fakeToolDef("exit", "Kill the server process", false),
118 fakeToolDef("big", "Return a very large text", false),
119 fakeToolDef("ask_roots", "Ask the client for roots and ping", false),
120 fakeToolDef("ask_ping", "Ask the client for a ping over the response stream", false),
121 fakeToolDef("add_tool", "Add a tool and notify list_changed", false),
122 fakeToolDef("getenv", "Read an environment variable", false),
123 fakeToolDef("pid", "Return the server process id", false),
124 fakeToolDef("spawn", "Start a descendant process and return its pid", false),
125 fakeToolDef("stall", "Stop reading stdin", false),
126 fakeToolDef("stderr_flood", "Write a huge stderr line", false),
127 fakeToolDef("stderr_secret", "Write a stderr line with FAKE_SECRET straddling byte 4096", false),
128 fakeToolDef("cwd", "Return the server working directory", false),
129 }
130 if b.extraTool.Load() {
131 tools = append(tools, fakeToolDef("extra", "Added later", false))
132 }
133 page := map[string]any{"tools": tools}
134 if b.repeatPage {
135 page["nextCursor"] = "p2"
136 }
137 return page
138}
139
140func fakeText(s string) map[string]any { return map[string]any{"type": "text", "text": s} }
141
142func fakeCallResult(content ...any) map[string]any {
143 return map[string]any{"content": content}
144}
145
146// callTool handles the tools that need no transport-specific behaviour.
147func (b *fakeBackend) callTool(name string, args json.RawMessage) (any, *RPCError) {
148 var in struct {
149 Text string `json:"text"`
150 Name string `json:"name"`
151 }
152 _ = json.Unmarshal(args, &in)
153 switch name {
154 case "echo":
155 return fakeCallResult(fakeText("echo: " + in.Text)), nil
156 case "image":
157 return fakeCallResult(fakeText("an image"), map[string]any{
158 "type": "image", "mimeType": "image/png",
159 "data": base64.StdEncoding.EncodeToString(fakeImageBytes),
160 }), nil
161 case "resource":
162 return fakeCallResult(map[string]any{
163 "type": "resource",
164 "resource": map[string]any{
165 "uri": "mem://emb", "mimeType": "text/plain", "text": "embedded body",
166 },
167 }), nil
168 case "err":
169 return map[string]any{"isError": true, "content": []any{fakeText("tool failed")}}, nil
170 case "structured":
171 return map[string]any{"content": []any{}, "structuredContent": map[string]any{"answer": 42}}, nil
172 case "big":
173 return fakeCallResult(fakeText(strings.Repeat("a", 60*1024)), fakeText("tail")), nil
174 case "pid":
175 return fakeCallResult(fakeText(strconv.Itoa(os.Getpid()))), nil
176 case "cwd":
177 dir, _ := os.Getwd()
178 return fakeCallResult(fakeText(dir)), nil
179 case "getenv":
180 return fakeCallResult(fakeText("[" + os.Getenv(in.Name) + "]")), nil
181 }
182 return nil, &RPCError{Code: -32602, Message: "unknown tool: " + name}
183}
184
185func (b *fakeBackend) handle(method string, params json.RawMessage) (any, *RPCError) {
186 var p struct {
187 Cursor string `json:"cursor"`
188 Name string `json:"name"`
189 Args json.RawMessage `json:"arguments"`
190 URI string `json:"uri"`
191 }
192 _ = json.Unmarshal(params, &p)
193 notFound := &RPCError{Code: codeMethodNotFound, Message: "method not found: " + method}
194 switch method {
195 case "initialize":
196 proto := b.protocol
197 if proto == "" {
198 proto = "2025-06-18"
199 }
200 caps := map[string]any{"tools": map[string]any{"listChanged": true}}
201 if !b.noResources {
202 caps["resources"] = map[string]any{}
203 }
204 return map[string]any{
205 "protocolVersion": proto,
206 "capabilities": caps,
207 "serverInfo": map[string]any{"name": "fake", "version": "1.2.3"},
208 "instructions": "Be nice",
209 }, nil
210 case "ping":
211 return map[string]any{}, nil
212 case "tools/list":
213 return b.toolPage(p.Cursor), nil
214 case "tools/call":
215 return b.callTool(p.Name, p.Args)
216 case "resources/list":
217 if b.noResources {
218 return nil, notFound
219 }
220 fakeEvent(b.events, "list-resources")
221 if b.hugeDescription > 0 {
222 return map[string]any{"resources": []any{map[string]any{
223 "uri": "mem://huge", "name": "huge", "description": strings.Repeat("h", b.hugeDescription),
224 }, map[string]any{"uri": "mem://after", "name": "after"}}}, nil
225 }
226 if b.manyResources > 0 {
227 list := make([]any, 0, b.manyResources)
228 for i := 0; i < b.manyResources; i++ {
229 list = append(list, map[string]any{
230 "uri": fmt.Sprintf("mem://item/%d", i), "name": "item", "description": strings.Repeat("d", 100),
231 })
232 }
233 return map[string]any{"resources": list}, nil
234 }
235 return map[string]any{"resources": []any{map[string]any{
236 "uri": "mem://a", "name": "A", "mimeType": "text/plain", "description": "first",
237 }}}, nil
238 case "resources/templates/list":
239 if b.noResources {
240 return nil, notFound
241 }
242 fakeEvent(b.events, "list-templates")
243 return map[string]any{"resourceTemplates": []any{map[string]any{
244 "uriTemplate": "mem://item/{id}", "name": "Item",
245 }}}, nil
246 case "resources/read":
247 if b.noResources {
248 return nil, notFound
249 }
250 return map[string]any{"contents": []any{map[string]any{
251 "uri": p.URI, "mimeType": "text/plain", "text": "content of " + p.URI,
252 }}}, nil
253 }
254 return nil, notFound
255}
256
257// ---- stdio server ----
258
259func fakeEvent(path, line string) {
260 if path == "" {
261 return
262 }
263 f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
264 if err != nil {
265 return
266 }
267 _, _ = f.WriteString(line + "\n")
268 _ = f.Close()
269}
270
271func runFakeStdioServer() {
272 backend := fakeBackendFromEnv()
273 events := os.Getenv("FAKE_EVENT_FILE")
274 fakeEvent(events, "start")
275 if os.Getenv("FAKE_JUNK") == "1" {
276 // Stray non-JSON output on stdout must be ignored by the client.
277 fmt.Println("starting up... not json")
278 }
279
280 var writeMu sync.Mutex
281 send := func(v any) {
282 raw, err := json.Marshal(v)
283 if err != nil {
284 return
285 }
286 writeMu.Lock()
287 _, _ = os.Stdout.Write(append(raw, '\n'))
288 writeMu.Unlock()
289 }
290 respond := func(id json.RawMessage, result any, rpcErr *RPCError) {
291 if frame, err := encodeResponse(id, result, rpcErr); err == nil {
292 writeMu.Lock()
293 _, _ = os.Stdout.Write(append(frame, '\n'))
294 writeMu.Unlock()
295 }
296 }
297
298 var mu sync.Mutex
299 waiters := map[string]chan message{}
300 cancels := map[string]context.CancelFunc{}
301
302 // askClient sends a server->client request and waits for the answer.
303 askClient := func(id, method string) (json.RawMessage, bool) {
304 key := fmt.Sprintf("%q", id)
305 ch := make(chan message, 1)
306 mu.Lock()
307 waiters[key] = ch
308 mu.Unlock()
309 send(map[string]any{"jsonrpc": "2.0", "id": id, "method": method})
310 select {
311 case msg := <-ch:
312 if msg.Error != nil {
313 return json.RawMessage(`"error"`), true
314 }
315 return msg.Result, true
316 case <-time.After(10 * time.Second):
317 return nil, false
318 }
319 }
320
321 serve := func(msg message) {
322 key := idKey(msg.ID)
323 ctx, cancel := context.WithCancel(context.Background())
324 mu.Lock()
325 cancels[key] = cancel
326 mu.Unlock()
327 defer func() {
328 mu.Lock()
329 delete(cancels, key)
330 mu.Unlock()
331 cancel()
332 }()
333
334 if msg.Method == "initialize" {
335 if ms, _ := time.ParseDuration(os.Getenv("FAKE_INIT_DELAY_MS") + "ms"); ms > 0 {
336 time.Sleep(ms)
337 }
338 if os.Getenv("FAKE_INIT_ERR") == "1" {
339 line := "fatal: missing credentials"
340 if echo := os.Getenv("FAKE_ERR_ECHO"); echo != "" {
341 line += " token=" + echo + " see http://example.invalid/x?k=" + echo
342 }
343 if os.Getenv("FAKE_ANSI") == "1" {
344 line += " \x1b[31mred\x1b[0m\x1b]0;evil-title\x07 end"
345 }
346 fmt.Fprintln(os.Stderr, line)
347 respond(msg.ID, nil, &RPCError{Code: -32000, Message: "init refused " + os.Getenv("FAKE_ERR_ECHO")})
348 return
349 }
350 }
351 if msg.Method == "tools/call" {
352 var p struct {
353 Name string `json:"name"`
354 Args json.RawMessage `json:"arguments"`
355 }
356 _ = json.Unmarshal(msg.Params, &p)
357 switch p.Name {
358 case "slow":
359 fakeEvent(events, "slow-started")
360 select {
361 case <-ctx.Done():
362 case <-time.After(30 * time.Second):
363 respond(msg.ID, fakeCallResult(fakeText("slow done")), nil)
364 }
365 return
366 case "exit":
367 fakeEvent(events, "exit-tool")
368 os.Exit(0)
369 case "spawn":
370 var in struct {
371 Mode string `json:"mode"`
372 }
373 _ = json.Unmarshal(p.Args, &in)
374 child := exec.Command(os.Args[0], "-test.run=^$")
375 child.Env = append(os.Environ(), fakeServerEnv+"="+in.Mode)
376 if err := child.Start(); err != nil {
377 respond(msg.ID, nil, &RPCError{Code: -32000, Message: err.Error()})
378 return
379 }
380 respond(msg.ID, fakeCallResult(fakeText(strconv.Itoa(child.Process.Pid))), nil)
381 return
382 case "stderr_secret":
383 secret := os.Getenv("FAKE_SECRET")
384 pad := 4096 - len(secret)/2
385 fmt.Fprintln(os.Stderr, strings.Repeat("x", pad)+secret+strings.Repeat("y", 200))
386 fmt.Fprintln(os.Stderr, "after-secret")
387 respond(msg.ID, fakeCallResult(fakeText("written")), nil)
388 return
389 case "stderr_flood":
390 os.Stderr.WriteString(strings.Repeat("x", 3<<20) + "\n")
391 fmt.Fprintln(os.Stderr, "after-flood")
392 respond(msg.ID, fakeCallResult(fakeText("flooded")), nil)
393 return
394 case "ask_roots":
395 roots, ok1 := askClient("srv-roots", "roots/list")
396 ping, ok2 := askClient("srv-ping", "ping")
397 if !ok1 || !ok2 {
398 respond(msg.ID, nil, &RPCError{Code: -32000, Message: "client did not answer"})
399 return
400 }
401 respond(msg.ID, fakeCallResult(fakeText(fmt.Sprintf("roots=%s ping=%s", roots, ping))), nil)
402 return
403 case "add_tool":
404 backend.extraTool.Store(true)
405 send(map[string]any{"jsonrpc": "2.0", "method": "notifications/tools/list_changed"})
406 respond(msg.ID, fakeCallResult(fakeText("added")), nil)
407 return
408 }
409 }
410 result, rpcErr := backend.handle(msg.Method, msg.Params)
411 respond(msg.ID, result, rpcErr)
412 if msg.Method == "tools/list" && os.Getenv("FAKE_NOTIFY_ON_LIST") == "1" {
413 var p struct {
414 Cursor string `json:"cursor"`
415 }
416 _ = json.Unmarshal(msg.Params, &p)
417 if p.Cursor == "p2" && !backend.extraTool.Load() {
418 // The tool set changes right after the initial listing.
419 backend.extraTool.Store(true)
420 send(map[string]any{"jsonrpc": "2.0", "method": "notifications/tools/list_changed"})
421 }
422 }
423 }
424
425 sc := bufio.NewScanner(os.Stdin)
426 sc.Buffer(make([]byte, 0, 64*1024), 16<<20)
427 for sc.Scan() {
428 var msg message
429 if err := json.Unmarshal(sc.Bytes(), &msg); err != nil {
430 continue
431 }
432 switch {
433 case msg.isResponse():
434 mu.Lock()
435 ch := waiters[idKey(msg.ID)]
436 delete(waiters, idKey(msg.ID))
437 mu.Unlock()
438 if ch != nil {
439 ch <- msg
440 }
441 case msg.isRequest():
442 if msg.Method == "tools/call" && bytes.Contains(msg.Params, []byte(`"name":"stall"`)) {
443 // Stop reading stdin so the client's writes back up.
444 fakeEvent(events, "stalled")
445 time.Sleep(time.Minute)
446 }
447 go serve(msg)
448 case msg.Method == "notifications/cancelled":
449 var p struct {
450 RequestID json.RawMessage `json:"requestId"`
451 }
452 _ = json.Unmarshal(msg.Params, &p)
453 fakeEvent(events, "cancelled:"+idKey(p.RequestID))
454 mu.Lock()
455 cancel := cancels[idKey(p.RequestID)]
456 mu.Unlock()
457 if cancel != nil {
458 cancel()
459 }
460 case msg.Method == "notifications/initialized":
461 fakeEvent(events, "initialized")
462 }
463 }
464 fakeEvent(events, "stdin-closed")
465}
466
467// fakeStdioConfig returns a ServerConfig that launches the fake stdio server
468// and the path of its event log. extra entries become the server's Env.
469func fakeStdioConfig(t *testing.T, extra map[string]string) (ServerConfig, string) {
470 t.Helper()
471 events := filepath.Join(t.TempDir(), "events.log")
472 // GORACE: under -race the runtime sleeps at exit, which would make every
473 // graceful shutdown look like it needed SIGTERM.
474 env := map[string]string{fakeServerEnv: "1", "FAKE_EVENT_FILE": events, "GORACE": "atexit_sleep_ms=0"}
475 for k, v := range extra {
476 env[k] = v
477 }
478 exe, err := os.Executable()
479 if err != nil {
480 exe = os.Args[0]
481 }
482 return ServerConfig{
483 Command: exe,
484 Args: []string{"-test.run=^$"},
485 Env: env,
486 }, events
487}
488
489func readEvents(path string) []string {
490 raw, err := os.ReadFile(path)
491 if err != nil {
492 return nil
493 }
494 return strings.Fields(string(raw))
495}
496
497func countEvents(path, want string) int {
498 n := 0
499 for _, e := range readEvents(path) {
500 if e == want {
501 n++
502 }
503 }
504 return n
505}
506
507// eventually polls cond until it holds or five seconds pass.
508func eventually(t *testing.T, what string, cond func() bool) {
509 t.Helper()
510 deadline := time.Now().Add(5 * time.Second)
511 for !cond() {
512 if time.Now().After(deadline) {
513 t.Fatalf("timed out waiting for %s", what)
514 }
515 time.Sleep(5 * time.Millisecond)
516 }
517}
518
519func waitEvent(t *testing.T, path, prefix string) {
520 t.Helper()
521 eventually(t, "event "+prefix, func() bool {
522 for _, e := range readEvents(path) {
523 if strings.HasPrefix(e, prefix) {
524 return true
525 }
526 }
527 return false
528 })
529}
530
531// ---- HTTP server ----
532
533type httpRecord struct {
534 Method string
535 Session string
536 Protocol string
537 Auth string
538 Static string
539 EnvHeader string
540 RPCMethod string
541}
542
543// httpFake is a streamable HTTP MCP server. With sse set every POST answer
544// is a text/event-stream, otherwise application/json.
545type httpFake struct {
546 backend *fakeBackend
547 sse bool
548 wantAuth string
549 sessionID string
550
551 // expireNextCall makes the next tools/call that carries a session id
552 // answer 404 (session expired).
553 expireNextCall atomic.Bool
554 // hangInit makes initialize block until the client goes away.
555 hangInit bool
556 // cancelGate, when set, makes notifications/cancelled block until it is
557 // closed or the client disconnects.
558 cancelGate chan struct{}
559 // errorBody, when set, is returned with a 500 for every tools/list.
560 errorBody string
561
562 mu sync.Mutex
563 records []httpRecord
564 pingReplies chan message
565 deleted chan string
566 initCount int
567 callCount int
568}
569
570func newHTTPFake(sse bool) *httpFake {
571 return &httpFake{
572 backend: &fakeBackend{},
573 sse: sse,
574 sessionID: "sess-abc",
575 pingReplies: make(chan message, 4),
576 deleted: make(chan string, 4),
577 }
578}
579
580func (f *httpFake) start(t *testing.T) *httptest.Server {
581 srv := httptest.NewServer(f)
582 t.Cleanup(srv.Close)
583 return srv
584}
585
586func (f *httpFake) counts() (inits, calls int) {
587 f.mu.Lock()
588 defer f.mu.Unlock()
589 return f.initCount, f.callCount
590}
591
592func (f *httpFake) snapshot() []httpRecord {
593 f.mu.Lock()
594 defer f.mu.Unlock()
595 return append([]httpRecord(nil), f.records...)
596}
597
598func (f *httpFake) ServeHTTP(w http.ResponseWriter, r *http.Request) {
599 rec := httpRecord{
600 Method: r.Method,
601 Session: r.Header.Get("Mcp-Session-Id"),
602 Protocol: r.Header.Get("MCP-Protocol-Version"),
603 Auth: r.Header.Get("Authorization"),
604 Static: r.Header.Get("X-Static"),
605 EnvHeader: r.Header.Get("X-Env"),
606 }
607 if f.wantAuth != "" && rec.Auth != f.wantAuth {
608 f.mu.Lock()
609 f.records = append(f.records, rec)
610 f.mu.Unlock()
611 http.Error(w, "unauthorized", http.StatusUnauthorized)
612 return
613 }
614 switch r.Method {
615 case http.MethodGet:
616 f.mu.Lock()
617 f.records = append(f.records, rec)
618 f.mu.Unlock()
619 w.Header().Set("Allow", "POST, DELETE")
620 http.Error(w, "no server stream", http.StatusMethodNotAllowed)
621 case http.MethodDelete:
622 f.mu.Lock()
623 f.records = append(f.records, rec)
624 f.mu.Unlock()
625 f.deleted <- rec.Session
626 w.WriteHeader(http.StatusOK)
627 case http.MethodPost:
628 f.servePost(w, r, rec)
629 default:
630 http.Error(w, "bad method", http.StatusMethodNotAllowed)
631 }
632}
633
634func (f *httpFake) servePost(w http.ResponseWriter, r *http.Request, rec httpRecord) {
635 var msg message
636 if err := json.NewDecoder(r.Body).Decode(&msg); err != nil {
637 http.Error(w, "bad json", http.StatusBadRequest)
638 return
639 }
640 rec.RPCMethod = msg.Method
641 f.mu.Lock()
642 f.records = append(f.records, rec)
643 f.mu.Unlock()
644
645 if msg.isResponse() {
646 f.pingReplies <- msg
647 w.WriteHeader(http.StatusAccepted)
648 return
649 }
650 if !msg.isRequest() {
651 if msg.Method == "notifications/cancelled" && f.cancelGate != nil {
652 select {
653 case <-f.cancelGate:
654 case <-r.Context().Done():
655 }
656 }
657 w.WriteHeader(http.StatusAccepted)
658 return
659 }
660
661 var p struct {
662 Name string `json:"name"`
663 }
664 _ = json.Unmarshal(msg.Params, &p)
665 if msg.Method == "initialize" {
666 if f.hangInit {
667 <-r.Context().Done()
668 return
669 }
670 f.mu.Lock()
671 f.initCount++
672 n := f.initCount
673 f.mu.Unlock()
674 sid := f.sessionID
675 if n > 1 {
676 sid = fmt.Sprintf("%s-%d", f.sessionID, n)
677 }
678 w.Header().Set("Mcp-Session-Id", sid)
679 }
680 if msg.Method == "tools/call" {
681 f.mu.Lock()
682 f.callCount++
683 f.mu.Unlock()
684 if rec.Session != "" && f.expireNextCall.CompareAndSwap(true, false) {
685 http.Error(w, "session not found", http.StatusNotFound)
686 return
687 }
688 }
689 if msg.Method == "tools/list" && f.errorBody != "" {
690 w.Header().Set("Content-Type", "text/plain")
691 w.WriteHeader(http.StatusInternalServerError)
692 _, _ = w.Write([]byte(f.errorBody))
693 return
694 }
695
696 if msg.Method == "tools/call" && p.Name == "slow" {
697 // Hold the request open until the client gives up (disconnects).
698 select {
699 case <-r.Context().Done():
700 case <-time.After(10 * time.Second):
701 }
702 return
703 }
704
705 if !f.sse {
706 result, rpcErr := f.backend.handle(msg.Method, msg.Params)
707 frame, _ := encodeResponse(msg.ID, result, rpcErr)
708 w.Header().Set("Content-Type", "application/json")
709 _, _ = w.Write(frame)
710 return
711 }
712
713 w.Header().Set("Content-Type", "text/event-stream")
714 w.WriteHeader(http.StatusOK)
715 flusher, _ := w.(http.Flusher)
716 event := func(payload []byte) {
717 fmt.Fprintf(w, "event: message\ndata: %s\n\n", payload)
718 if flusher != nil {
719 flusher.Flush()
720 }
721 }
722 fmt.Fprint(w, ": keepalive\n\n")
723 if msg.Method == "tools/call" {
724 // A notification before the response must not end the wait.
725 note, _ := encodeNotification("notifications/message", map[string]any{"level": "info"})
726 event(note)
727 }
728 var (
729 result any
730 rpcErr *RPCError
731 )
732 if msg.Method == "tools/call" && p.Name == "ask_ping" {
733 ping, _ := encodeRequest(777, "ping", nil)
734 event(ping)
735 select {
736 case reply := <-f.pingReplies:
737 result = fakeCallResult(fakeText("ping reply: " + string(reply.Result)))
738 case <-time.After(5 * time.Second):
739 rpcErr = &RPCError{Code: -32000, Message: "no ping reply"}
740 }
741 } else {
742 result, rpcErr = f.backend.handle(msg.Method, msg.Params)
743 }
744 frame, _ := encodeResponse(msg.ID, result, rpcErr)
745 event(frame)
746}