Module: OKF::MCP::HTTP

Defined in:
lib/okf/mcp/http.rb

Overview

Bridges the SDK's StreamableHTTPTransport — a plain Rack app — onto the WEBrick that okf already depends on, mirroring the kernel's server/runner pattern (rack and webrick both arrive via the okf gem). Stateless JSON mode: every POST is a self-contained JSON-RPC exchange answered with a single JSON object, so plain buffered responses hold — no SSE stream to keep open. The SDK validates Host/Origin against DNS rebinding for locally bound servers.

Defined Under Namespace

Classes: Stream, Streams

Constant Summary collapse

LOOPBACK_HOSTS =

The Hosts the SDK's StreamableHTTPTransport admits without any extra allowlist — its DNS-rebinding protection accepts these out of the box.

%w[127.0.0.1 ::1 localhost].freeze
WILDCARD_BINDS =

Binds that mean "every interface" rather than one address.

%w[0.0.0.0 :: *].freeze
MAX_REQUEST_BYTES =

The largest request body the bridge will hand the transport — the Rack seam's constant, aliased because #read_body enforces it here too: anything past it is 413 before it is allocated.

App::MAX_REQUEST_BYTES
MAX_LISTEN_STREAMS =

Cap on concurrent subscriptions/listen streams, far below the SDK's 1000 default because the costs differ in kind: under a Rack 3 server a stream holds no thread, but on this bridge each one parks a WEBrick handler thread and occupies one of WEBrick's 100 connection tokens — at the SDK default the tokens exhaust at 100 and every tool call queues behind held streams. 32 leaves two-thirds of the tokens for request traffic. A constant, not a flag: zero-config is this mode's posture, and an operator who needs more has the Rack seam.

32

Class Method Summary collapse

Class Method Details

.allowed_hosts_for(bind, extra: []) ⇒ Object 



224
225
226
227
228
229
230
 
# File 'lib/okf/mcp/http.rb', line 224

def allowed_hosts_for(bind, extra: [])
  extra = Array(extra).reject { |host| OKF.blank?(host) }
  return (extra.empty? ? nil : extra) if LOOPBACK_HOSTS.include?(bind)

  hosts = WILDCARD_BINDS.include?(bind.to_s) ? local_hosts : [ bind ]
  (hosts + extra).uniq
end

.announce(httpd, bind:, out:) ⇒ Object

The Host allowlist a bind address needs: nil for loopback, which the SDK already admits.

A wildcard bind is the case that was broken. --bind 0.0.0.0 says "serve every interface", and allowlisting the literal string "0.0.0.0" allowlists a Host header no client ever sends: the server bound everything and answered every real request with 403. What a client actually sends is the address it dialled, so a wildcard expands to this machine's own addresses plus its hostname.

extra (the repeatable --allow-host) covers what cannot be derived: a DNS name or a reverse proxy's Host, which no local interface knows. The boot line, plus what a non-loopback bind actually means.

The Host allowlist below is a defence against DNS rebinding — a browser walked into this port by a page the reader never meant to give it to. It is not access control and must never be sold as one: a client that is not a browser sets Host to whatever it likes, and there is no authentication behind it. So binding anywhere but loopback publishes every served bundle to anything that can reach the port, and the boot line says so rather than reading like a URL somebody can safely share.

Loopback stays quiet: it is the default and the posture the tool was built for, and a warning printed every time is a warning nobody reads.



202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/okf/mcp/http.rb', line 202

def announce(httpd, bind:, out:)
  # The listener's own port, not the asked-for one: with --port 0 the OS
  # picks, and a boot line reporting 0 would name a port nothing answers.
  bound = httpd.listeners.first.addr[1]
  say(out, "okf-mcp listening on http://#{bind}:#{bound}")
  return if LOOPBACK_HOSTS.include?(bind.to_s)

  say(out, "okf-mcp: WARNING — #{bind} is not loopback. Every served bundle is")
  say(out, "  readable by anything that can reach this port, with no authentication. The Host")
  say(out, "  allowlist only stops browser DNS rebinding; it is not access control.")
end

.app_for(server, bind:, allowed_hosts: allowed_hosts_for(bind), **listen_options) ⇒ Object

The SDK transport wired to the server, in stateless JSON mode. A non-loopback bind (e.g. 0.0.0.0) is refused by the SDK's DNS-rebinding guard unless its Host is allowlisted; loopback binds keep the SDK defaults. Protection itself stays on either way. listen_options passes max_listen_subscriptions: / listen_keepalive_interval: through to the SDK — real configuration for a caller composing the bridge directly; the CLI keeps the defaults above.



173
174
175
176
 
