Class: ModelContextProtocol::Server::StreamableHttpTransport::SessionStore

Inherits:
Object
  • Object
show all
Defined in:
lib/model_context_protocol/server/streamable_http_transport/session_store.rb

Instance Method Summary collapse

Constructor Details

#initialize(redis_client, ttl: 3600) ⇒ SessionStore

Returns a new instance of SessionStore.



8
9
10
11
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 8

def initialize(redis_client, ttl: 3600)
  @redis = redis_client
  @ttl = ttl
end

Instance Method Details

#cleanup_session(session_id) ⇒ Object 



62
63
64
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 62

def cleanup_session(session_id)
  @redis.del("session:#{session_id}")
end

#create_session(session_id, data) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 13

def create_session(session_id, data)
  session_data = {
    id: session_id,
    server_instance: data[:server_instance],
    context: data[:context] || {},
    created_at: data[:created_at] || Time.now.to_f,
    last_activity: Time.now.to_f,
    active_stream: false
  }

  @redis.hset("session:#{session_id}", session_data.transform_values(&:to_json))
  @redis.expire("session:#{session_id}", @ttl)
  session_id
end

#get_registered_handlers(session_id) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 93

def get_registered_handlers(session_id)
  data = @redis.hmget("session:#{session_id}",
    "registered_prompts", "registered_resources", "registered_tools")

  # Return nil if none of the fields have meaningful data
  return nil if data.all? { |d| d.nil? || d.empty? }

  {
    prompts: (data[0] && !data[0].empty?) ? JSON.parse(data[0]) : [],
    resources: (data[1] && !data[1].empty?) ? JSON.parse(data[1]) : [],
    tools: (data[2] && !data[2].empty?) ? JSON.parse(data[2]) : []
  }
end

#get_session_context(session_id) ⇒ Object 



57
58
59
60
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 57

def get_session_context(session_id)
  context_data = @redis.hget("session:#{session_id}", "context")
  context_data ? JSON.parse(context_data, symbolize_names: true) : {}
end

#mark_stream_active(session_id, server_instance) ⇒ Object 



28
29
30
31
32
33
34
35
36
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 28

def mark_stream_active(session_id, server_instance)
  @redis.multi do |multi|
    multi.hset("session:#{session_id}",
      "active_stream", true.to_json,
      "stream_server", server_instance.to_json,
      "last_activity", Time.now.to_f.to_json)
    multi.expire("session:#{session_id}", @ttl)
  end
end

#mark_stream_inactive(session_id) ⇒ Object 



38
39
40
41
42
43
44
45
46
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 38

def mark_stream_inactive(session_id)
  @redis.multi do |multi|
    multi.hset("session:#{session_id}",
      "active_stream", false.to_json,
      "stream_server", nil.to_json,
      "last_activity", Time.now.to_f.to_json)
    multi.expire("session:#{session_id}", @ttl)
  end
end

#poll_messages_for_session(session_id) ⇒ Object 



76
77
78
79
80
81
82
83
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 76

def poll_messages_for_session(session_id)
  return [] unless session_exists?(session_id)

  queue = SessionMessageQueue.new(@redis, session_id, ttl: @ttl)
  queue.poll_messages
rescue
  []
end

#queue_message_for_session(session_id, message) ⇒ Object 



66
67
68
69
70
71
72
73
74
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 66

def queue_message_for_session(session_id, message)
  return false unless session_exists?(session_id)

  queue = SessionMessageQueue.new(@redis, session_id, ttl: @ttl)
  queue.push_message(message)
  true
rescue
  false
end

#session_exists?(session_id) ⇒ Boolean

Returns:



48
49
50
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 48

def session_exists?(session_id)
  @redis.exists("session:#{session_id}") == 1
end

#session_has_active_stream?(session_id) ⇒ Boolean

Returns:



52
53
54
55
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 52

def session_has_active_stream?(session_id)
  stream_data = @redis.hget("session:#{session_id}", "active_stream")
  stream_data ? JSON.parse(stream_data) : false
end

#store_registered_handlers(session_id, prompts:, resources:, tools:) ⇒ Object 



85
86
87
88
89
90
91
 
# File 'lib/model_context_protocol/server/streamable_http_transport/session_store.rb', line 85

def store_registered_handlers(session_id, prompts:, resources:, tools:)
  @redis.hset("session:#{session_id}",
    "registered_prompts", prompts.to_json,
    "registered_resources", resources.to_json,
    "registered_tools", tools.to_json)
  @redis.expire("session:#{session_id}", @ttl)
end