feat: working with initialisation over /messages

hugoduncan · hugoduncan · commit 3369f2b82647 · 2025-01-05T20:22:19.000-05:00
diff --git a/components/http-server/src/mcp_clj/http_server/ring_adapter.clj b/components/http-server/src/mcp_clj/http_server/ring_adapter.clj
@@ -35,9 +35,14 @@
     (.sendResponseHeaders exchange status 0)
     (with-open [os (.getResponseBody exchange)]
       (try
-        (body os)
-        (.flush os)
+        (try
+          (body os)
+          (.flush os)
+          (catch sun.net.httpserver.StreamClosedException _
+            ;; Client disconnected - normal for SSE
+            nil))
         (catch Exception e
+          ;; Log other exceptions
           (.printStackTrace e))))))
 
 (defn- send-ring-response
diff --git a/components/json-rpc/src/mcp_clj/json_rpc/server.clj b/components/json-rpc/src/mcp_clj/json_rpc/server.clj
@@ -4,105 +4,94 @@
    [clojure.data.json :as json]
    [mcp-clj.http-server.ring-adapter :as http]
    [mcp-clj.json-rpc.protocol :as protocol]
-   [ring.util.response :as response])
-  (:import
-   [com.sun.net.httpserver HttpServer]
-   [java.io OutputStreamWriter BufferedWriter]))
-
-(defn- write-sse-message
-  "Write a Server-Sent Event message"
-  [^BufferedWriter writer message]
-  (doto writer
-    (.write (str "data: " message "\n\n"))
-    (.flush)))
+   [ring.util.response :as response]))
 
 (defn- handle-json-rpc
-  "Process a single JSON-RPC request and return response"
+  "Process a JSON-RPC request and return response"
   [handlers request]
   (if-let [validation-error (protocol/validate-request request)]
-    validation-error
+    (assoc validation-error :id (:id request))
     (let [{:keys [method params id]} request
           handler                    (get handlers method)]
       (if handler
         (try
-          (let [result (handler params)]
-            (protocol/result-response id result))
+          {:jsonrpc "2.0"
+           :id      id
+           :result  (handler params)}
           (catch Exception e
-            (protocol/error-response
-             (get protocol/error-codes :internal-error)
-             (.getMessage e)
-             {:id id})))
-        (protocol/error-response
-         (get protocol/error-codes :method-not-found)
-         (str "Method not found: " method)
-         {:id id})))))
+            {:jsonrpc "2.0"
+             :id      id
+             :error   {:code    (get protocol/error-codes :internal-error)
+                       :message (.getMessage e)}}))
+        {:jsonrpc "2.0"
+         :id      id
+         :error   {:code    (get protocol/error-codes :method-not-found)
+                   :message (str "Method not found: " method)}}))))
+
+(defn- handle-post
+  "Handle JSON-RPC POST request"
+  [handlers {:keys [body] :as request}]
+  (try
+    (let [request-data (json/read-str (slurp body) :key-fn keyword)
+          response     (handle-json-rpc handlers request-data)
+          response-str (json/write-str response)]
+      (-> (response/response response-str)
+          (response/status 200)
+          (response/content-type "application/json")))
+    (catch Exception e
+      (-> (response/response
+           (json/write-str
+            {:jsonrpc "2.0"
+             :error   {:code    (get protocol/error-codes :internal-error)
+                       :message (str "Unexpected error: " (ex-message e))}}))
+          (response/status 500)
+          (response/content-type "application/json")))))
 
