Class: A2A::Server::Middleware::SSEStream

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

Overview

Sets up an SSE stream builder on env["a2a.stream"].

The builder detects the protocol binding (REST vs JSON-RPC) from env and creates the correct stream subclass when open is called. The open block runs inside an Async fiber and the stream is automatically finished when the block exits (even on exception).

If the agent never calls open, the builder is removed from env so the binding layer doesn't mistake it for a real stream.

Part of the A2A::Server middleware stack, so the agent gets env without declaring anything.

Usage:

case env["a2a.operation"]
in "SendStreamingMessage"
env["a2a.stream"].open(task_id: "t1", context_id: "c1") do |s|
  s.task(status: { state: "TASK_STATE_WORKING" })
  s.status_update(status: { state: "TASK_STATE_COMPLETED" })
end
end

Instance Method Summary collapse

Constructor Details

#initialize(app) ⇒ SSEStream

Returns a new instance of SSEStream.



35
36
37
 
# File 'lib/a2a/server/middleware/sse_stream.rb', line 35

def initialize(app)
  @app = app
end

Instance Method Details

#call(env) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/a2a/server/middleware/sse_stream.rb', line 39

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

  result = @app.call(env)

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

  result
end