http.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. package agentmcp
  2. import (
  3. "context"
  4. "crypto/subtle"
  5. "errors"
  6. "net/http"
  7. "strings"
  8. "time"
  9. "github.com/mark3labs/mcp-go/server"
  10. "go.uber.org/zap"
  11. "fwh-toolbox-server/agentmcp/kit"
  12. )
  13. // Handler 返回 HTTP 处理器:/mcp(Streamable HTTP,无状态,需 Bearer 令牌)与 /healthz。
  14. // tasks 为持久化长任务工具:任务协议请求由 kit.TaskRouter 在 mcp-go 之前处理(任务 id 以访问令牌签名)。
  15. func Handler(s *server.MCPServer, token string, tasks []kit.TaskTool, log *zap.Logger) http.Handler {
  16. mcpHandler := server.NewStreamableHTTPServer(s,
  17. server.WithStateLess(true),
  18. server.WithEndpointPath("/mcp"),
  19. )
  20. mux := http.NewServeMux()
  21. mux.Handle("/mcp", bearerAuth(token, kit.NewTaskRouter(mcpHandler, tasks, token, log)))
  22. mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
  23. w.WriteHeader(http.StatusOK)
  24. _, _ = w.Write([]byte("ok"))
  25. })
  26. return mux
  27. }
  28. // bearerAuth 校验 Authorization: Bearer <token>,常量时间比较。
  29. func bearerAuth(token string, next http.Handler) http.Handler {
  30. want := []byte("Bearer " + token)
  31. return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  32. got := []byte(strings.TrimSpace(r.Header.Get("Authorization")))
  33. if subtle.ConstantTimeCompare(got, want) != 1 {
  34. w.Header().Set("Content-Type", "application/json")
  35. w.WriteHeader(http.StatusUnauthorized)
  36. _, _ = w.Write([]byte(`{"jsonrpc":"2.0","error":{"code":-32001,"message":"unauthorized"},"id":null}`))
  37. return
  38. }
  39. next.ServeHTTP(w, r)
  40. })
  41. }
  42. // Serve 启动 HTTP 服务,ctx 取消时优雅关闭。
  43. func Serve(ctx context.Context, addr string, h http.Handler, log *zap.Logger) error {
  44. srv := &http.Server{
  45. Addr: addr,
  46. Handler: h,
  47. ReadHeaderTimeout: 10 * time.Second,
  48. }
  49. errCh := make(chan error, 1)
  50. go func() { errCh <- srv.ListenAndServe() }()
  51. log.Info("[agent-mcp] HTTP 服务已启动", zap.String("addr", addr))
  52. select {
  53. case <-ctx.Done():
  54. shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  55. defer cancel()
  56. return srv.Shutdown(shutdownCtx)
  57. case err := <-errCh:
  58. if errors.Is(err, http.ErrServerClosed) {
  59. return nil
  60. }
  61. return err
  62. }
  63. }