-(defn- handle-sse-request
-  "Handle SSE stream setup and message processing"
-  [handlers message-stream]
-  (fn [^java.io.OutputStream output-stream]
-    (with-open [writer (-> output-stream
-                           (OutputStreamWriter. "UTF-8")
-                           BufferedWriter.)]
+(defn- handle-sse
+  "Handle SSE stream setup"
+  [response-stream]
+  (fn [output]
+    (let [writer (java.io.OutputStreamWriter. output "UTF-8")]
       (try
-        (let [control-ch message-stream]
-          (loop []
-            (when-let [message @control-ch]
-              (let [[request parse-error] (protocol/parse-json message)
-                    response              (if parse-error
-                                            parse-error
-                                            (handle-json-rpc handlers request))
-                    [json-response error] (protocol/write-json response)]
-                (when json-response
-                  (write-sse-message writer json-response))
-                (when error
-                  (write-sse-message writer
-                                     (json/write-str
-                                      (protocol/error-response
-                                       (get protocol/error-codes :internal-error)
-                                       "Response encoding error")))))
-              (reset! control-ch nil)
-              (recur))))
+        (loop []
+          (when-let [response @response-stream]
+            (.write writer (str "data: " (json/write-str response) "\n\n"))
+            (.flush writer)
+            (reset! response-stream nil)
+            (recur)))
         (catch Exception e
-          (write-sse-message writer
-                             (json/write-str
-                              (protocol/error-response
-                               (get protocol/error-codes :internal-error)
-                               "Stream error"))))))))
-
-(defn get-server-port
-  "Get the actual port a server is listening on"
-  [^HttpServer server]
-  (.getPort (.getAddress server)))
+          (.printStackTrace e))))))
 
 (defn create-server
-  "Create a new JSON-RPC SSE server.
+  "Create JSON-RPC server with SSE support.
 
    Configuration options:
-   - :port           Port number (default: 0 for auto-assignment)
-   - :handlers       Map of method names to handler functions
-   - :message-stream Shared atom for message passing
-
-   Returns map with:
-   - :server   The server instance
-   - :port     The actual port the server is running on
-   - :stop     Function to stop the server"
-  [{:keys [port handlers message-stream]
-    :or   {port 0}
-    :as   config}]
+   - :port     Port number (default: 0 for auto-assignment)
+   - :handlers Map of method names to handler functions"
+  [{:keys [port handlers]
+    :or   {port 0}}]
   (when-not (map? handlers)
-    (throw (ex-info "Handlers must be a map" {:config config})))
-  (when-not (instance? clojure.lang.Atom message-stream)
-    (throw (ex-info "Message stream must be an atom" {:config config})))
+    (throw (ex-info "Handlers must be a map" {:handlers handlers})))
+
+  (let [response-stream (atom nil)
+        handler         (fn [{:keys [request-method uri] :as request}]
+                          (case [request-method uri]
+                            [:post "/message"]
+                            (handle-post handlers request)
+
+                            [:get "/sse"]
+                            (-> (response/response (handle-sse response-stream))
+                                (response/status 200)
+                                (response/content-type "text/event-stream")
+                                (response/header "Cache-Control" "no-cache")
+                                (response/header "Connection" "keep-alive"))
+
+                            (-> (response/response "Not Found")
+                                (response/status 404)
+                                (response/content-type "text/plain"))))
 
-  (let [{:keys [server stop]} (http/run-server
-                               (fn [request]
-                                 (-> (response/response
-                                      (handle-sse-request handlers message-stream))
-                                     (response/content-type "text/event-stream")
-                                     (response/header "Cache-Control" "no-cache")
-                                     (response/header "Connection" "keep-alive")))
-                               {:port port})]
-    {:server server
-     :port   (get-server-port server)
-     :stop   stop}))
+        {:keys [server stop]} (http/run-server handler {:port port})]
+    {:server          server
+     :response-stream response-stream
+     :port            (.getPort (.getAddress server))
+     :stop            stop}))
diff --git a/components/mcp-server/src/mcp_clj/mcp_server/core.clj b/components/mcp-server/src/mcp_clj/mcp_server/core.clj
@@ -12,6 +12,7 @@
   "Handle initialize request from client.
    Returns server info, protocol version and capabilities."
   [server {:keys [capabilities clientInfo protocolVersion] :as params}]
