Class: AgUi::Server::Middleware::SSEStream

Inherits:
Object
  • Object
show all
Defined in:
lib/ag_ui/server/middleware/sse_stream.rb

Overview

Sets up an SSE stream builder on env["ag_ui.stream"], ported from a2a's Server::Middleware::SSEStream.

The open block runs inside an Async fiber and the stream is automatically finished when the block exits (even on exception). If the handler never calls open, the builder is removed from env so the response layer doesn't mistake it for a real stream.

Usage (inside the /run handler):

env["ag_ui.stream"].open(thread_id:, run_id:) do |s|
s.run_started
s.text_message_content(message_id: "m1", delta: "Hi")
s.run_finished
end

Instance Method Summary collapse

Constructor Details

#initialize(app) ⇒ SSEStream

Returns a new instance of SSEStream.



27
28
29
 
# File 'lib/ag_ui/server/middleware/sse_stream.rb', line 27

def initialize(app)
  @app = app
end

Instance Method Details

#call(env) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/ag_ui/server/middleware/sse_stream.rb', line 31

def call(env)
  builder = StreamBuilder.new(env)
  env["ag_ui.stream"] = builder

  result = @app.call(env)

  # If open was never called, clear the builder so the response
  # layer doesn't mistake it for a real stream.
  if env["ag_ui.stream"].equal?(builder)
    env.delete("ag_ui.stream")
  end

  result
end