# File 'lib/okf/mcp/http.rb', line 173

def app_for(server, bind:, allowed_hosts: allowed_hosts_for(bind), **listen_options)
  App.transport(server, allowed_hosts: allowed_hosts,
    max_listen_subscriptions: MAX_LISTEN_STREAMS, **listen_options)
end

.build(app, bind:, port:) ⇒ Object

Returns an unstarted server so tests can drive an ephemeral port. The stream ledger rides the server instance — one per bridge, reachable from #stop — because the module itself serves many servers at once under the test suite.



247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/okf/mcp/http.rb', line 247

def build(app, bind:, port:)
  streams = Streams.new
  httpd = WEBrick::HTTPServer.new(
    BindAddress: bind,
    Port: port,
    Logger: WEBrick::Log.new($stderr, WEBrick::Log::WARN),
    AccessLog: []
  )
  httpd.config[:okf_mcp_streams] = streams
  httpd.mount_proc("/") { |request, response| handle(app, request, response, streams) }
  httpd
end

.env_for(request, body = request.body.to_s) ⇒ Object 



368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
 
# File 'lib/okf/mcp/http.rb', line 368

def env_for(request, body = request.body.to_s)
  env = {
    "REQUEST_METHOD" => request.request_method,
    "SCRIPT_NAME" => "",
    "PATH_INFO" => request.path,
    "QUERY_STRING" => request.query_string.to_s,
    "SERVER_NAME" => request.host.to_s,
    "SERVER_PORT" => request.port.to_s,
    "rack.url_scheme" => "http",
    "rack.input" => StringIO.new(body.to_s),
    "rack.errors" => $stderr
  }
  request.each do |name, value|
    key = name.upcase.tr("-", "_")
    key = "HTTP_#{key}" unless %w[CONTENT_TYPE CONTENT_LENGTH].include?(key)
    env[key] = value
  end
  env
end

.handle(app, request, response, streams = nil) ⇒ Object 



260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/okf/mcp/http.rb', line 260

def handle(app, request, response, streams = nil)
  # The MCP endpoint is the root and nothing else. The SDK transport
  # routes on method alone, so handing it every path answered the OAuth
  # discovery probes a connecting host sends first (GET /.well-known/*,
  # POST /register — Claude Desktop does) with a 405 and a 200-wrapped
  # JSON-RPC parse error: a *broken* sign-in service instead of an
  # absent one, and the host refused the connector on it. 404 is the
  # answer that reads as absence.
  return not_found(response) unless request.path == "/"

  # The cap is enforced *here*, before the body is materialized. The SDK
  # transport has its own `max_request_bytes` and never reads more than
  # that off `rack.input` — but WEBrick's `request.body` has no limit, so
  # handing the transport a StringIO of it meant the allocation the cap
  # exists to prevent had already happened. A few concurrent 2 GB POSTs
  # would OOM the warm shared process that `--http` exists to provide.
  body = read_body(request)
  return oversized(response) if body.nil?

  status, headers, out = app.call(env_for(request, body))
  response.status = status
  headers.each { |name, value| response[name] = value }
  # A callable body is the Rack 3 streaming shape — the SDK's
  # `subscriptions/listen` answers with one — and buffering it here
  # would block forever on a stream that only ends when the peer goes.
  # The close is per-branch, not a method-level ensure: WEBrick invokes
  # a streaming body only after this method returns, and a callable
  # that also responds to close (a shape Rack 3 sanctions) must not be
  # closed before it is served.
  if out.respond_to?(:call)
    stream_response(response, out, streams)
  else
    begin
      buffer = String.new
      out.each { |chunk| buffer << chunk }
      response.body = buffer
    ensure
      out.close if out.respond_to?(:close)
    end
  end
end

.local_hosts ⇒ Object

Every address this machine answers on, plus its hostname. Loopback is left out only because the SDK admits it already.



234
235
236
237
238
239
240
241
 
# File 'lib/okf/mcp/http.rb', line 234

def local_hosts
  addresses = Socket.ip_address_list.reject(&:ipv4_loopback?).reject(&:ipv6_loopback?)
  names = addresses.map(&:ip_address)
  names << Socket.gethostname.to_s
  names.reject { |name| OKF.blank?(name) }.uniq
rescue SocketError, SystemCallError
  []
end

.not_found(response) ⇒ Object 



353
354
355
356
357
 
# File 'lib/okf/mcp/http.rb', line 353

def not_found(response)
  response.status = 404
  response["Content-Type"] = "application/json"
  response.body = JSON.generate(error: "not found: the MCP endpoint is /")
end

.oversized(response) ⇒ Object 



359
360
361
362
363
364
365
366
 
