Tasks

The Tasks extension (io.modelcontextprotocol/tasks) lets a tool answer tools/call with a task instead of a result. The client polls the task with tasks/get until it finishes, answers its input requests with tasks/update, and may cancel it with tasks/cancel. Tasks require MCP 2026-07-28.

Phantom handles the protocol. You store the tasks, run the work, and move each task through its statuses:

Returning a task

A tool returns a Phantom.Tasks struct like any other result. Only answer with a task when the client declared the extension on this request. Otherwise, do the work in a process, report progress, and respond when it finishes:

def export_report(%{"report_id" => report_id} = params, session) do
  if Phantom.Session.tasks_supported?(session) do
    {:reply, start_export(params, session), session}
  else
    Task.async(fn ->
      csv =
        Reports.export!(report_id,
          on_progress: &Phantom.Session.notify_progress(session, &1, &2, "Exporting")
        )

      Phantom.Session.respond(session, Tool.text(csv))
    end)

    {:noreply, session}
  end
end

If a tool returns a task anyway, Phantom answers with the -32021 missing capability error. A tool that only works as a task can return that error itself, before creating anything:

{:error, Phantom.Request.missing_task_capability(), session}

Router callbacks

Implementing Phantom.Router.get_task/2 enables the extension:

• Phantom.Router.get_task/2 — fetch a task for tasks/get, tasks/update, and tasks/cancel. Return {:error, :not_found} when the session may not see it, so the client cannot tell whether it exists.
• Phantom.Router.update_task/3 — receive answers to the task's input_requests. Without it, Phantom acknowledges tasks/update and drops the answers.
• Phantom.Router.cancel_task/2 — cancel the work. Without it, Phantom acknowledges tasks/cancel and the task carries on, which the spec allows.
• Phantom.Router.authorize_task_subscriptions/2 — which tasks a subscriptions/listen stream may follow. Defaults to the tasks get_task/2 returns; override it to check many with one query. Before each notification, the stream fetches the task with get_task/2, which checks access again and sends the stored task.

Example: Ecto and Oban

This example stores tasks in Postgres and runs them with Oban. Inserting the task and its job in one transaction means a task never exists without its job, or the reverse.

Migration

defmodule MyApp.Repo.Migrations.CreateMCPTasks do
  use Ecto.Migration

  def change do
    create table(:mcp_tasks, primary_key: false) do
      add :id, :string, primary_key: true
      add :user_id, references(:users, on_delete: :delete_all), null: false
      add :status, :string, null: false
      add :status_message, :string
      add :input_requests, :map, null: false, default: %{}
      add :input_responses, :map, null: false, default: %{}
      add :result, :map
      add :error, :map
      add :job_id, :bigint
      add :expires_at, :utc_datetime_usec, null: false

      timestamps(type: :utc_datetime_usec)
    end

    create index(:mcp_tasks, [:user_id])
    create index(:mcp_tasks, [:expires_at])
  end
end

Schema and context

defmodule MyApp.MCP.TaskRecord do
  use Ecto.Schema

  @primary_key {:id, :string, autogenerate: false}
  @timestamps_opts [type: :utc_datetime_usec]

  schema "mcp_tasks" do
    belongs_to :user, MyApp.Accounts.User
    field :status, Ecto.Enum, values: [:working, :input_required, :completed, :failed, :cancelled]
    field :status_message, :string
    field :input_requests, :map, default: %{}
    field :input_responses, :map, default: %{}
    field :result, :map
    field :error, :map
    field :job_id, :integer
    field :expires_at, :utc_datetime_usec

    timestamps()
  end
end

