Merge pull request #335 from atesgoral/ag/use-sdk-streamable-http-client-example

Update Streamable HTTP client example to use MCP::Client

examples/streamable_http_client.rb

@@ -1,49 +1,25 @@
 # frozen_string_literal: true
 
+$LOAD_PATH.unshift(File.expand_path("../lib", __dir__))
+require "mcp"
 require "net/http"
 require "uri"
 require "json"
 require "logger"
+require "event_stream_parser"
 
-# Logger for client operations
+SERVER_URL = "http://localhost:9393"
-logger = Logger.new($stdout)
+
-logger.formatter = proc do |severity, datetime, _progname, msg|
+def create_logger
+  logger = Logger.new($stdout)
+  logger.formatter = proc do |severity, datetime, _progname, msg|
     "[CLIENT] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n"
+  end
+  logger
 end
 
-# Server configuration
+# The SDK does not yet implement the optional GET SSE stream, so this example
-SERVER_URL = "http://localhost:9393/mcp"
+# uses MCP::Client for JSON-RPC requests and raw Net::HTTP for the event stream.
-PROTOCOL_VERSION = "2024-11-05"
-
-# Helper method to make JSON-RPC requests
-def make_request(session_id, method, params = {}, id = nil)
-  uri = URI(SERVER_URL)
-  http = Net::HTTP.new(uri.host, uri.port)
-
-  request = Net::HTTP::Post.new(uri)
-  request["Content-Type"] = "application/json"
-  request["Mcp-Session-Id"] = session_id if session_id
-
-  body = {
-    jsonrpc: "2.0",
-    method: method,
-    params: params,
-    id: id || SecureRandom.uuid,
-  }
-
-  request.body = body.to_json
-  response = http.request(request)
-
-  {
-    status: response.code,
-    headers: response.to_hash,
-    body: JSON.parse(response.body),
-  }
-rescue => e
-  { error: e.message }
-end
-
-# Connect to SSE stream
 def connect_sse(session_id, logger)
   uri = URI(SERVER_URL)
 
@@ -59,17 +35,13 @@ def connect_sse(session_id, logger)
       if response.code == "200"
         logger.info("SSE stream connected successfully")
 
+        parser = EventStreamParser::Parser.new
         response.read_body do |chunk|
-          chunk.split("\n").each do |line|
+          parser.feed(chunk) do |type, data, _id|
-            if line.start_with?("data: ")
+            if type.empty?
-              data = line[6..-1]
+              logger.info("SSE event: #{data}")
-              begin
+            else
-                logger.info("SSE data: #{data}")
+              logger.info("SSE event (#{type}): #{data}")
-              rescue JSON::ParserError
-                logger.debug("Non-JSON SSE data: #{data}")
-              end
-            elsif line.start_with?(": ")
-              logger.debug("SSE keepalive received: #{line}")
             end
           end
         end
@@ -79,106 +51,104 @@ def connect_sse(session_id, logger)
     end
   end
 rescue Interrupt
-  logger.info("SSE connection interrupted by user")
+  logger.info("SSE connection interrupted")
 rescue => e
   logger.error("SSE connection error: #{e.message}")
 end
 
-# Main client flow
+def print_response(response)
-def main
+  if response.nil?
-  logger = Logger.new($stdout)
+    puts "Response accepted; watch the SSE stream for the server response."
-  logger.formatter = proc do |severity, datetime, _progname, msg|
+  else
-    "[CLIENT] #{severity} #{datetime.strftime("%H:%M:%S.%L")} - #{msg}\n"
+    puts "Response: #{JSON.pretty_generate(response)}"
   end
+end
 
-  puts "=== MCP SSE Test Client ==="
+def main
+  logger = create_logger
 
-  # Step 1: Initialize session
+  puts <<~MESSAGE
-  logger.info("Initializing session...")
+    MCP Streamable HTTP Client
+    Make sure the server is running (ruby examples/streamable_http_server.rb)
+    #{"=" * 60}
+  MESSAGE
 