# File 'lib/okf/mcp/http.rb', line 359

def oversized(response)
  response.status = 413
  response["Content-Type"] = "application/json"
  response.body = JSON.generate(
    jsonrpc: "2.0", id: nil,
    error: { code: -32_600, message: "Request body exceeds #{MAX_REQUEST_BYTES} bytes" }
  )
end

.prepare(server, bind:, port:, allow_hosts: [], out: $stderr) ⇒ Object

Everything before accepting — the bind (where EADDRINUSE, the boot failure that actually happens, raises), the traps and the boot line — split from #start so the CLI can run this under its boot rescue and file a bind failure as the usage error it is, while a mid-serve errno out of #start stays the crash it is.



139
140
141
142
143
144
145
146
147
 
# File 'lib/okf/mcp/http.rb', line 139

def prepare(server, bind:, port:, allow_hosts: [], out: $stderr)
  app = app_for(server, bind: bind, allowed_hosts: allowed_hosts_for(bind, extra: allow_hosts))
  httpd = build(app, bind: bind, port: port)
  # The trap spawns a thread because #stop takes the transport's mutex,
  # and a mutex inside trap context is ThreadError on Ruby 2.7.
  %w[INT TERM].each { |signal| trap(signal) { Thread.new { stop(httpd, app) } } }
  announce(httpd, bind: bind, out: out)
  httpd
end

.read_body(request) ⇒ Object

The request body, or nil when it exceeds MAX_REQUEST_BYTES. A declared Content-Length past the cap is refused without reading a byte; an undeclared (chunked) body is streamed and abandoned the moment it grows past it.



334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/okf/mcp/http.rb', line 334

def read_body(request)
  # The raw header, not WEBrick's #content_length: that helper is
  # `Integer(self["content-length"])`, which raises TypeError on a
  # chunked request that legitimately carries no Content-Length at all.
  declared = request["content-length"]
  return nil if declared && declared.to_i > MAX_REQUEST_BYTES

  # Always the block form: `request.body` with no block materializes the
  # whole entity body, which for an undeclared (chunked) request is the
  # very allocation this method exists to bound. With a block WEBrick
  # streams, and a request carrying no body simply never yields.
  buffer = String.new
  request.body do |chunk|
    buffer << chunk
    return nil if buffer.bytesize > MAX_REQUEST_BYTES
  end
  buffer
end

.say(out, line) ⇒ Object

The boot line is diagnostics, and diagnostics are best-effort: stderr belongs to whoever spawned the process, and a collector that died must not take a bound, healthy server down with it. An EPIPE here is lost output, not a lost server.



218
219
220
221
222
 
# File 'lib/okf/mcp/http.rb', line 218

def say(out, line)
  out.puts(line)
rescue Errno::EPIPE, Errno::ECONNRESET
  nil
end

.stop(httpd, app) ⇒ Object

Teardown in the only order that terminates: the transport first, so every open listen stream is closed and its parked handler thread returns (see #stream_response) — WEBrick's own shutdown joins the connection threads, so closing it first would hang on any open stream.

The ensure is the signal's guarantee: a supervisor sends exactly one TERM, so whatever the transport's close raises, the stream latch still trips (a listen racing the close is closed on admission, never parked) and WEBrick still comes down. One signal, one dead server, always.



158
159
160
161
162
163
 
# File 'lib/okf/mcp/http.rb', line 158

def stop(httpd, app)
  app.close
ensure
  httpd.config[:okf_mcp_streams]&.close_all
  httpd.shutdown
end

.stream_response(response, body, streams = nil) ⇒ Object

Serves a Rack streaming body through WEBrick's proc-body path: with chunked = true, WEBrick calls the proc with a ChunkedWrapper after the headers are out, and finalizes the response when it returns. The SDK's callable returns immediately (it registers the stream, writes the acknowledgement, and starts its keepalive thread), so the proc parks this handler thread in Stream#wait until the SDK ends the stream — a dead peer's EPIPE out of a keepalive write, or the transport's close on shutdown.



310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/okf/mcp/http.rb', line 310

def stream_response(response, body, streams = nil)
  response.keep_alive = false # an SSE stream ends with its connection
  response.chunked = true
  response.body = lambda do |wire|
    stream = Stream.new(wire)
    # Admitted before the SDK sees it: a stream arriving after the stop
    # latch tripped is closed here and now, so the SDK's first write
    # raises IOError into its own cleanup and this thread never parks.
    streams.admit(stream) if streams
    begin
      body.call(stream)
      stream.wait
    ensure
      stream.close # idempotent; covers a body that raised
      streams.discard(stream) if streams
      body.close if body.respond_to?(:close)
    end
  end
end