defmodule MyApp.MCP.TaskRecords do
  import Ecto.Query

  alias MyApp.MCP.TaskRecord
  alias MyApp.Repo

  @ttl :timer.hours(6)

  def create!(user, attrs \\ %{}) do
    now = DateTime.utc_now()

    %TaskRecord{
      id: Phantom.Tasks.generate_id(),
      user_id: user.id,
      status: :working,
      expires_at: DateTime.add(now, @ttl, :millisecond)
    }
    |> Ecto.Changeset.change(attrs)
    |> Repo.insert!()
  end

  def get(id, user), do: Repo.get_by(TaskRecord, id: id, user_id: user.id)
  def get!(id), do: Repo.get!(TaskRecord, id)

  def to_phantom(%TaskRecord{} = record) do
    Phantom.Tasks.new(
      id: record.id,
      status: record.status,
      status_message: record.status_message,
      created_at: record.inserted_at,
      last_updated_at: record.updated_at,
      ttl_ms: DateTime.diff(record.expires_at, record.inserted_at, :millisecond),
      poll_interval_ms: 2_000,
      input_requests: record.input_requests,
      result: record.result,
      error: record.error
    )
  end

  def working(id, message), do: transition(id, fn _ -> %{status: :working, status_message: message} end)

  def complete(id, result),
    do: transition(id, fn _ -> %{status: :completed, result: result, input_requests: %{}} end)

  def fail(id, error),
    do: transition(id, fn _ -> %{status: :failed, error: error, input_requests: %{}} end)

  def cancel(id),
    do: transition(id, fn _ -> %{status: :cancelled, status_message: "Cancelled"} end)

  def need_input(id, requests),
    do: transition(id, fn _ -> %{status: :input_required, input_requests: requests} end)

  # Answered requests are removed; the task works again once none remain.
  def answer(id, responses) do
    transition(id, fn record ->
      remaining = Map.drop(record.input_requests, Map.keys(responses))

      %{
        status: if(remaining == %{}, do: :working, else: :input_required),
        input_requests: remaining,
        input_responses: Map.merge(record.input_responses, responses)
      }
    end)
  end

  # Only a running task changes. A finished task stays finished, so a job
  # that completes after the client cancelled leaves it cancelled.
  defp transition(id, fun) do
    result =
      Repo.transact(fn ->
        query =
          from t in TaskRecord,
            where: t.id == ^id and t.status in [:working, :input_required],
            lock: "FOR UPDATE"

        case Repo.one(query) do
          nil -> {:error, :finished}
          record -> record |> Ecto.Changeset.change(fun.(record)) |> Repo.update()
        end
      end)

    with {:ok, record} <- result do
      Phantom.Tracker.notify_task_updated(to_phantom(record))
    end

    result
  end

  def prune do
    Repo.delete_all(from t in TaskRecord, where: t.expires_at < ^DateTime.utc_now())
  end
end

Repo.transact/2 requires Ecto 3.13. Use Repo.transaction/1 on older versions.

The tool and its job

defmodule MyApp.MCP.Router do
  use Phantom.Router, name: "MyApp", vsn: "1.0"

  alias MyApp.MCP.TaskRecords

  @description "Export a report as CSV. Large reports run as a task."
  tool :export_report do
    field :report_id, :integer, required: true
  end

  def export_report(%{"report_id" => report_id}, session) do
    if Phantom.Session.tasks_supported?(session) do
      {:ok, record} =
        MyApp.Repo.transact(fn ->
          record = TaskRecords.create!(session.assigns.current_user)
          job = Oban.insert!(MyApp.ExportWorker.new(%{task_id: record.id, report_id: report_id}))
          {:ok, MyApp.Repo.update!(Ecto.Changeset.change(record, job_id: job.id))}
        end)

      {:reply, TaskRecords.to_phantom(record), session}
    else
      Task.async(fn ->
        csv =
          MyApp.Reports.export!(report_id,
            on_progress: &Phantom.Session.notify_progress(session, &1, &2, "Exporting")
          )

        Phantom.Session.respond(session, Tool.text(csv))
      end)

      {:noreply, session}
    end
  end

  def get_task(id, session) do
    case TaskRecords.get(id, session.assigns.current_user) do
      nil -> {:error, :not_found}
      record -> {:ok, TaskRecords.to_phantom(record)}
    end
  end

  # Once every request is answered, run the job that asked for input again.
  def update_task(task, responses, _session) do
    case TaskRecords.answer(task.id, responses) do
      {:ok, %{status: :working, job_id: job_id}} -> Oban.retry_job(job_id)
      _ -> :ok
    end
  end

  def cancel_task(task, _session) do
    with {:ok, record} <- TaskRecords.cancel(task.id), do: Oban.cancel_job(record.job_id)
    :ok
  end