+  (binding [*out* *err*] (prn "handle-initialize"))
   (when-not (= protocolVersion required-client-version)
     (throw (ex-info (str "Unsupported protocol version. Required " required-client-version)
                     {:code -32001
@@ -36,16 +37,23 @@
 (defn handle-initialized
   "Handle initialized notification from client."
   [server _params]
+  (println "handle-initialized server:" server)
+  (println "handle-initialized state:" @(:state server))
   (swap! (:state server) assoc :initialized? true)
+  (println "handle-initialized state after:" @(:state server))
   nil)
 
 (defn ping
   "Handle ping request."
   [server _params]
+  (println "ping server:" server)
+  (println "ping state:" @(:state server))
   (when-not (:initialized? @(:state server))
+    (println "ping throwing not initialized")
     (throw (ex-info "Server not initialized"
                     {:code -32002
                      :data {:state "uninitialized"}})))
+  (println "ping returning empty map")
   {})
 
 (defn create-handlers
@@ -63,11 +71,14 @@
    - :tools    Optional. Map of tool implementations"
   [{:keys [port tools] :as options}]
   (let [state           (atom {:initialized? false})
-        handlers        (create-handlers {:state state})
+        message-stream  (atom nil)
+        server         (->MCPServer nil state)
+        handlers       (create-handlers server)
         json-rpc-server (json-rpc/create-server
-                         {:port     port
-                          :handlers handlers})
-        server          (->MCPServer json-rpc-server state)]
-    (assoc server
+                         {:port           port
+                          :handlers       handlers
+                          :message-stream message-stream})]
+    (assoc (assoc server :json-rpc-server json-rpc-server)
+           :message-stream message-stream
            :stop #(do (reset! state {:initialized? false})
                       ((:stop json-rpc-server))))))
diff --git a/components/mcp-server/test/mcp_clj/mcp_server_test.clj b/components/mcp-server/test/mcp_clj/mcp_server_test.clj
@@ -1,6 +1,8 @@
 (ns mcp-clj.mcp-server-test
   (:require
    [clojure.test :refer [deftest testing is use-fixtures]]
+   [clojure.data.json :as json]
+   [hato.client :as hato]
    [mcp-clj.mcp-server.core :as mcp]))
 
 (def valid-client-info
@@ -53,15 +55,55 @@
            (mcp/ping server nil))
           "Ping throws when not initialized"))))
 
-(deftest server-creation-test
-  (testing "create server"
-    (let [server (mcp/create-server {:port 8080})]
-      (is (instance? mcp_clj.mcp_server.core.MCPServer server)
-          "Returns MCPServer instance")
-      (is (fn? (:stop server))
-          "Has stop function")
-      (is (not (:initialized? @(:state server)))
-          "Initial state is uninitialized")
-
-      ;; Stop server
-      ((:stop server)))))
+(def valid-client-info
+  {:clientInfo      {:name "test-client" :version "1.0"}
+   :protocolVersion "0.1"
+   :capabilities    {:tools {}}})
+
+(defn send-request
+  "Send a JSON-RPC request and get response"
+  [url request]
+  (let [response (hato/post (str url "/message")
+                            {:headers {"Content-Type" "application/json"}
+                             :body    (json/write-str request)})
+        body     (json/read-str (:body response))]
+    (println "sent:" request)
+    (println "received:" body)
+    body))
+
+(deftest integration-test
+  (testing "server request handling"
+    (let [server   (mcp/create-server {:port 0})
+          port     (get-in server [:json-rpc-server :port])
+          base-url (format "http://localhost:%d" port)
+
+          initialize-request {:jsonrpc "2.0"
+                              :method  "initialize"
+                              :id      1
+                              :params  valid-client-info}
+
+          initialized-notification {:jsonrpc "2.0"
+                                    :method  "notifications/initialized"}  ; No id for notification
+
+          ping-request {:jsonrpc "2.0"
+                        :method  "ping"
+                        :id      2}]
+
+      (try
+        (let [init-response (send-request base-url initialize-request)]
+          (is (= 1 (get init-response "id"))
+              "Initialize response has correct id")
+          (is (get-in init-response ["result" "serverInfo"])
+              "Initialize response contains server info"))
+
+        ;; Send initialized notification
+        (send-request base-url initialized-notification)
+
+        (let [ping-response (send-request base-url ping-request)]
+          (is (= 2 (get ping-response "id"))
+              "Ping response has correct id")
+          (is (= {} (get ping-response "result"))
+              "Ping response result is empty map"))
+
+        (finally
+          ((:stop server)))))))
diff --git a/deps.edn b/deps.edn
@@ -13,7 +13,8 @@
                          "components/json-rpc/test"
                          "components/mcp-server/test"]
            :extra-deps
-           {org.clojure/test.check {:mvn/version "1.1.1"}}
+           {hato/hato              {:mvn/version "1.0.0"}
+            org.clojure/test.check {:mvn/version "1.1.1"}}
            :exec-fn     kaocha.runner/exec-fn
            :exec-args   {}
            :jvm-opts    ["-XX:-OmitStackTraceInFastThrow"