-  init_response = make_request(
+  http_transport = MCP::Client::HTTP.new(url: SERVER_URL)
-    nil,
+  client = MCP::Client.new(transport: http_transport)
-    "initialize",
+  sse_thread = nil
-    {
+
-      protocolVersion: PROTOCOL_VERSION,
+  begin
-      capabilities: {},
+    puts "=== Initializing session ==="
-      clientInfo: {
+    server_info = client.connect(
-        name: "sse-test-client",
+      client_info: { name: "streamable-http-client", version: "1.0" },
-        version: "1.0",
-      },
-    },
-    "init-1",
     )
 
-  if init_response[:error]
+    puts <<~MESSAGE
-    logger.error("Failed to initialize: #{init_response[:error]}")
+      ID: #{http_transport.session_id}
-    exit(1)
+      Version: #{http_transport.protocol_version}
+      Server: #{server_info["serverInfo"]}
+    MESSAGE
+
+    unless http_transport.session_id
+      logger.error("No session ID received; this example requires a stateful Streamable HTTP session.")
+      return
     end
 
-  session_id = init_response[:headers]["mcp-session-id"]&.first
+    puts "=== Listing tools ==="
+    tools = client.tools
+    tools.each { |tool| puts "  - #{tool.name}: #{tool.description}" }
 
-  if session_id.nil?
+    echo_tool = tools.find { |tool| tool.name == "echo" }
-    logger.error("No session ID received")
+    notification_tool = tools.find { |tool| tool.name == "notification_tool" }
-    exit(1)
-  end
 
-  if init_response[:body].dig("result", "capabilities", "logging")
+    sse_thread = Thread.new { connect_sse(http_transport.session_id, logger) }
-    make_request(session_id, "logging/setLevel", { level: "info" })
-  end
-
-  logger.info("Session initialized: #{session_id}")
-  logger.info("Server info: #{init_response[:body]["result"]["serverInfo"]}")
-
-  # Step 2: Start SSE connection in a separate thread
-  sse_thread = Thread.new { connect_sse(session_id, logger) }
-
-  # Give SSE time to connect
     sleep(1)
 
-  # Step 3: Interactive menu
+    # Once the optional SSE stream is active, POST requests may receive only a
+    # 202 ACK while the actual JSON-RPC response is delivered over SSE.
     loop do
-    puts <<~MESSAGE.chomp
+      puts <<~MENU.chomp
 
         === Available Actions ===
-      1. Send custom notification
+        1. Send notification (triggers SSE event)
-      2. Test echo
+        2. Echo message
-      3. List tools
+        3. Show cached tools
         0. Exit
 
         Choose an action:#{" "}
-    MESSAGE
+      MENU
 
-    choice = gets.chomp
+      case gets.chomp
-
-    case choice
       when "1"
+        if notification_tool
           print("Enter notification message: ")
           message = gets.chomp
           print("Enter delay in seconds (0 for immediate): ")
           delay = gets.chomp.to_f
 
-      response = make_request(
+          puts "=== Calling tool: notification_tool ==="
-        session_id,
+          response = client.call_tool(
-        "tools/call",
+            tool: notification_tool,
-        {
+            arguments: { message: message, delay: delay },
-          name: "notification_tool",
-          arguments: {
-            message: message,
-            delay: delay,
-          },
-        },
           )
-      if response[:body]["accepted"]
+          print_response(response)
-        logger.info("Notification sent successfully")
         else
-        logger.error("Error: #{response[:body]["error"]}")
+          puts "notification_tool not available"
         end
       when "2"
+        if echo_tool
           print("Enter message to echo: ")
           message = gets.chomp
-      make_request(session_id, "tools/call", { name: "echo", arguments: { message: message } })
+
+          puts "=== Calling tool: echo ==="
+          response = client.call_tool(tool: echo_tool, arguments: { message: message })
+          print_response(response)
+        else
+          puts "echo tool not available"
+        end
       when "3"
-      make_request(session_id, "tools/list")
+        puts "=== Cached tools ==="
+        tools.each { |tool| puts "  - #{tool.name}: #{tool.description}" }
       when "0"
         logger.info("Exiting...")
         break
@@ -186,22 +156,24 @@ def main
         puts "Invalid choice"
       end
     end
+  rescue MCP::Client::SessionExpiredError => e
+    logger.error("Session expired: #{e.message}")
+  rescue MCP::Client::RequestHandlerError => e
+    logger.error("Request error: #{e.message}")
+  rescue Interrupt
+    logger.info("Client interrupted")
+  rescue => e
+    logger.error("Error: #{e.message}")
+    logger.error(e.backtrace.first(5).join("\n"))
+  ensure
+    sse_thread.kill if sse_thread&.alive?
 
-  # Clean up
+    if http_transport.connected?
-  sse_thread.kill if sse_thread.alive?
+      puts "=== Closing session ==="
-
+      http_transport.close
-  # Close session
+      puts "Session closed"
-  logger.info("Closing session...")
+    end
-  make_request(session_id, "close")
+  end
-  logger.info("Session closed")
-rescue Interrupt
-  logger.info("Client interrupted by user")
-rescue => e
-  logger.error("Client error: #{e.message}")
-  logger.error(e.backtrace.join("\n"))
 end
 
-# Run the client
+main if __FILE__ == $PROGRAM_NAME
-if __FILE__ == $PROGRAM_NAME
-  main
-end