end

The worker reports through the context. A tool error is still a completed task with isError: true; failed is only for JSON-RPC errors.

defmodule MyApp.ExportWorker do
  use Oban.Worker, queue: :exports, max_attempts: 3

  alias MyApp.MCP.TaskRecords
  alias Phantom.Tool

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"task_id" => id, "report_id" => report_id}}) do
    TaskRecords.working(id, "Rendering report #{report_id}")

    case MyApp.Reports.export(report_id) do
      {:ok, csv} ->
        TaskRecords.complete(id, Tool.response(Tool.text(csv)))
        :ok

      {:error, :too_large} ->
        TaskRecords.complete(id, Tool.error("The report is too large"))
        :ok

      # Oban retries the job; the task stays working meanwhile.
      {:error, reason} ->
        {:error, reason}
    end
  end
end

A job that crashes on its last attempt would leave its task working until its TTL. Fail the task from Oban's telemetry instead:

# In your application's start/2
:telemetry.attach("mcp-task-failures", [:oban, :job, :exception], &MyApp.MCP.TaskTelemetry.handle_exception/4, nil)

defmodule MyApp.MCP.TaskTelemetry do
  alias MyApp.MCP.TaskRecords

  def handle_exception(_event, _measure, %{job: %{args: %{"task_id" => id}} = job} = meta, _config)
      when meta.state == :discard or job.attempt >= job.max_attempts do
    TaskRecords.fail(id, Phantom.Request.internal_error("The task stopped unexpectedly"))
  end

  def handle_exception(_event, _measure, _meta, _config), do: :ok
end

Prune expired tasks with Oban's cron plugin:

config :my_app, Oban,
  plugins: [{Oban.Plugins.Cron, crontab: [{"@hourly", MyApp.PruneMCPTasksWorker}]}]

defmodule MyApp.PruneMCPTasksWorker do
  use Oban.Worker, queue: :maintenance

  @impl Oban.Worker
  def perform(_job) do
    MyApp.MCP.TaskRecords.prune()
    :ok
  end
end

Asking for input during a task

A task in input_required lists its requests, and the client answers them with tasks/update. Each key must be new for the task's lifetime. The update_task/3 callback above stores the answers and retries the job once none are outstanding, so the job reads them from the record. Here, a job for an export_all tool confirms first:

def perform(%Oban.Job{args: %{"task_id" => id}}) do
  record = TaskRecords.get!(id)

  case record.input_responses do
    %{"confirm" => %{"action" => "accept"}} ->
      TaskRecords.complete(id, Tool.response(Tool.text(MyApp.Reports.export_all!())))

    %{"confirm" => _declined} ->
      TaskRecords.complete(id, Tool.response(Tool.text("Export skipped")))

    _ ->
      elicit =
        Phantom.Elicit.form(%{
          message: "Export every report? This may take a while.",
          requested_schema: [%{name: "confirm", type: :boolean, required: true}]
        })

      TaskRecords.need_input(id, %{"confirm" => Phantom.Elicit.to_input_request(elicit)})
  end

  :ok
end

Example: Oban Pro

With Oban Pro, a worker hook can fail the task when the job is discarded, in place of the telemetry handler. after_process/3 runs for :complete, :cancel, :discard, :error, and :snooze; Phantom.Router.cancel_task/2 already records cancellations:

defmodule MyApp.ExportWorker do
  use Oban.Pro.Worker, queue: :exports, max_attempts: 3

  alias MyApp.MCP.TaskRecords
  alias Phantom.Tool

  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"report_id" => report_id} = args}) do
    mark_working(args, "Rendering report #{report_id}")
    progress = &Phantom.Session.notify_progress(args["progress"], &1, &2, "Exporting")

    with {:ok, csv} <- MyApp.Reports.export(report_id, on_progress: progress) do
      complete(args, csv)
      {:ok, csv}
    end
  end

  @impl Oban.Pro.Worker
  def after_process(:discard, %{args: %{"task_id" => id}}, _result),
    do: TaskRecords.fail(id, Phantom.Request.internal_error("The export failed"))

  def after_process(_state, _job, _result), do: :ok

  defp mark_working(%{"task_id" => id}, message), do: TaskRecords.working(id, message)
  defp mark_working(_args, _message), do: :ok

  defp complete(%{"task_id" => id}, csv), do: TaskRecords.complete(id, Tool.response(Tool.text(csv)))
  defp complete(_args, _csv), do: :ok
