Class: ActionMCP::Session::Task

Inherits:
ApplicationRecord show all
Defined in:
app/models/action_mcp/session/task.rb

Overview

Represents a Task in an MCP session as per MCP 2025-11-25 specification. Tasks provide durable state machines for tracking async request execution.

State Machine:

working -> input_required -> working (via resume)
working -> completed | failed | cancelled
input_required -> completed | failed | cancelled

Constant Summary collapse 

"io.modelcontextprotocol/related-task"

Instance Method Summary collapse

Instance Method Details

#await_input!(prompt:, context: {}) ⇒ Object

Transition to input_required state and store pending input prompt

Parameters:

  • prompt (String) —

    The prompt/question for the user

  • context (Hash) (defaults to: {}) —

    Additional context about the input request



276
277
278
279
 
# File 'app/models/action_mcp/session/task.rb', line 276

def await_input!(prompt:, context: {})
  record_step!(:awaiting_input, data: { prompt: prompt, context: context })
  require_input!
end

#broadcast_status_change(transition = nil) ⇒ Object

Broadcast status change notification to the session

Parameters:

  • transition (StateMachines::Transition) (defaults to: nil) —

    The state transition that occurred



226
227
228
229
230
231
232
233
 
# File 'app/models/action_mcp/session/task.rb', line 226

def broadcast_status_change(transition = nil)
  return unless session

  handler = ActionMCP::Server::TransportHandler.new(session)
  handler.send_task_status_notification(self)
rescue StandardError => e
  Rails.logger.warn "Failed to broadcast task status change: #{e.message}"
end

#expired? ⇒ Boolean

TTL management

Returns:

  • (Boolean) —

    true if task has exceeded its TTL



132
133
134
135
136
137
 
# File 'app/models/action_mcp/session/task.rb', line 132

def expired?
  return false if ttl.nil?

  # TTL is stored in milliseconds (MCP spec)
  created_at + (ttl / 1000.0).seconds < Time.current
end

#non_terminal? ⇒ Boolean

Check if task is in a non-terminal state

Returns:

  • (Boolean) 


149
150
151
 
# File 'app/models/action_mcp/session/task.rb', line 149

def non_terminal?
  !terminal?
end

#record_step!(step_name, cursor: nil, data: {}) ⇒ Object

Record step execution state for job resumption

Parameters:

  • step_name (Symbol) —

    Name of the step

  • cursor (Integer, String) (defaults to: nil) —

    Optional cursor for resuming iteration

  • data (Hash) (defaults to: {}) —

    Additional step data to persist



241
242
243
244
245
246
247
248
249
250
251
 
# File 'app/models/action_mcp/session/task.rb', line 241

def record_step!(step_name, cursor: nil, data: {})
  update!(
    continuation_state: {
      step: step_name,
      cursor: cursor,
      data: data,
      timestamp: Time.current.iso8601
    },
    last_step_at: Time.current
  )
end

#related_task_meta ⇒ Object 



205
206
207
208
209
210
211
 
# File 'app/models/action_mcp/session/task.rb', line 205

def related_task_meta
  {
    RELATED_TASK_META_KEY => {
      "taskId" => id
    }
  }
end

#request_meta_with_related_task(meta = nil) ⇒ Object 



213
214
215
216
217
218
219
220
221
222
 
# File 'app/models/action_mcp/session/task.rb', line 213

def request_meta_with_related_task(meta = nil)
  existing_meta =
    if meta.respond_to?(:to_h)
      meta.to_h.deep_dup
    else
      {}
    end

  existing_meta.deep_merge(related_task_meta)
end

#result_ready? ⇒ Boolean

Returns:

  • (Boolean) 


144
145
146
 
# File 'app/models/action_mcp/session/task.rb', line 144

def result_ready?
  terminal? || input_required?
end

#resume_from_continuation! ⇒ void

This method returns an undefined value.

Resume task from input_required state and re-enqueue job



283
284
285
286
287
288
289
 
# File 'app/models/action_mcp/session/task.rb', line 283

def resume_from_continuation!
  return unless input_required?

  resume!
  # Re-enqueue the job to continue execution
  ActionMCP::ToolExecutionJob.perform_later(id, request_name, request_params, {})
end

#store_partial_result!(result_fragment) ⇒ Object

Store partial result fragment (for streaming/incremental results)

Parameters:

  • result_fragment (Hash) —

    Partial result to append



255
256
257
258
259
260
 
# File 'app/models/action_mcp/session/task.rb', line 255

def store_partial_result!(result_fragment)
  payload = result_payload || {}
  payload[:partial] ||= []
  payload[:partial] << result_fragment
  update!(result_payload: payload)
end

#terminal? ⇒ Boolean

Check if task is in a terminal state

Returns:

  • (Boolean) 


140
141
142
 
# File 'app/models/action_mcp/session/task.rb', line 140

def terminal?
  status.in?(%w[completed failed cancelled])
end

#to_create_task_result ⇒ Object 



169
170
171
172
173
174
 
# File 'app/models/action_mcp/session/task.rb', line 169

def to_create_task_result
  {
    task: to_task_data,
    _meta: related_task_meta
  }
end

#to_task_data ⇒ Hash

Convert to task data format per MCP spec

Returns:

  • (Hash) —

    Task data for JSON-RPC responses



155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'app/models/action_mcp/session/task.rb', line 155

def to_task_data
  data = {
    taskId: id,
    status: status,
    createdAt: created_at.iso8601(3),
    lastUpdatedAt: last_updated_at.iso8601(3),
    ttl: ttl
  }
  data[:statusMessage] = status_message if status_message.present?
  data[:pollInterval] = poll_interval if poll_interval.present?

  data
end

#to_task_error ⇒ Object 



190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'app/models/action_mcp/session/task.rb', line 190

def to_task_error
  return unless result_payload.is_a?(Hash)

  code = result_payload["code"] || result_payload[:code]
  message = result_payload["message"] || result_payload[:message]
  return unless code && message
  return if result_payload.key?("content") || result_payload.key?(:content)
  return if result_payload.key?("isError") || result_payload.key?(:isError)

  error = { code: code, message: message }
  data = result_payload["data"] || result_payload[:data]
  error[:data] = data unless data.nil?
  error
end

#to_task_result ⇒ Hash

Convert to the original request’s result payload for tasks/result. The result carries related-task metadata because its structure does not otherwise include the task identifier.

Returns:

  • (Hash) —

    Result payload for tasks/result response



180
181
182
183
184
185
186
187
188
 
# File 'app/models/action_mcp/session/task.rb', line 180

def to_task_result
  payload = result_payload.is_a?(Hash) ? result_payload.deep_dup : {}
  meta = payload.delete("_meta") || payload.delete(:_meta) || {}
  meta = meta.to_h if meta.respond_to?(:to_h)
  meta = {} unless meta.is_a?(Hash)

  payload[:_meta] = meta.deep_merge(related_task_meta)
  payload
end

#update_progress!(percent:, message: nil) ⇒ Object

Update progress indicators for long-running tasks

Parameters:

  • percent (Integer) —

    Progress percentage (0-100)

  • message (String) (defaults to: nil) —

    Optional progress message



265
266
267
268
269
270
271
 
# File 'app/models/action_mcp/session/task.rb', line 265

def update_progress!(percent:, message: nil)
  update!(
    progress_percent: percent.clamp(0, 100),
    progress_message: message,
    last_step_at: Time.current
  )
end