end

The worker returns {:ok, csv}, so the same job also serves clients without tasks. Oban.Pro.Relay inserts it and awaits its result from any node, and the tool responds when it arrives. Phantom.Session.progress_ref/1 lets the job report progress to the waiting request from whichever node runs it; with no ref in the arguments, notify_progress/4 does nothing:

alias Oban.Pro.Relay

def export_report(%{"report_id" => report_id}, session) do
  if Phantom.Session.tasks_supported?(session) do
    # ...insert the task and the job, as above
  else
    Task.async(fn ->
      response =
        %{report_id: report_id, progress: Phantom.Session.progress_ref(session)}
        |> MyApp.ExportWorker.new()
        |> Relay.async()
        |> Relay.await(timeout: :timer.minutes(5))
        |> case do
          {:ok, csv} -> Tool.text(csv)
          {:error, :timeout} -> Tool.error("The export took too long")
          _failed -> Tool.error("The export failed")
        end

      Phantom.Session.respond(session, response)
    end)

    {:noreply, session}
  end
end

Relay needs a working PubSub notifier, and the Postgres notifier limits results to 8kb compressed; use the PG notifier for larger results.

A workflow splits the task into steps. Each step reports its progress as the status message, and the last step builds the result from the steps' recorded output:

alias Oban.Pro.Workflow

def export_report(%{"report_id" => report_id}, session) do
  {:ok, record} =
    MyApp.Repo.transact(fn ->
      record = TaskRecords.create!(session.assigns.current_user)
      args = %{task_id: record.id, report_id: report_id}

      Workflow.new()
      |> Workflow.add(:fetch, MyApp.FetchWorker.new(args))
      |> Workflow.add(:render, MyApp.RenderWorker.new(args), deps: :fetch)
      |> Workflow.add(:finish, MyApp.FinishWorker.new(args), deps: :render)
      |> Oban.insert_all()

      {:ok, record}
    end)

  {:reply, TaskRecords.to_phantom(record), session}
end

Each step that the next one reads must be recorded. MyApp.FetchWorker uses recorded: true and returns {:ok, rows} the same way.

defmodule MyApp.RenderWorker do
  use Oban.Pro.Worker, queue: :exports, recorded: true

  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"task_id" => id}} = job) do
    TaskRecords.working(id, "Rendering")
    rows = Workflow.get_recorded(job, :fetch)
    {:ok, MyApp.Reports.to_csv(rows)}
  end
end

defmodule MyApp.FinishWorker do
  use Oban.Pro.Worker, queue: :exports

  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"task_id" => id}} = job) do
    csv = Workflow.get_recorded(job, :render)
    TaskRecords.complete(id, Phantom.Tool.response(Phantom.Tool.text(csv)))
    :ok
  end
end

To cancel a workflow, cancel its jobs in Phantom.Router.cancel_task/2 with Workflow.cancel_jobs/3.

Testing

Phantom.Test sends task requests with the extension declared. Build the session with the stateless protocol. With Oban's testing: :inline, jobs run inside Oban.insert/2, so the task is finished by the time the tool replies:

setup do
  Phantom.Test.start(router: MyApp.MCP.Router)
  user = insert(:user)
  {:ok, session: build_session(MyApp.MCP.Router, protocol_version: "2026-07-28", assigns: %{current_user: user})}
end

test "exports a report as a task", %{session: session} do
  %{taskId: id} =
    session
    |> call_tool(:export_report, %{report_id: 1}, tasks: true)
    |> assert_task(status: :working)

  session
  |> get_task(id)
  |> assert_task(status: :completed)
end

Use update_task/4 and cancel_task/3 for the other methods.

Load balancers

Clients send the task ID in the Mcp-Name header on tasks/get, tasks/update, and tasks/cancel, and Phantom rejects a mismatch. When tasks live in a shared database, any node can answer them. When they live in one node's memory, route on Mcp-Name.