diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bbaaec4..8d374d1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,6 +19,8 @@ jobs: otp: "27" - elixir: "1.20" otp: "29" + # Formatter output differs between Elixir versions. + check_format: true steps: - uses: actions/checkout@v7 @@ -52,6 +54,7 @@ jobs: run: mix deps.unlock --check-unused - name: Check formatting + if: matrix.check_format run: mix format --check-formatted - name: Compile (warnings as errors) diff --git a/CHANGELOG.md b/CHANGELOG.md index e209ca0..4bfc813 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,30 @@ ## Unreleased +- New: the Tasks extension (`io.modelcontextprotocol/tasks`) under MCP + 2026-07-28. A tool returns `{:reply, %Phantom.Tasks{}, session}` to answer + with a task, and the router implements `c:Phantom.Router.get_task/2`, and + optionally `update_task/3`, `cancel_task/2`, and + `authorize_task_subscriptions/2`, for `tasks/get`, `tasks/update`, and + `tasks/cancel`. Phantom does not store tasks. See the Tasks guide for Ecto + and Oban examples. +- New: `Phantom.Session.tasks_supported?/1`, and + `Phantom.Tracker.notify_task_updated/1`. Each `subscriptions/listen` stream + that follows the task fetches it with `get_task/2` and sends + `notifications/tasks`. +- New: `Phantom.Test.get_task/3`, `update_task/4`, `cancel_task/3`, + `assert_task/2`, and a `:tasks` option for `call_tool/4`. +- New: `Phantom.Session.progress_ref/1` returns a JSON-safe reference to the + current request. `Phantom.Session.notify_progress/4` accepts it, so a job + on any node can report progress to a waiting `tools/call` or `prompts/get` + through PubSub. +- New: `Phantom.Elicit.to_input_request/1` renders one embedded + `elicitation/create` request to place under your own `inputRequests` key, + such as a task's input requests. +- New: `Phantom.Request.missing_task_capability/0`, the error for a tool that + only runs as a task when the client did not declare the extension. +- `Phantom.Request.missing_capability/1` accepts a `ClientCapabilities` map, + for capabilities whose names contain dots. + Streamable HTTP transport fixes for the protocol versions before 2026-07-28: - New: `Phantom.Plug` option `validate_session: true` answers `404 Not Found` diff --git a/README.md b/README.md index beba52a..cfdaf94 100644 --- a/README.md +++ b/README.md @@ -717,6 +717,10 @@ Phantom will implement these MCP requests on your behalf: `input_required` result + encrypted `requestState`; both flows go through `Phantom.Session.elicit/3` and behave identically from the tool's point of view. +- `tasks/get`, `tasks/update`, `tasks/cancel` - the Tasks extension under MCP + `2026-07-28`, when the router implements `c:Phantom.Router.get_task/2`. A tool + answers with a `Phantom.Tasks` and you store and run the task, such as with + Oban. See the [Tasks guide](guides/tasks.md). Phantom **does not yet support these methods**: @@ -872,8 +876,10 @@ the JSON-RPC body into HTTP headers so load balancers, proxies, and WAFs can route on the operation without parsing the body: - `Mcp-Method` — mirrors `method`. Required on every POST. -- `Mcp-Name` — mirrors `params.name` (for `tools/call`, `prompts/get`) - or `params.uri` (for `resources/read`). +- `Mcp-Name` — mirrors `params.name` (for `tools/call`, `prompts/get`), + `params.uri` (for `resources/read`), or `params.taskId` (for `tasks/get`, + `tasks/update`, `tasks/cancel`). Route on it when tasks live in one node's + memory. Header names are case-insensitive; values are case-sensitive (so `Mcp-Method: TOOLS/CALL` does *not* match a body method of `tools/call`). diff --git a/guides/tasks.md b/guides/tasks.md new file mode 100644 index 0000000..8e92952 --- /dev/null +++ b/guides/tasks.md @@ -0,0 +1,557 @@ +# Tasks + +The [Tasks extension](https://modelcontextprotocol.io/specification/2026-07-28/extensions/tasks) +(`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: + +| Phantom | You | +|---|---| +| Builds `CreateTaskResult` and `tasks/get` responses from `Phantom.Tasks` | Generate unguessable task IDs (`Phantom.Tasks.generate_id/0`) | +| Advertises the extension and rejects clients that did not declare it | Store a task *before* returning it, so an immediate `tasks/get` finds it | +| Validates the `Mcp-Name` routing header on `tasks/*` | Check access in `c:Phantom.Router.get_task/2` | +| Passes only outstanding `inputResponses` to `c:Phantom.Router.update_task/3` | Never reuse an input request key during a task's lifetime | +| Sends `notifications/tasks` to `subscriptions/listen` streams | Call `Phantom.Tracker.notify_task_updated/1` with the task or its ID after each change | +| | Delete tasks after their TTL | + +## 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: + +```elixir +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: + +```elixir +{:error, Phantom.Request.missing_task_capability(), session} +``` + +## Router callbacks + +Implementing `c:Phantom.Router.get_task/2` enables the extension: + +- `c: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. +- `c:Phantom.Router.update_task/3` — receive answers to the task's + `input_requests`. Without it, Phantom acknowledges `tasks/update` and drops + the answers. +- `c:Phantom.Router.cancel_task/2` — cancel the work. Without it, Phantom + acknowledges `tasks/cancel` and the task carries on, which the spec allows. +- `c: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](https://hexdocs.pm/oban). Inserting the task and its job in one +transaction means a task never exists without its job, or the reverse. + +### Migration + +```elixir +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 + +```elixir +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 +``` + +```elixir +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 + +```elixir +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. + +```elixir +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: + +```elixir +# In your application's start/2 +:telemetry.attach("mcp-task-failures", [:oban, :job, :exception], &MyApp.MCP.TaskTelemetry.handle_exception/4, nil) +``` + +```elixir +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: + +```elixir +config :my_app, Oban, + plugins: [{Oban.Plugins.Cron, crontab: [{"@hourly", MyApp.PruneMCPTasksWorker}]}] +``` + +```elixir +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: + +```elixir +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](https://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`; +`c:Phantom.Router.cancel_task/2` already records cancellations: + +```elixir +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`](https://oban.pro/docs/pro/Oban.Pro.Relay.html) +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: + +```elixir +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](https://oban.pro/docs/pro/Oban.Pro.Workflow.html) 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: + +```elixir +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. + +```elixir +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 `c: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: + +```elixir +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`. diff --git a/lib/phantom/elicit.ex b/lib/phantom/elicit.ex index 5bd22fa..e943a32 100644 --- a/lib/phantom/elicit.ex +++ b/lib/phantom/elicit.ex @@ -429,13 +429,18 @@ defmodule Phantom.Elicit do Render an `Elicit` struct as the MCP `2026-07-28` `inputRequests` map, which a tool or prompt returns in an `input_required` result. """ - def to_input_requests(%__MODULE__{} = elicit) do - %{ - "elicitation" => %{ - method: "elicitation/create", - params: Map.delete(to_json(elicit), :elicitationId) - } - } + def to_input_requests(%__MODULE__{} = elicit), + do: %{"elicitation" => to_input_request(elicit)} + + @doc """ + Render an `Elicit` struct as one embedded `elicitation/create` request, to + place under a key of your choosing in `inputRequests`, such as a task's + `:input_requests` (see `Phantom.Tasks`). + + %{"confirm" => Phantom.Elicit.to_input_request(elicit)} + """ + def to_input_request(%__MODULE__{} = elicit) do + %{method: "elicitation/create", params: Map.delete(to_json(elicit), :elicitationId)} end def to_json(%__MODULE__{mode: :url} = elicit) do diff --git a/lib/phantom/plug.ex b/lib/phantom/plug.ex index d29e554..b2c3f97 100644 --- a/lib/phantom/plug.ex +++ b/lib/phantom/plug.ex @@ -386,13 +386,32 @@ defmodule Phantom.Plug do true accept -> - accept - |> Enum.flat_map(&String.split(&1, ",")) - |> Enum.map(&(&1 |> String.split(";") |> hd() |> String.trim() |> String.downcase())) - |> Enum.any?(&(&1 in ~w[text/event-stream text/* */*])) + ranges = accept |> Enum.flat_map(&String.split(&1, ",")) |> Enum.map(&media_range/1) + + # The most specific matching range decides, and q=0 refuses it (RFC 9110 §12.5.1). + case Enum.find_value(~w[text/event-stream text/* */*], &List.keyfind(ranges, &1, 0)) do + {_type, q} -> q > 0 + nil -> false + end + end + end + + defp media_range(range) do + [type | params] = + range |> String.split(";") |> Enum.map(&(&1 |> String.trim() |> String.downcase())) + + {type, Enum.find_value(params, 1.0, &quality/1)} + end + + defp quality("q=" <> value) do + case Float.parse(value) do + {q, ""} -> q + _invalid -> 1.0 end end + defp quality(_param), do: nil + defp validate_protocol_request(%Plug.Conn{halted: true} = conn), do: conn defp validate_protocol_request(%Plug.Conn{method: "POST"} = conn) do @@ -596,7 +615,16 @@ defmodule Phantom.Plug do header_method = mcp_header(conn, "mcp-method") body_name = name_from_params(body_method, Map.get(params, "params")) header_name = mcp_header(conn, "mcp-name") |> decode_header_value() - needs_name? = body_method in ["tools/call", "prompts/get", "resources/read"] + + needs_name? = + body_method in [ + "tools/call", + "prompts/get", + "resources/read", + "tasks/get", + "tasks/update", + "tasks/cancel" + ] cond do is_nil(header_method) -> @@ -623,6 +651,7 @@ defmodule Phantom.Plug do end defp name_from_params("resources/read", params) when is_map(params), do: params["uri"] + defp name_from_params("tasks/" <> _, params) when is_map(params), do: params["taskId"] defp name_from_params(_method, params) when is_map(params), do: params["name"] defp name_from_params(_method, _params), do: nil @@ -859,6 +888,7 @@ defmodule Phantom.Plug do {state, exceptions} :ok -> + subscribe_request(state.session, request) run_dispatch(state, request, exceptions) end end @@ -1270,6 +1300,13 @@ defmodule Phantom.Plug do defp claim_in_flight(_session_id, _request), do: :ok + # `Phantom.Session.notify_progress/4` reaches the request through this topic. + defp subscribe_request(session, %Request{id: id, method: method}) + when method in @dedupable_methods and not is_nil(id), + do: Phantom.Tracker.subscribe_request(session.pubsub, session.id, id) + + defp subscribe_request(_session, _request), do: :ok + defp release_in_flight(session_id, %Request{id: id, method: method}) when method in @dedupable_methods and not is_nil(id), do: Phantom.Tracker.untrack_in_flight(session_id, id) diff --git a/lib/phantom/request.ex b/lib/phantom/request.ex index e390838..e5eac8c 100644 --- a/lib/phantom/request.ex +++ b/lib/phantom/request.ex @@ -45,6 +45,9 @@ defmodule Phantom.Request do completion/complete subscriptions/listen notifications/cancelled + tasks/get + tasks/update + tasks/cancel ] @modern_cacheable_methods ~w[ @@ -102,17 +105,10 @@ defmodule Phantom.Request do A request omitted a client capability required to process it. Capabilities are named by path, such as `"sampling"` or `"elicitation.url"`, - and reported to the client as a `ClientCapabilities` object. + and reported to the client as a `ClientCapabilities` object. Pass that + object directly when a name contains dots, such as an extension. """ - def missing_capability(capabilities) do - required = - capabilities - |> List.wrap() - |> Enum.reduce(%{}, fn capability, acc -> - path = String.split(capability, ".") - update_in(acc, Enum.map(path, &Access.key(&1, %{})), & &1) - end) - + def missing_capability(required) when is_map(required) do %{ code: @missing_capability, message: "Missing required client capability", @@ -120,6 +116,32 @@ defmodule Phantom.Request do } end + def missing_capability(capabilities) do + capabilities + |> List.wrap() + |> Enum.reduce(%{}, fn capability, acc -> + path = String.split(capability, ".") + update_in(acc, Enum.map(path, &Access.key(&1, %{})), & &1) + end) + |> missing_capability() + end + + @doc false + def task_not_found, + do: %{code: @invalid_params, message: "Failed to retrieve task: Task not found"} + + @doc false + def task_expired, + do: %{code: @invalid_params, message: "Failed to retrieve task: Task has expired"} + + @doc """ + The request did not declare the Tasks extension + (`io.modelcontextprotocol/tasks`), such as when a tool only runs as a task. + See `Phantom.Tasks`. + """ + def missing_task_capability, + do: missing_capability(%{extensions: %{Phantom.Tasks.extension() => %{}}}) + @doc "The requested MCP protocol version is unsupported." def unsupported_protocol(requested) do %{ @@ -577,6 +599,11 @@ defmodule Phantom.Request do %{jsonrpc: "2.0", method: "notifications/resources/updated", params: content} end + @doc "Task status notification" + def task_updated(content) do + %{jsonrpc: "2.0", method: "notifications/tasks", params: content} + end + @doc "Tools List updated notification" def tools_updated do %{jsonrpc: "2.0", method: "notifications/tools/list_changed"} diff --git a/lib/phantom/router.ex b/lib/phantom/router.ex index db606bd..c9d8844 100644 --- a/lib/phantom/router.ex +++ b/lib/phantom/router.ex @@ -41,6 +41,7 @@ defmodule Phantom.Router do alias Phantom.Resource alias Phantom.ResourceTemplate alias Phantom.Session + alias Phantom.Tasks alias Phantom.Tool alias Phantom.Tool.JSONSchema @@ -176,6 +177,60 @@ defmodule Phantom.Router do @callback authorize_resource_subscriptions([resolved_resource()], Session.t()) :: [resolved_resource() | String.t()] | nil + @doc """ + Fetch a task the client asked about with `tasks/get`, `tasks/update`, or + `tasks/cancel`. Implementing it enables the Tasks extension + (`io.modelcontextprotocol/tasks`); see `Phantom.Tasks`. + + This is where you check that the session may access the task. Return + `{:error, :not_found}` when it may not, so the client cannot tell the task + exists. Return `{:error, :expired}` for a task past its TTL, or a JSON-RPC + error map for anything else. + """ + @callback get_task(task_id :: String.t(), Session.t()) :: + {:ok, Phantom.Tasks.t()} | {:error, :not_found | :expired | map()} + + @doc """ + Receive the client's responses to a task's `input_requests` from + `tasks/update`, such as to resume the work. + + The task is the one `c:get_task/2` returned. Only responses for keys still + outstanding on it are passed; Phantom acknowledges the request without + calling this when none are, or when this callback is not implemented. + """ + @callback update_task(Phantom.Tasks.t(), input_responses :: map(), Session.t()) :: + :ok | {:error, map()} + + @doc """ + Cancel a task, such as by cancelling its job, when the client sends + `tasks/cancel`. + + The task is the one `c:get_task/2` returned. Cancellation is cooperative: + the task may still finish with another status. Without this callback, + Phantom acknowledges the request and the task carries on. + """ + @callback cancel_task(Phantom.Tasks.t(), Session.t()) :: :ok | {:error, map()} + + @doc """ + Authorize task status notifications for task IDs a client listens to with + `subscriptions/listen`. + + Return the allowed task IDs. Returning `nil` or an empty list, raising, or + returning an invalid value rejects them all, and IDs the client did not ask + for are ignored. + + Phantom invokes this callback when a client listens. Override it to check + many tasks with one query. Before each `notifications/tasks`, Phantom fetches + the task with `c:get_task/2`, so revoked access takes effect on the next + update and the client gets the stored task. + + The default implementation allows the tasks `c:get_task/2` returns. + """ + @callback authorize_task_subscriptions([task_id :: String.t()], Session.t()) :: + [String.t()] | nil + + @optional_callbacks get_task: 2, update_task: 3, cancel_task: 2 + @dialyzer {:nowarn_function, default_vsn: 1} defp default_vsn(nil) do Mix.Project.config()[:version] @@ -240,6 +295,10 @@ defmodule Phantom.Router do def connect(session, _auth_info), do: {:ok, session} def disconnect(session), do: {:ok, session} def authorize_resource_subscriptions(resources, _session), do: resources + + def authorize_task_subscriptions(task_ids, session), + do: Phantom.Router.authorize_gettable_tasks(__MODULE__, task_ids, session) + def terminate(session), do: {:error, nil} def instructions(_session), do: {:ok, @instructions} @@ -474,6 +533,7 @@ defmodule Phantom.Router do |> Phantom.Router.completion_capability(__MODULE__, session) |> Phantom.Router.logging_capability(__MODULE__, session) |> Phantom.Router.ui_capability(__MODULE__, session) + |> Phantom.Router.tasks_capability(__MODULE__) {:reply, %{ @@ -504,8 +564,11 @@ defmodule Phantom.Router do request, session ) do - case Session.listen(session, request.id, notifications) do - {:ok, session} -> {:noreply, session} + with :ok <- Phantom.Router.validate_task_subscriptions(notifications, session), + {:ok, session} <- Session.listen(session, request.id, notifications) do + {:noreply, session} + else + {:error, error} -> {:error, error, session} :error -> {:error, Request.invalid_params(), session} end end @@ -525,6 +588,10 @@ defmodule Phantom.Router do Phantom.Router.get_tool(__MODULE__, session, params, request) end + def dispatch_method("tasks/" <> _ = method, params, _request, session) do + Phantom.Router.task_request(__MODULE__, method, params, session) + end + def dispatch_method( "completion/complete", %{ @@ -611,13 +678,13 @@ defmodule Phantom.Router do def dispatch_method(_method, _params, %{id: request_id, response: %{} = response}, session) when is_binary(request_id) do - Phantom.Router.route_client_response(request_id, response) + Phantom.Router.route_client_response(session.pubsub, request_id, response) {:reply, nil, session} end def dispatch_method(_method, _params, %{id: request_id, error: %{} = error}, session) when is_binary(request_id) do - Phantom.Router.route_client_response(request_id, {:error, error}) + Phantom.Router.route_client_response(session.pubsub, request_id, {:error, error}) {:reply, nil, session} end @@ -628,6 +695,7 @@ defmodule Phantom.Router do @doc false defoverridable list_resources: 2, authorize_resource_subscriptions: 2, + authorize_task_subscriptions: 2, server_info: 1, disconnect: 1, connect: 2, @@ -1283,6 +1351,76 @@ defmodule Phantom.Router do [] end + @doc false + def authorize_task_subscriptions(_router, [], _session), do: [] + + def authorize_task_subscriptions(router, task_ids, session) do + case router.authorize_task_subscriptions(task_ids, session) do + allowed when is_list(allowed) -> + allowed = MapSet.new(allowed) + Enum.filter(task_ids, &MapSet.member?(allowed, &1)) + + _invalid -> + [] + end + rescue + exception -> + Logger.error( + "Task subscription authorization failed closed in #{inspect(router)}: " <> + Exception.message(exception) + ) + + [] + catch + kind, reason -> + Logger.error( + "Task subscription authorization failed closed in #{inspect(router)}: " <> + Exception.format_banner(kind, reason) + ) + + [] + end + + # Fails closed, as a stream must not crash on the user's callback. + @doc false + def get_task_for_notification(router, task_id, session) do + case router.get_task(task_id, session) do + {:ok, %Tasks{} = task} -> {:ok, task} + _ -> :error + end + rescue + exception -> + Logger.error( + "Task notification failed closed in #{inspect(router)}: " <> Exception.message(exception) + ) + + :error + catch + kind, reason -> + Logger.error( + "Task notification failed closed in #{inspect(router)}: " <> + Exception.format_banner(kind, reason) + ) + + :error + end + + @doc false + def authorize_gettable_tasks(router, task_ids, session) do + if tasks_enabled?(router), + do: Enum.filter(task_ids, &match?({:ok, _}, router.get_task(&1, session))), + else: [] + end + + @doc false + def validate_task_subscriptions(%{"taskIds" => [_ | _]}, session) do + if Session.tasks_supported?(session), + do: :ok, + else: {:error, Request.missing_task_capability()} + end + + def validate_task_subscriptions(_notifications, _session), do: :ok + defp resolve_resource(router, session, uri) when is_binary(uri) do with {:ok, %{scheme: scheme} = uri_struct} when is_binary(scheme) <- URI.new(uri), resource_router when not is_nil(resource_router) <- @@ -1396,6 +1534,16 @@ defmodule Phantom.Router do end end + @doc false + def tasks_capability(capabilities, router) do + if tasks_enabled?(router) do + extensions = Map.get(capabilities, :extensions, %{}) + Map.put(capabilities, :extensions, Map.put(extensions, Tasks.extension(), %{})) + else + capabilities + end + end + @doc false def completion_capability(capabilities, router, session) do resource_templates = Cache.list(session, router, :resource_templates) @@ -1412,8 +1560,9 @@ defmodule Phantom.Router do @doc false # Deliver the client's response to a server-initiated request (a result - # map, or `{:error, error}`) to the process waiting on it. - def route_client_response(request_id, response) do + # map, or `{:error, error}`) to the process waiting on it. PubSub reaches it + # at once on any node; a Tracker lookup can miss an unreplicated request. + def route_client_response(nil, request_id, response) do require Logger case await_request_meta(request_id) do @@ -1428,6 +1577,9 @@ defmodule Phantom.Router do end end + def route_client_response(pubsub, request_id, response), + do: Phantom.Tracker.cast_client_response(pubsub, request_id, response) + @doc false # Wait for a tracked request to become visible via Tracker replication. # In a distributed setup, the request may be tracked on one node but @@ -1819,6 +1971,14 @@ defmodule Phantom.Router do end end + defp finalize_result(:tool, {:reply, %Tasks{} = task, %Session{}}, _spec, _params, session) do + if Session.tasks_supported?(session) do + task |> Tasks.to_create_result() |> respond_to_caller() + else + respond_error_to_caller(Request.missing_task_capability()) + end + end + defp finalize_result(kind, {:reply, result, %Session{}}, spec, _params, session) do formatted = format_response(kind, result, session) @@ -1849,7 +2009,7 @@ defmodule Phantom.Router do elicitations |> Enum.with_index() |> Map.new(fn {elicit, index} -> - request = elicit |> Elicit.to_input_requests() |> Map.fetch!("elicitation") + request = Elicit.to_input_request(elicit) {"elicitation-#{index}", request} end) @@ -2072,6 +2232,96 @@ defmodule Phantom.Router do defp elicit_response_args(response) when is_map(response), do: response defp elicit_response_args(_response), do: %{} + @doc false + def task_request(router, method, params, session) do + with :ok <- validate_task_method(router, method, session), + :ok <- validate_tasks_supported(session), + {:ok, task_id} <- fetch_task_id(params), + {:ok, task} <- fetch_task(router, task_id, session) do + run_task_method(method, router, task, params, session) + else + {:error, error} -> {:error, error, session} + end + end + + @doc false + def tasks_enabled?(router), do: function_exported?(router, :get_task, 2) + + @task_methods ["tasks/get", "tasks/update", "tasks/cancel"] + + # The extension exists only under MCP 2026-07-28. + defp validate_task_method(router, method, session) do + if method in @task_methods and tasks_enabled?(router) and Session.stateless?(session), + do: :ok, + else: {:error, Request.not_found()} + end + + defp validate_tasks_supported(session) do + if Session.tasks_supported?(session), + do: :ok, + else: {:error, Request.missing_task_capability()} + end + + defp fetch_task_id(%{"taskId" => task_id}) when is_binary(task_id), do: {:ok, task_id} + + defp fetch_task_id(_params), + do: {:error, Request.invalid_params(%{taskId: "must be a string"})} + + defp fetch_task(router, task_id, session) do + case router.get_task(task_id, session) do + {:ok, %Tasks{} = task} -> {:ok, task} + {:error, :not_found} -> {:error, Request.task_not_found()} + {:error, :expired} -> {:error, Request.task_expired()} + {:error, error} when is_map(error) -> {:error, error} + other -> raise_callback_result!(router, "get_task/2", other) + end + end + + defp run_task_method("tasks/get", _router, task, _params, session), + do: {:reply, Tasks.to_json(task), session} + + # The spec recommends ignoring responses to keys the task is not waiting on. + defp run_task_method("tasks/update", router, task, %{"inputResponses" => responses}, session) + when is_map(responses) do + outstanding = + if task.status == :input_required, + do: Map.take(responses, Map.keys(task.input_requests)), + else: %{} + + if map_size(outstanding) > 0 and function_exported?(router, :update_task, 3), + do: + task_callback_result( + router, + "update_task/3", + router.update_task(task, outstanding, session), + session + ), + else: {:reply, %{}, session} + end + + defp run_task_method("tasks/update", _router, _task, _params, session), + do: {:error, Request.invalid_params(%{inputResponses: "must be an object"}), session} + + defp run_task_method("tasks/cancel", router, task, _params, session) do + if function_exported?(router, :cancel_task, 2), + do: + task_callback_result(router, "cancel_task/2", router.cancel_task(task, session), session), + else: {:reply, %{}, session} + end + + defp task_callback_result(_router, _callback, :ok, session), do: {:reply, %{}, session} + + defp task_callback_result(_router, _callback, {:error, error}, session) when is_map(error), + do: {:error, error, session} + + defp task_callback_result(router, callback, other, _session), + do: raise_callback_result!(router, callback, other) + + defp raise_callback_result!(router, callback, value) do + raise ArgumentError, + "#{inspect(router)}.#{callback} returned an unexpected value: #{inspect(value)}" + end + @doc false def get_prompt(router, session, name) do Enum.find(Cache.list(session, router, :prompts), &(&1.name == name)) diff --git a/lib/phantom/session.ex b/lib/phantom/session.ex index 6f52a91..315114c 100644 --- a/lib/phantom/session.ex +++ b/lib/phantom/session.ex @@ -275,6 +275,21 @@ defmodule Phantom.Session do def stateless?(_), do: false + @doc """ + Whether the client declared the Tasks extension on the current request, so + a tool may answer it with a `Phantom.Tasks`. The extension requires MCP + `2026-07-28`, and only the current request counts. + """ + def tasks_supported?(%__MODULE__{request: request} = session) do + stateless?(session) and + is_map( + get_in(Request.client_capabilities(request) || %{}, [ + "extensions", + Phantom.Tasks.extension() + ]) + ) + end + defp do_elicit(session, elicitation, opts) do timeout = Keyword.get(opts, :timeout, @elicitation_timeout) @@ -388,6 +403,16 @@ defmodule Phantom.Session do |> then(&Phantom.Router.authorize_resource_subscriptions(session.router, &1, session)) |> Enum.each(&Phantom.Tracker.subscribe_resource/1) + task_ids = + Phantom.Router.authorize_task_subscriptions( + session.router, + Map.get(filter, "taskIds", []), + session + ) + + Enum.each(task_ids, &Phantom.Tracker.subscribe_task/1) + filter = filter |> Map.delete("taskIds") |> maybe_put_subscription("taskIds", task_ids) + session = %{ session | close_after_complete: false, @@ -550,17 +575,74 @@ defmodule Phantom.Session do def ping(%__MODULE__{pid: pid}), do: ping(pid) def ping(pid) when is_pid(pid), do: GenServer.cast(pid, :ping) + @doc """ + A reference to the current request that another process, on any node, can + pass to `notify_progress/4`. It is a JSON-safe map, so it can travel in job + arguments, such as an Oban job's. + + Returns `nil` when the client did not ask for progress or the session has + no PubSub, and `notify_progress/4` does nothing with `nil`. + + args = %{report_id: id, progress: Phantom.Session.progress_ref(session)} + + # In the job, on any node: + Phantom.Session.notify_progress(args["progress"], 50, 100, "Halfway") + """ + @spec progress_ref(t()) :: map() | nil + def progress_ref(%__MODULE__{pubsub: pubsub, request: %Request{id: request_id}} = session) + when is_atom(pubsub) and not is_nil(pubsub) do + case progress_token(session) do + nil -> + nil + + token -> + %{ + "pubsub" => Atom.to_string(pubsub), + "session_id" => session.id, + "request_id" => request_id, + "progress_token" => token + } + end + end + + def progress_ref(%__MODULE__{}), do: nil + @doc """ Send a progress notification to the client the `progress` and `total` can be a integer or float, but must be ever-increasing. the `total` is optional. + Pass the session, or a reference from `progress_ref/1` to notify from + another process or node. + https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress """ - @spec notify_progress(t, number(), nil | number(), String.t() | nil) :: :ok + @spec notify_progress(t | map() | nil, number(), nil | number(), String.t() | nil) :: :ok def notify_progress(session, progress, total \\ nil, message \\ nil) + def notify_progress(nil, _progress, _total, _message), do: :ok + + def notify_progress( + %{ + "pubsub" => pubsub, + "session_id" => session_id, + "request_id" => request_id, + "progress_token" => token + }, + progress, + total, + message + ) do + pubsub + |> String.to_existing_atom() + |> Phantom.Tracker.cast_request( + session_id, + request_id, + {:progress, token, progress, total, message} + ) + end + def notify_progress(%__MODULE__{} = session, progress, total, message) do {pid, _request_id, token} = route(session) @@ -636,6 +718,8 @@ defmodule Phantom.Session do {request, ref} = Phantom.Elicit.prepare_request(state.session.id, tool_call_id, elicitation) + Phantom.Tracker.subscribe_client_response(state.session.pubsub, request.id) + caller = %{from: from, request_id: request.id} state = @@ -869,6 +953,33 @@ defmodule Phantom.Session do end end + def handle_cast({:task_updated, task_id}, state) do + notifications = + case Phantom.Router.get_task_for_notification(state.session.router, task_id, state.session) do + {:ok, task} -> + if task.status in [:completed, :failed, :cancelled], + do: Phantom.Tracker.unsubscribe_task(task.id) + + subscription_notifications( + state.session, + "taskIds", + task.id, + Request.task_updated(Phantom.Tasks.to_json(task)) + ) + + :error -> + [] + end + + if notifications != [] do + cancel_inactivity(state) + state = stream_notifications(state, notifications) + {:noreply, state |> set_activity() |> schedule_inactivity()} + else + {:noreply, state} + end + end + def handle_cast({:resource_updated, uri}, state), do: handle_cast({:resources_updated, [uri]}, state) @@ -945,14 +1056,16 @@ defmodule Phantom.Session do defp normalize_subscription_filter(filter) do resource_subscriptions = Map.get(filter, "resourceSubscriptions", []) + task_ids = Map.get(filter, "taskIds", []) - if is_list(resource_subscriptions) and Enum.all?(resource_subscriptions, &is_binary/1) do + if string_list?(resource_subscriptions) and string_list?(task_ids) do normalized = %{} |> maybe_put_subscription("toolsListChanged", filter["toolsListChanged"] == true) |> maybe_put_subscription("promptsListChanged", filter["promptsListChanged"] == true) |> maybe_put_subscription("resourcesListChanged", filter["resourcesListChanged"] == true) |> maybe_put_subscription("resourceSubscriptions", Enum.uniq(resource_subscriptions)) + |> maybe_put_subscription("taskIds", Enum.uniq(task_ids)) {:ok, normalized} else @@ -960,6 +1073,8 @@ defmodule Phantom.Session do end end + defp string_list?(list), do: is_list(list) and Enum.all?(list, &is_binary/1) + defp maybe_put_subscription(filter, _key, false), do: filter defp maybe_put_subscription(filter, _key, []), do: filter defp maybe_put_subscription(filter, key, value), do: Map.put(filter, key, value) @@ -1060,6 +1175,12 @@ defmodule Phantom.Session do {:noreply, Map.put(state, :workers, workers)} end + def handle_info({:phantom_client_response, request_id, response}, state) do + Phantom.Tracker.unsubscribe_client_response(state.session.pubsub, request_id) + Phantom.Tracker.untrack_request(request_id) + {:noreply, route_local_elicit_response({request_id, response}, state)} + end + def handle_info({:phantom_elicitation_response, ref, response}, state) do case pop_in(state, [:elicitation_callers, ref]) do {nil, state} -> @@ -1203,6 +1324,8 @@ defmodule Phantom.Session do error = Request.error(request_id, Request.duplicate_request()) state.stream_fun.(state, request_id, "message", error) else + Phantom.Tracker.subscribe_request(state.session.pubsub, state.session.id, request_id) + state |> claim_in_flight(request_id) |> do_dispatch_stdio_request(request) diff --git a/lib/phantom/tasks.ex b/lib/phantom/tasks.ex new file mode 100644 index 0000000..11b4516 --- /dev/null +++ b/lib/phantom/tasks.ex @@ -0,0 +1,214 @@ +defmodule Phantom.Tasks do + @moduledoc """ + A task from the MCP Tasks extension (`io.modelcontextprotocol/tasks`), + available under MCP `2026-07-28`. + + A tool handler may answer a call with a task instead of a result, and the + client polls for the outcome. Return one from the handler and Phantom + responds with a `CreateTaskResult`: + + def export_report(params, session) do + task = MyApp.MCPTasks.create!(session, params) + {:reply, Phantom.Tasks.new(id: task.id, status: :working, created_at: task.inserted_at), session} + end + + Phantom does not store tasks. You store them, run the work, and move them + through their statuses; Phantom asks your router for them when the client + polls. Check `Phantom.Session.tasks_supported?/1` before returning a task, + since a client that did not declare the extension cannot receive one. + + Return the task from the handler. `Phantom.Session.respond/2` does not + accept one, since the task is already the asynchronous answer. Only tools + answer with tasks. + + ## Fields + + - `:id` - (required) The task ID. It acts as a bearer token, so it must be + unguessable; see `generate_id/0`. + - `:status` - (required) One of `:working`, `:input_required`, + `:completed`, `:failed`, or `:cancelled`. Strings are accepted. + - `:created_at` - (required) A `DateTime`. + - `:last_updated_at` - A `DateTime`. Defaults to `:created_at`. + - `:status_message` - An optional message describing the current status. + - `:ttl_ms` - Milliseconds from creation that the task is kept, or `nil` + for unlimited. + - `:poll_interval_ms` - Suggested milliseconds between client polls. + - `:input_requests` - Required when `:input_required`. A map of keys to + `Phantom.Elicit` structs or embedded request maps. Keys must not be + reused over the task's lifetime. + - `:result` - Required when `:completed`. The final `CallToolResult`, such + as `Phantom.Tool.response(Phantom.Tool.text("Done"))`. A tool error is + still `:completed`. + - `:error` - Required when `:failed`. A JSON-RPC error with `:code` and + `:message`, such as `Phantom.Request.internal_error("Timed out")`. + """ + + @statuses ~w[working input_required completed failed cancelled]a + @status_strings Map.new(@statuses, &{Atom.to_string(&1), &1}) + + @enforce_keys [:id, :status, :created_at, :last_updated_at] + defstruct [ + :id, + :status, + :status_message, + :created_at, + :last_updated_at, + :ttl_ms, + :poll_interval_ms, + :result, + :error, + input_requests: %{} + ] + + @type status :: :working | :input_required | :completed | :failed | :cancelled + + @type t :: %__MODULE__{ + id: String.t(), + status: status(), + status_message: String.t() | nil, + created_at: DateTime.t(), + last_updated_at: DateTime.t(), + ttl_ms: non_neg_integer() | nil, + poll_interval_ms: pos_integer() | nil, + input_requests: %{String.t() => map()}, + result: map() | nil, + error: map() | nil + } + + @doc """ + Build a task. Raises `ArgumentError` when the fields do not describe a + valid task for its status. + """ + @spec new(map() | Keyword.t()) :: t() + def new(attrs) do + attrs = Map.new(attrs) + + %__MODULE__{ + id: fetch_id!(attrs), + status: fetch_status!(attrs), + created_at: fetch_datetime!(attrs, :created_at), + last_updated_at: fetch_datetime!(attrs, :last_updated_at, attrs[:created_at]), + status_message: attrs[:status_message], + ttl_ms: fetch_duration!(attrs, :ttl_ms, 0), + poll_interval_ms: fetch_duration!(attrs, :poll_interval_ms, 1), + input_requests: Map.new(attrs[:input_requests] || %{}, &to_input_request/1), + result: attrs[:result], + error: attrs[:error] + } + |> validate_status_fields!() + end + + @doc false + def extension, do: "io.modelcontextprotocol/tasks" + + @doc "Generate an unguessable, URL-safe task ID with 256 bits of entropy." + @spec generate_id() :: String.t() + def generate_id do + 32 |> :crypto.strong_rand_bytes() |> Base.url_encode64(padding: false) + end + + # Tasks changed with struct update syntax are validated as they leave Phantom. + @doc false + def to_json(%__MODULE__{} = task) do + task = revalidate(task) + task |> base_json() |> put_status_fields(task) + end + + # The client fetches the result, error, and input requests with `tasks/get`. + @doc false + def to_create_result(%__MODULE__{} = task), + do: task |> revalidate() |> base_json() |> Map.put(:resultType, "task") + + defp revalidate(task), do: task |> Map.from_struct() |> new() + + defp base_json(task) do + %{ + taskId: task.id, + status: Atom.to_string(task.status), + createdAt: DateTime.to_iso8601(task.created_at), + lastUpdatedAt: DateTime.to_iso8601(task.last_updated_at), + ttlMs: task.ttl_ms + } + |> maybe_put(:statusMessage, task.status_message) + |> maybe_put(:pollIntervalMs, task.poll_interval_ms) + end + + defp put_status_fields(json, %{status: :input_required} = task), + do: Map.put(json, :inputRequests, task.input_requests) + + defp put_status_fields(json, %{status: :completed} = task), + do: Map.put(json, :result, task.result) + + defp put_status_fields(json, %{status: :failed} = task), + do: Map.put(json, :error, task.error) + + defp put_status_fields(json, _task), do: json + + defp fetch_id!(%{id: id}) when is_binary(id) and byte_size(id) > 0, do: id + + defp fetch_id!(attrs), + do: raise(ArgumentError, ":id must be a non-empty string, got: #{inspect(attrs[:id])}") + + defp fetch_status!(%{status: status}) when status in @statuses, do: status + + defp fetch_status!(%{status: status}) when is_map_key(@status_strings, status), + do: Map.fetch!(@status_strings, status) + + defp fetch_status!(attrs) do + raise ArgumentError, + "status must be one of #{inspect(@statuses)}, got: #{inspect(attrs[:status])}" + end + + defp fetch_datetime!(attrs, key, default \\ nil) do + case attrs[key] || default do + %DateTime{} = datetime -> datetime + other -> raise ArgumentError, "#{inspect(key)} must be a DateTime, got: #{inspect(other)}" + end + end + + defp fetch_duration!(attrs, key, min) do + case attrs[key] do + nil -> + nil + + ms when is_integer(ms) and ms >= min -> + ms + + other -> + raise ArgumentError, + "#{inspect(key)} must be nil or an integer >= #{min}, got: #{inspect(other)}" + end + end + + # Keys are strings, as the client's `inputResponses` keys are. + defp to_input_request({key, %Phantom.Elicit{} = elicit}), + do: {to_string(key), Phantom.Elicit.to_input_request(elicit)} + + defp to_input_request({key, request}), do: {to_string(key), request} + + defp validate_status_fields!(%{status: :completed, result: result}) + when not is_map(result), + do: raise(ArgumentError, "a completed task requires a :result map") + + defp validate_status_fields!(%{status: :failed, error: %{code: code, message: message}} = task) + when is_integer(code) and is_binary(message), + do: task + + defp validate_status_fields!( + %{status: :failed, error: %{"code" => code, "message" => message}} = task + ) + when is_integer(code) and is_binary(message), + do: task + + defp validate_status_fields!(%{status: :failed}), + do: raise(ArgumentError, "a failed task requires an :error with a :code and :message") + + defp validate_status_fields!(%{status: :input_required, input_requests: requests}) + when map_size(requests) == 0, + do: raise(ArgumentError, "an input_required task requires :input_requests") + + defp validate_status_fields!(task), do: task + + defp maybe_put(map, _key, nil), do: map + defp maybe_put(map, key, value), do: Map.put(map, key, value) +end diff --git a/lib/phantom/test.ex b/lib/phantom/test.ex index d62e4c3..a7640a4 100644 --- a/lib/phantom/test.ex +++ b/lib/phantom/test.ex @@ -203,6 +203,9 @@ defmodule Phantom.Test do Options: * `:timeout` - ms to wait for an async response (default: 1000) * `:progress_token` - sets `params._meta.progressToken` on the request + * `:tasks` - declares the Tasks extension on the request, so the tool may + answer with a `Phantom.Tasks`. Requires a session built with + `:protocol_version`. """ @spec call_tool(Session.t(), atom() | String.t(), map() | keyword(), keyword()) :: any() def call_tool(session, name, args \\ %{}, opts \\ []) do @@ -288,6 +291,33 @@ defmodule Phantom.Test do dispatch_blocking(session, "resources/list", %{"cursor" => cursor}, opts) end + @doc """ + Dispatch `tasks/get` for a task. Declares the Tasks extension on the + request, so the session must be built with `:protocol_version`. + + Returns the task as the client sees it, or `{:jsonrpc_error, error_map}`. + """ + @spec get_task(Session.t(), String.t(), keyword()) :: any() + def get_task(session, task_id, opts \\ []) do + dispatch_blocking(session, "tasks/get", %{"taskId" => task_id}, [tasks: true] ++ opts) + end + + @doc """ + Dispatch `tasks/update` with responses to a task's input requests. See + `get_task/3`. + """ + @spec update_task(Session.t(), String.t(), map() | keyword(), keyword()) :: any() + def update_task(session, task_id, input_responses, opts \\ []) do + params = %{"taskId" => task_id, "inputResponses" => stringify_keys(input_responses)} + dispatch_blocking(session, "tasks/update", params, [tasks: true] ++ opts) + end + + @doc "Dispatch `tasks/cancel` for a task. See `get_task/3`." + @spec cancel_task(Session.t(), String.t(), keyword()) :: any() + def cancel_task(session, task_id, opts \\ []) do + dispatch_blocking(session, "tasks/cancel", %{"taskId" => task_id}, [tasks: true] ++ opts) + end + @doc false def invoke_elicit_responder(%Phantom.Elicit{mode: mode} = elicitation) do case Process.get({__MODULE__, :elicit_responder, mode}) do @@ -347,7 +377,13 @@ defmodule Phantom.Test do defp dispatch_once(session, method, params, opts) do timeout = Keyword.get(opts, :timeout, @default_timeout) - meta = Map.merge(session_meta(session), progress_meta(opts)) + + meta = + session + |> session_meta() + |> Map.merge(progress_meta(opts)) + |> put_tasks_capability(opts[:tasks]) + request = %{build_request(method, params: Map.put(params, "_meta", meta)) | meta: meta} session = %{ @@ -371,6 +407,25 @@ defmodule Phantom.Test do defp session_meta(%Session{request: %Request{meta: meta}}) when is_map(meta), do: meta defp session_meta(_session), do: %{} + @protocol_version_key "io.modelcontextprotocol/protocolVersion" + @client_capabilities_key "io.modelcontextprotocol/clientCapabilities" + + defp put_tasks_capability(meta, true) do + if not is_map_key(meta, @protocol_version_key) do + raise ArgumentError, + "the Tasks extension requires a session built with :protocol_version, such as \"2026-07-28\"" + end + + capabilities = Map.get(meta, @client_capabilities_key, %{}) + + extensions = + capabilities |> Map.get("extensions", %{}) |> Map.put(Phantom.Tasks.extension(), %{}) + + Map.put(meta, @client_capabilities_key, Map.put(capabilities, "extensions", extensions)) + end + + defp put_tasks_capability(meta, _tasks), do: meta + defp progress_meta(opts) do case Keyword.get(opts, :progress_token) do nil -> %{} @@ -607,6 +662,32 @@ defmodule Phantom.Test do end end + @doc """ + Assert the result is a task, from a tool that answered with one or from + `get_task/3`. + + Options: `:id` matches the task ID, `:status` the status (such as + `:completed`). + """ + def assert_task(result, opts \\ []) do + case result do + %{taskId: id, status: status} -> + if expected = opts[:id] do + assert id == expected, "expected task id #{inspect(expected)}, got: #{inspect(id)}" + end + + if expected = opts[:status] do + assert status == to_string(expected), + "expected task status #{inspect(to_string(expected))}, got: #{inspect(status)}" + end + + result + + other -> + flunk("expected a task, got: #{inspect(other)}") + end + end + @doc "Assert the tool returned image content." def assert_tool_image(result, opts \\ []) do assert_content_type(result, :image, opts) diff --git a/lib/phantom/tracker.ex b/lib/phantom/tracker.ex index fff471b..7ef0e2e 100644 --- a/lib/phantom/tracker.ex +++ b/lib/phantom/tracker.ex @@ -27,6 +27,7 @@ defmodule Phantom.Tracker do @sessions "phantom:sessions" @requests "phantom:requests" @resources "phantom:resources" + @tasks "phantom:tasks" def available?, do: @available def resource_subscription_topic, do: @resources @@ -276,6 +277,17 @@ defmodule Phantom.Tracker do def list_resource_listeners, do: [] end + @doc "Return a list of all listening for task status notifications" + if @available do + def list_task_listeners do + Phoenix.Tracker.list(__MODULE__, @tasks) + rescue + _ -> [] + end + else + def list_task_listeners, do: [] + end + @doc "Fetch the PID of the open request by ID" def get_request(%Phantom.Request{id: request_id}), do: get_request(request_id) @@ -338,6 +350,13 @@ defmodule Phantom.Tracker do "phantom:listen:" <> Base.url_encode64(key, padding: false) end + defp request_topic(session_id, request_id) do + key = :crypto.hash(:sha256, :erlang.term_to_binary({session_id, request_id})) + "phantom:request:" <> Base.url_encode64(key, padding: false) + end + + defp client_response_topic(request_id), do: "phantom:client_response:#{request_id}" + if @available do @doc false def subscribe_session(nil, _session_id), do: :ok @@ -358,17 +377,31 @@ defmodule Phantom.Tracker do # stream answer. reply_to = Process.alias([:reply]) call = {:"$gen_call", {reply_to, reply_to}, message} + deadline = System.monotonic_time(:millisecond) + timeout + await_session_reply(pubsub, session_id, call, reply_to, deadline, 50) + end + + # A client may send a request before its GET stream subscribes, and a + # broadcast reaches only current subscribers, so the call repeats. + defp await_session_reply(pubsub, session_id, call, reply_to, deadline, interval) do Phoenix.PubSub.broadcast_from(pubsub, self(), session_topic(session_id), call) + remaining = deadline - System.monotonic_time(:millisecond) receive do {^reply_to, reply} -> reply after - timeout -> - Process.unalias(reply_to) - :error + max(min(interval, remaining), 0) -> + if remaining > interval, + do: await_session_reply(pubsub, session_id, call, reply_to, deadline, interval * 2), + else: give_up(reply_to) end end + defp give_up(reply_to) do + Process.unalias(reply_to) + :error + end + @doc false def subscribe_listen(nil, _principal, _request_id), do: :ok @@ -386,6 +419,52 @@ defmodule Phantom.Tracker do ) end + @doc false + def subscribe_client_response(nil, _request_id), do: :ok + + def subscribe_client_response(pubsub, request_id) do + Phoenix.PubSub.subscribe(pubsub, client_response_topic(request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def unsubscribe_client_response(nil, _request_id), do: :ok + + def unsubscribe_client_response(pubsub, request_id) do + Phoenix.PubSub.unsubscribe(pubsub, client_response_topic(request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def cast_client_response(pubsub, request_id, response) do + Phoenix.PubSub.broadcast( + pubsub, + client_response_topic(request_id), + {:phantom_client_response, request_id, response} + ) + end + + @doc false + def subscribe_request(nil, _session_id, _request_id), do: :ok + + # Progress is best-effort, so a missing PubSub must not fail the request. + def subscribe_request(pubsub, session_id, request_id) do + Phoenix.PubSub.subscribe(pubsub, request_topic(session_id, request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def cast_request(pubsub, session_id, request_id, message) do + Phoenix.PubSub.broadcast( + pubsub, + request_topic(session_id, request_id), + {:"$gen_cast", message} + ) + end + @doc false def cast_session(nil, _session_id, _message), do: :ok @@ -397,6 +476,11 @@ defmodule Phantom.Tracker do def cast_session(_pubsub, _session_id, _message), do: :ok def subscribe_listen(_pubsub, _principal, _request_id), do: :ok def cast_listen(_pubsub, _principal, _request_id, _message), do: :ok + def subscribe_request(_pubsub, _session_id, _request_id), do: :ok + def subscribe_client_response(_pubsub, _request_id), do: :ok + def unsubscribe_client_response(_pubsub, _request_id), do: :ok + def cast_client_response(_pubsub, _request_id, _response), do: :ok + def cast_request(_pubsub, _session_id, _request_id, _message), do: :ok end @doc "Untrack the processe for everything" @@ -470,6 +554,28 @@ defmodule Phantom.Tracker do def subscribe_resource(_uri), do: {:error, :not_available} end + @doc false + if @available do + def subscribe_task(task_id) do + Phoenix.Tracker.track(__MODULE__, self(), @tasks, task_id, %{pid: self()}) + rescue + _ -> {:error, :tracker_not_in_supervision_tree} + end + else + def subscribe_task(_task_id), do: {:error, :not_available} + end + + @doc false + if @available do + def unsubscribe_task(task_id) do + Phoenix.Tracker.untrack(__MODULE__, self(), @tasks, task_id) + rescue + _ -> :ok + end + else + def unsubscribe_task(_task_id), do: :ok + end + @doc "Unsubscribe the process to resource notifications from the PubSub on topic #{inspect(@resources)}" if @available do def unsubscribe_resource(uri) do @@ -539,6 +645,36 @@ defmodule Phantom.Tracker do def notify_resources_updated(_uris), do: {:ok, 0} end + @doc """ + Notify clients listening to a task that it changed. Call it after each + change you store, with the task or its ID. + + Each listening stream fetches the task with `c:Phantom.Router.get_task/2` + and sends what it returns as `notifications/tasks`, so clients get the + stored task even when notifications arrive out of order. Returns the number + of listening streams. + """ + @spec notify_task_updated(Phantom.Tasks.t() | String.t()) :: + {:ok, non_neg_integer()} | {:error, :tracker_not_in_supervision_tree} + def notify_task_updated(%Phantom.Tasks{id: task_id}), do: notify_task_updated(task_id) + + if @available do + def notify_task_updated(task_id) when is_binary(task_id) do + pids = + __MODULE__ + |> Phoenix.Tracker.get_by_key(@tasks, task_id) + |> Enum.map(fn {_key, %{pid: pid}} -> pid end) + |> Enum.uniq() + + Enum.each(pids, &GenServer.cast(&1, {:task_updated, task_id})) + {:ok, length(pids)} + rescue + _ -> {:error, :tracker_not_in_supervision_tree} + end + else + def notify_task_updated(task_id) when is_binary(task_id), do: {:ok, 0} + end + @doc "Convenience wrapper for notifying sessions about one updated resource" @spec notify_resource_updated(String.t()) :: {:ok, non_neg_integer()} | {:error, :tracker_not_in_supervision_tree} diff --git a/mix.exs b/mix.exs index 5415129..ef5a0d4 100644 --- a/mix.exs +++ b/mix.exs @@ -109,7 +109,7 @@ defmodule Phantom.MixProject do defp docs do [ main: "Phantom", - extras: ["guides/mcp_apps.md", "CHANGELOG.md"], + extras: ["guides/mcp_apps.md", "guides/tasks.md", "CHANGELOG.md"], assets: %{"assets" => "assets"}, before_closing_body_tag: %{html: @mermaidjs} ] diff --git a/test/conformance/conformance_test.exs b/test/conformance/conformance_test.exs index b8d7063..562772e 100644 --- a/test/conformance/conformance_test.exs +++ b/test/conformance/conformance_test.exs @@ -11,7 +11,17 @@ defmodule Phantom.ConformanceTest do @unscored_scenarios [ {"server-session-lifecycle", "2025-11-25"}, {"json-schema-2020-12", "2025-11-25"}, - {"json-schema-2020-12", "2026-07-28"} + {"json-schema-2020-12", "2026-07-28"}, + {"tasks-lifecycle", "2026-07-28"}, + {"tasks-capability-negotiation", "2026-07-28"}, + {"tasks-wire-fields", "2026-07-28"}, + {"tasks-request-state-removal", "2026-07-28"}, + {"tasks-mrtr-input", "2026-07-28"}, + {"tasks-request-headers", "2026-07-28"}, + {"tasks-dispatch-and-envelope", "2026-07-28"}, + {"tasks-status-notifications", "2026-07-28"}, + {"tasks-required-task-error", "2026-07-28"}, + {"tasks-mrtr-composition", "2026-07-28"} ] setup_all do diff --git a/test/conformance/router.ex b/test/conformance/router.ex index 82ffe0d..068afe5 100644 --- a/test/conformance/router.ex +++ b/test/conformance/router.ex @@ -590,6 +590,202 @@ defmodule Phantom.Conformance.MCP.Router do defp roots_request, do: %{method: "roots/list", params: %{}} + ## Tasks + + # A global Agent lets every node in the cluster answer `tasks/*`. + @tasks {:global, Phantom.Conformance.Tasks} + + tool :greet, description: "Returns a greeting without a task" do + field :name, :string, required: true + end + + tool :slow_compute, description: "Sleeps for the given seconds, then returns a result" do + field :seconds, :integer, required: true + field :label, :string + end + + tool :failing_job, description: "Requires a task; ends with a tool error after about a second" + tool :protocol_error_job, description: "Runs as a task that fails with a protocol error" + + tool :confirm_delete, description: "Asks to confirm, then deletes a file" do + field :filename, :string, required: true + end + + tool :multi_input, description: "Asks for two inputs at once, then completes" + tool :test_tool_with_task, description: "Asks for a name, then greets it from a task" + + def greet(%{"name" => name}, session), do: {:reply, Tool.text("Hello, #{name}!"), session} + + def slow_compute(%{"seconds" => seconds} = params, session) do + label = params["label"] || "compute" + + if seconds > 0 and Session.tasks_supported?(session) do + task = + start_task(fn -> + Process.sleep(:timer.seconds(seconds)) + {:completed, Tool.response(Tool.text("Computed #{label}"))} + end) + + {:reply, task, session} + else + Process.sleep(:timer.seconds(seconds)) + {:reply, Tool.text("Computed #{label}"), session} + end + end + + def failing_job(_params, session) do + if Session.tasks_supported?(session) do + task = + start_task(fn -> + Process.sleep(1_000) + {:completed, Tool.error("failing_job failed")} + end) + + {:reply, task, session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def protocol_error_job(_params, session) do + if Session.tasks_supported?(session) do + task = + start_task(fn -> + {:failed, Phantom.Request.internal_error("protocol_error_job failed")} + end) + + {:reply, task, session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def confirm_delete(%{"filename" => filename}, session) do + if Session.tasks_supported?(session) do + requests = %{"confirm" => elicit_request("Delete #{filename}?", "confirm", "boolean")} + {:reply, await_input_task(requests, "Deleted #{filename}"), session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def multi_input(_params, session) do + if Session.tasks_supported?(session) do + requests = %{ + "name" => elicit_request("What is your name?", "name"), + "confirm" => elicit_request("Continue?", "confirm", "boolean") + } + + {:reply, await_input_task(requests, "Received all input"), session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def test_tool_with_task(_params, session) do + case {session.state, input_responses(session)} do + {%{step: :named}, %{"user_name" => response}} -> + name = input_text(response, "name") + task = start_task(fn -> {:completed, Tool.response(Tool.text("Hello, #{name}!"))} end) + {:reply, task, session} + + _ -> + {:reply, + Tool.input_required( + input_requests: %{"user_name" => elicit_request("What is your name?", "name")}, + state: %{step: :named} + ), session} + end + end + + def get_task(task_id, _session) do + case Agent.get(tasks(), &Map.get(&1, task_id)) do + nil -> {:error, :not_found} + %{task: task} -> {:ok, task} + end + end + + def update_task(task, responses, _session) do + update_stored_task(task.id, fn %{task: task, on_input: result} = entry -> + remaining = Map.drop(task.input_requests, Map.keys(responses)) + + task = + if map_size(remaining) == 0, + do: %{task | status: :completed, input_requests: %{}, result: Tool.response(result)}, + else: %{task | input_requests: remaining} + + %{entry | task: task} + end) + end + + def cancel_task(task, _session) do + update_stored_task(task.id, fn %{task: task} = entry -> + %{entry | task: %{task | status: :cancelled, status_message: "Cancelled by the client"}} + end) + end + + defp start_task(work) do + task = new_task(:working, %{}) + Agent.update(tasks(), &Map.put(&1, task.id, %{task: task, on_input: nil})) + + spawn(fn -> + outcome = work.() + + update_stored_task(task.id, fn + %{task: %{status: :working} = task} = entry -> %{entry | task: finish(task, outcome)} + entry -> entry + end) + end) + + task + end + + defp await_input_task(requests, result) do + task = new_task(:input_required, requests) + Agent.update(tasks(), &Map.put(&1, task.id, %{task: task, on_input: Tool.text(result)})) + task + end + + defp new_task(status, input_requests) do + Phantom.Tasks.new( + id: Phantom.Tasks.generate_id(), + status: status, + created_at: DateTime.utc_now(), + ttl_ms: :timer.minutes(5), + poll_interval_ms: 200, + input_requests: input_requests + ) + end + + defp finish(task, {:completed, result}), do: %{task | status: :completed, result: result} + defp finish(task, {:failed, error}), do: %{task | status: :failed, error: error} + + # A terminal task never changes, so a late update or cancel is a no-op. + defp update_stored_task(task_id, fun) do + updated = + Agent.get_and_update(tasks(), fn tasks -> + case Map.fetch(tasks, task_id) do + {:ok, %{task: %{status: status}} = entry} when status in [:working, :input_required] -> + entry = fun.(entry) + entry = put_in(entry.task.last_updated_at, DateTime.utc_now()) + {entry.task, Map.put(tasks, task_id, entry)} + + _ -> + {nil, tasks} + end + end) + + if updated, do: Phantom.Tracker.notify_task_updated(updated) + :ok + end + + defp tasks do + case Agent.start(fn -> %{} end, name: @tasks) do + {:ok, pid} -> pid + {:error, {:already_started, pid}} -> pid + end + end + ## Resources resource "test://static-text", :static_text, diff --git a/test/phantom/client_response_test.exs b/test/phantom/client_response_test.exs new file mode 100644 index 0000000..6f181c6 --- /dev/null +++ b/test/phantom/client_response_test.exs @@ -0,0 +1,27 @@ +defmodule Phantom.ClientResponseTest do + use ExUnit.Case, async: true + + @pubsub Phantom.ClientResponseTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + # No `Phantom.Tracker` runs here, so delivery cannot depend on its replication. + test "routes a client response to the waiting process through PubSub" do + Phantom.Tracker.subscribe_client_response(@pubsub, "elicit-1") + + Phantom.Router.route_client_response(@pubsub, "elicit-1", %{"action" => "accept"}) + assert_receive {:phantom_client_response, "elicit-1", %{"action" => "accept"}} + + Phantom.Router.route_client_response(@pubsub, "elicit-1", {:error, %{"code" => -1}}) + assert_receive {:phantom_client_response, "elicit-1", {:error, %{"code" => -1}}} + end + + test "does not deliver a response for another request" do + Phantom.Tracker.subscribe_client_response(@pubsub, "elicit-1") + Phantom.Router.route_client_response(@pubsub, "elicit-2", %{"action" => "accept"}) + refute_receive {:phantom_client_response, _, _} + end +end diff --git a/test/phantom/distributed_test.exs b/test/phantom/distributed_test.exs index eb95412..97d03e4 100644 --- a/test/phantom/distributed_test.exs +++ b/test/phantom/distributed_test.exs @@ -447,6 +447,96 @@ defmodule Phantom.DistributedTest do end end + describe "cross-node task notifications" do + test "a task update on node 2 reaches a subscriptions/listen stream on node 1" do + task_id = "task-cross-node" + + meta = %{ + "io.modelcontextprotocol/protocolVersion" => "2026-07-28", + "io.modelcontextprotocol/clientCapabilities" => %{ + "extensions" => %{"io.modelcontextprotocol/tasks" => %{}} + } + } + + stream_resp = + post_mcp( + @node1_port, + %{ + jsonrpc: "2.0", + id: "listen:tasks", + method: "subscriptions/listen", + params: %{"notifications" => %{"taskIds" => [task_id]}, "_meta" => meta} + }, + headers: [ + {"mcp-protocol-version", "2026-07-28"}, + {"mcp-method", "subscriptions/listen"} + ] + ) + + await_task_tracked(@node2, task_id) + + assert {:ok, 1} = :rpc.call(@node2, Phantom.Tracker, :notify_task_updated, [task_id]) + + notification = + poll_for_sse_event(stream_resp, 10_000, &(&1["method"] == "notifications/tasks")) + + assert notification, "Expected notifications/tasks on the node 1 listen stream" + assert notification["params"]["taskId"] == task_id + assert notification["params"]["status"] == "working" + end + end + + describe "cross-node progress" do + test "a progress reference reaches the request's stream from another node" do + resp = + post_mcp( + @node1_port, + %{ + jsonrpc: "2.0", + id: "progress-1", + method: "tools/call", + params: %{ + "name" => "remote_progress_tool", + "arguments" => %{}, + "_meta" => Map.put(@modern_meta, "progressToken", "remote-token") + } + }, + headers: [ + {"mcp-protocol-version", "2026-07-28"}, + {"mcp-method", "tools/call"}, + {"mcp-name", "remote_progress_tool"} + ] + ) + + notification = + poll_for_sse_event(resp, 10_000, &(&1["method"] == "notifications/progress")) + + assert notification, "Expected notifications/progress from another node" + + assert %{"progressToken" => "remote-token", "progress" => 50, "total" => 100} = + notification["params"] + + assert notification["params"]["message"] =~ "node" + refute notification["params"]["message"] =~ "#{@node1}" + end + end + + defp await_task_tracked(node, task_id, attempts \\ 100) + + defp await_task_tracked(node, task_id, attempts) when attempts > 0 do + listeners = :rpc.call(node, Phantom.Tracker, :list_task_listeners, []) + + if Enum.any?(listeners, &match?({^task_id, _}, &1)) do + :ok + else + Process.sleep(50) + await_task_tracked(node, task_id, attempts - 1) + end + end + + defp await_task_tracked(node, task_id, 0), + do: flunk("Task #{task_id} not visible on #{node} within timeout") + describe "stateless core (2026-07-28) — no Tracker, no sticky session" do # Node A returns `inputRequired` with an encrypted `requestState`. # Node B — with no prior knowledge of the original call — decodes the diff --git a/test/phantom/elicit_test.exs b/test/phantom/elicit_test.exs index 8f6b378..1fa638b 100644 --- a/test/phantom/elicit_test.exs +++ b/test/phantom/elicit_test.exs @@ -235,6 +235,32 @@ defmodule Phantom.ElicitTest do end end + describe "to_input_request/1" do + test "builds one embedded elicitation request" do + elicit = + Elicit.form(%{ + message: "Continue?", + requested_schema: [%{type: :boolean, name: "confirm", required: true}] + }) + + assert %{ + method: "elicitation/create", + params: %{mode: "form", message: "Continue?", requestedSchema: %{type: "object"}} + } = Elicit.to_input_request(elicit) + end + + test "omits the legacy elicitation id from a URL request" do + elicit = + Elicit.url(%{ + message: "Authenticate", + url: "https://example.com", + elicitation_id: "legacy-id" + }) + + refute Map.has_key?(Elicit.to_input_request(elicit).params, :elicitationId) + end + end + describe "to_input_requests/1" do test "builds the modern embedded elicitation request map" do elicit = diff --git a/test/phantom/plug_test.exs b/test/phantom/plug_test.exs index 426ef49..b0f9bb2 100644 --- a/test/phantom/plug_test.exs +++ b/test/phantom/plug_test.exs @@ -581,6 +581,34 @@ defmodule Phantom.PlugTest do assert error["error"]["code"] == -32600 end + test "GET request that refuses text/event-stream with q=0 returns 406" do + for accept <- [ + "text/event-stream;q=0", + "application/json, text/event-stream; q=0.0", + "text/*;q=0", + "*/*;q=0.5, text/event-stream;q=0" + ] do + :get + |> conn("/mcp") + |> put_req_header("accept", accept) + |> call() + + assert_receive {:conn, conn} + assert conn.status == 406, "expected 406 for Accept: #{accept}" + end + end + + test "GET request accepting text/event-stream with a nonzero q opens a stream" do + for accept <- ["text/event-stream;q=0.5", "application/json, */*;q=0.1"] do + :get + |> conn("/mcp") + |> put_req_header("accept", accept) + |> call() + + assert_sse_connected() + end + end + test "GET request accepting any media type opens a stream" do :get |> conn("/mcp") @@ -1615,6 +1643,55 @@ defmodule Phantom.PlugTest do refute_receive {:response, nil, "message", %{method: "notifications/message"}} end + test "tasks methods require Mcp-Name to carry the taskId" do + for method <- ~w[tasks/get tasks/update tasks/cancel] do + post_stateless( + %{jsonrpc: "2.0", id: 20, method: method, params: %{"taskId" => "task-1"}}, + [{"mcp-method", method}] + ) + + assert_receive {:conn, conn} + assert conn.status == 400 + error = JSON.decode!(conn.resp_body) + assert error["error"]["code"] == -32020 + assert error["error"]["message"] =~ "Mcp-Name" + end + end + + test "tasks methods reject an Mcp-Name that is not the taskId" do + post_stateless( + %{jsonrpc: "2.0", id: 21, method: "tasks/get", params: %{"taskId" => "task-1"}}, + [{"mcp-method", "tasks/get"}, {"mcp-name", "task-2"}] + ) + + assert_receive {:conn, conn} + assert conn.status == 400 + assert JSON.decode!(conn.resp_body)["error"]["code"] == -32020 + end + + test "tasks methods with a matching Mcp-Name reach the router" do + meta = %{ + "io.modelcontextprotocol/clientCapabilities" => %{ + "extensions" => %{"io.modelcontextprotocol/tasks" => %{}} + } + } + + post_stateless( + %{ + jsonrpc: "2.0", + id: 22, + method: "tasks/get", + params: %{"taskId" => "task-1", "_meta" => meta} + }, + [{"mcp-method", "tasks/get"}, {"mcp-name", "task-1"}] + ) + + assert_connected(_conn) + + assert_receive {:response, 22, "message", + %{result: %{taskId: "task-1", status: "working", resultType: "complete"}}} + end + test "missing Mcp-Method rejects with 400 + -32020 HeaderMismatch" do post_stateless( %{ diff --git a/test/phantom/progress_ref_test.exs b/test/phantom/progress_ref_test.exs new file mode 100644 index 0000000..d251cfd --- /dev/null +++ b/test/phantom/progress_ref_test.exs @@ -0,0 +1,69 @@ +defmodule Phantom.ProgressRefTest do + use ExUnit.Case, async: true + + alias Phantom.Request + alias Phantom.Session + + @pubsub Phantom.ProgressRefTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + defp session(params) do + request = %Request{id: "req-1", method: "tools/call", params: params} + Session.new("session-1", pubsub: @pubsub, pid: self(), request: request) + end + + describe "progress_ref/1" do + test "is a JSON-safe reference to the request" do + ref = Session.progress_ref(session(%{"_meta" => %{"progressToken" => "tok"}})) + + assert ref == %{ + "pubsub" => "Elixir.Phantom.ProgressRefTest.PubSub", + "session_id" => "session-1", + "request_id" => "req-1", + "progress_token" => "tok" + } + + assert ref |> JSON.encode!() |> JSON.decode!() == ref + end + + test "is nil when the client did not ask for progress" do + assert Session.progress_ref(session(%{})) == nil + end + + test "is nil without PubSub" do + session = %{session(%{"_meta" => %{"progressToken" => "tok"}}) | pubsub: nil} + assert Session.progress_ref(session) == nil + end + end + + describe "notify_progress/4 with a progress ref" do + test "reaches the request's stream through PubSub" do + ref = Session.progress_ref(session(%{"_meta" => %{"progressToken" => "tok"}})) + Phantom.Tracker.subscribe_request(@pubsub, "session-1", "req-1") + + assert :ok = Session.notify_progress(ref, 50, 100, "Halfway") + assert_receive {:"$gen_cast", {:progress, "tok", 50, 100, "Halfway"}} + end + + test "works after a JSON round trip" do + ref = + session(%{"_meta" => %{"progressToken" => 7}}) + |> Session.progress_ref() + |> JSON.encode!() + |> JSON.decode!() + + Phantom.Tracker.subscribe_request(@pubsub, "session-1", "req-1") + + assert :ok = Session.notify_progress(ref, 1) + assert_receive {:"$gen_cast", {:progress, 7, 1, nil, nil}} + end + + test "does nothing for a nil ref" do + assert :ok = Session.notify_progress(nil, 50, 100, "Halfway") + end + end +end diff --git a/test/phantom/request_test.exs b/test/phantom/request_test.exs index ed4f099..1bcb939 100644 --- a/test/phantom/request_test.exs +++ b/test/phantom/request_test.exs @@ -159,6 +159,18 @@ defmodule Phantom.RequestTest do end end + describe "missing_task_capability/0" do + test "names the Tasks extension" do + assert Request.missing_task_capability() == %{ + code: -32021, + message: "Missing required client capability", + data: %{ + requiredCapabilities: %{extensions: %{"io.modelcontextprotocol/tasks" => %{}}} + } + } + end + end + describe "trace_context/1" do test "extracts the three W3C fields when present" do {:ok, request} = diff --git a/test/phantom/session_call_test.exs b/test/phantom/session_call_test.exs new file mode 100644 index 0000000..2591f3b --- /dev/null +++ b/test/phantom/session_call_test.exs @@ -0,0 +1,44 @@ +defmodule Phantom.SessionCallTest do + use ExUnit.Case, async: true + + @pubsub Phantom.SessionCallTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + defp start_stream(session_id, delay) do + spawn_link(fn -> + Process.sleep(delay) + Phantom.Tracker.subscribe_session(@pubsub, session_id) + + receive do + {:"$gen_call", from, message} -> GenServer.reply(from, {:handled, message}) + end + + Process.sleep(:infinity) + end) + end + + test "reaches a session stream that subscribes after the call starts" do + start_stream("late", 300) + + assert Phantom.Tracker.call_session(@pubsub, "late", :set_level, 2_000) == + {:handled, :set_level} + end + + test "returns the first reply once" do + start_stream("ready", 0) + Process.sleep(50) + + assert Phantom.Tracker.call_session(@pubsub, "ready", :set_level, 2_000) == + {:handled, :set_level} + + refute_receive _ + end + + test "is an error when no stream answers in time" do + assert Phantom.Tracker.call_session(@pubsub, "missing", :set_level, 300) == :error + end +end diff --git a/test/phantom/task_subscription_test.exs b/test/phantom/task_subscription_test.exs new file mode 100644 index 0000000..51e00c6 --- /dev/null +++ b/test/phantom/task_subscription_test.exs @@ -0,0 +1,259 @@ +defmodule Phantom.TaskSubscriptionTest.Router do + use Phantom.Router, name: "Task subscription test" + + def connect(session, %Plug.Conn{} = conn) do + user_id = conn |> Plug.Conn.get_req_header("x-user-id") |> List.first() + {:ok, Phantom.Session.assign(session, user_id: user_id)} + end + + def connect(session, _context), do: {:ok, session} + + def get_task(task_id, session) do + owners = :persistent_term.get({__MODULE__, :owners}, %{}) + + status = :persistent_term.get({__MODULE__, :status}, :working) + + if Map.get(owners, task_id) == session.assigns[:user_id], + do: {:ok, task(task_id, status)}, + else: {:error, :not_found} + end + + def task(id, status \\ :working) do + attrs = [id: id, status: status, created_at: ~U[2025-11-25 10:30:00Z]] + + if status == :completed, + do: Phantom.Tasks.new([result: %{content: [%{type: "text", text: "Done"}]}] ++ attrs), + else: Phantom.Tasks.new(attrs) + end +end + +defmodule Phantom.TaskSubscriptionTest.CustomRouter do + use Phantom.Router, name: "Task subscription custom authorization test" + + def get_task(id, _session), do: {:ok, Phantom.TaskSubscriptionTest.Router.task(id)} + + def authorize_task_subscriptions(task_ids, _session) do + case :persistent_term.get({__MODULE__, :return}, :all) do + :raise -> raise "authorization failed" + :invalid -> %{allowed: task_ids} + :all -> task_ids + allowed -> allowed + end + end +end + +defmodule Phantom.TaskSubscriptionTest do + use ExUnit.Case + + import ExUnit.CaptureLog + import Phantom.TestDispatcher + import Plug.Conn + + alias Phantom.TaskSubscriptionTest.CustomRouter + alias Phantom.TaskSubscriptionTest.Router + alias Phantom.Session + + @tasks_ext "io.modelcontextprotocol/tasks" + + setup do + start_supervised({Phoenix.PubSub, name: Test.PubSub}) + start_supervised({Phantom.Tracker, [name: Phantom.Tracker, pubsub_server: Test.PubSub]}) + Phantom.Cache.register(Router) + Phantom.Cache.register(CustomRouter) + :persistent_term.put({Router, :owners}, %{"alice-1" => "alice", "bob-1" => "bob"}) + + on_exit(fn -> + :persistent_term.erase({Router, :owners}) + :persistent_term.erase({Router, :status}) + :persistent_term.erase({CustomRouter, :return}) + end) + + :ok + end + + describe "authorize_task_subscriptions" do + test "defaults to the tasks get_task/2 returns" do + session = Session.new("s", router: Router) |> Session.assign(user_id: "alice") + + assert Phantom.Router.authorize_task_subscriptions(Router, ["alice-1", "bob-1"], session) == + ["alice-1"] + end + + test "keeps only requested IDs from a custom callback" do + session = Session.new("s", router: CustomRouter) + :persistent_term.put({CustomRouter, :return}, ["a", "not-requested"]) + + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a", "b"], session) == + ["a"] + end + + test "invalid callback results and exceptions fail closed" do + session = Session.new("s", router: CustomRouter) + + :persistent_term.put({CustomRouter, :return}, :invalid) + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == [] + + :persistent_term.put({CustomRouter, :return}, nil) + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == [] + + :persistent_term.put({CustomRouter, :return}, :raise) + + assert capture_log(fn -> + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == + [] + end) =~ "failed closed" + end + end + + describe "subscriptions/listen with taskIds" do + test "acknowledges and tracks only authorized tasks" do + stream_pid = listen(["alice-1", "bob-1"], "alice") + + assert_notify(%{ + method: "notifications/subscriptions/acknowledged", + params: %{notifications: %{"taskIds" => ["alice-1"]}} + }) + + wait_for_task_listeners(["alice-1"]) + tracked = Enum.map(Phantom.Tracker.list_task_listeners(), &elem(&1, 0)) + refute "bob-1" in tracked + + Session.finish(stream_pid) + end + + test "sends notifications/tasks with the stored task" do + stream_pid = listen(["alice-1", "bob-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :status}, :completed) + task = Router.task("alice-1", :completed) + # A stale copy still sends the stored task. + Phantom.Tracker.notify_task_updated(Router.task("alice-1", :working)) + Phantom.Tracker.notify_task_updated("bob-1") + + expected = Phantom.Tasks.to_json(task) + + assert_notify(%{ + method: "notifications/tasks", + params: %{_meta: %{"io.modelcontextprotocol/subscriptionId" => "listen:0"}} = params + }) + + assert Map.delete(params, :_meta) == expected + + refute_receive {:response, nil, "message", + %{method: "notifications/tasks", params: %{taskId: "bob-1"}}} + + Session.finish(stream_pid) + end + + test "accepts a task ID" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + Phantom.Tracker.notify_task_updated("alice-1") + + assert_notify(%{ + method: "notifications/tasks", + params: %{taskId: "alice-1", status: "working"} + }) + + Session.finish(stream_pid) + end + + test "stops following a task once it finishes" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :status}, :completed) + Phantom.Tracker.notify_task_updated("alice-1") + assert_notify(%{method: "notifications/tasks", params: %{status: "completed"}}) + + wait_for_no_task_listeners("alice-1") + Session.finish(stream_pid) + end + + test "re-authorizes before each notification" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :owners}, %{"alice-1" => "bob"}) + Phantom.Tracker.notify_task_updated(Router.task("alice-1", :completed)) + + refute_receive {:response, nil, "message", %{method: "notifications/tasks"}} + + Session.finish(stream_pid) + end + + test "ignores an empty taskIds list when the request did not declare the extension" do + stream_pid = listen([], "alice", %{}) + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + Session.finish(stream_pid) + end + + test "is a missing capability error when the request did not declare the extension" do + listen(["alice-1"], "alice", %{}) + + assert_receive {:conn, conn} + error = JSON.decode!(conn.resp_body)["error"] + assert error["code"] == -32021 + assert error["data"] == %{"requiredCapabilities" => %{"extensions" => %{@tasks_ext => %{}}}} + end + end + + defp listen(task_ids, user_id, capabilities \\ %{"extensions" => %{@tasks_ext => %{}}}) do + :post + |> Plug.Test.conn("/mcp", %{ + "jsonrpc" => "2.0", + "id" => "listen:0", + "method" => "subscriptions/listen", + "params" => %{ + "notifications" => %{"taskIds" => task_ids}, + "_meta" => %{ + "io.modelcontextprotocol/protocolVersion" => "2026-07-28", + "io.modelcontextprotocol/clientCapabilities" => capabilities + } + } + }) + |> put_req_header("content-type", "application/json") + |> put_req_header("mcp-protocol-version", "2026-07-28") + |> put_req_header("mcp-method", "subscriptions/listen") + |> call(%{ + router: Router, + before_call: &put_req_header(&1, "x-user-id", user_id) + }) + end + + defp wait_for_task_listeners(expected, attempts \\ 50) + + defp wait_for_task_listeners(expected, attempts) when attempts > 0 do + actual = Phantom.Tracker.list_task_listeners() |> Enum.map(&elem(&1, 0)) |> MapSet.new() + + if MapSet.subset?(MapSet.new(expected), actual) do + :ok + else + Process.sleep(10) + wait_for_task_listeners(expected, attempts - 1) + end + end + + defp wait_for_no_task_listeners(task_id, attempts \\ 50) + + defp wait_for_no_task_listeners(task_id, attempts) when attempts > 0 do + if Enum.any?(Phantom.Tracker.list_task_listeners(), &match?({^task_id, _}, &1)) do + Process.sleep(10) + wait_for_no_task_listeners(task_id, attempts - 1) + else + :ok + end + end + + defp wait_for_no_task_listeners(task_id, 0), + do: flunk("task listener was still tracked: #{inspect(task_id)}") + + defp wait_for_task_listeners(expected, 0), + do: flunk("task listeners were not tracked: #{inspect(expected)}") +end diff --git a/test/phantom/tasks_dispatch_test.exs b/test/phantom/tasks_dispatch_test.exs new file mode 100644 index 0000000..061983f --- /dev/null +++ b/test/phantom/tasks_dispatch_test.exs @@ -0,0 +1,380 @@ +defmodule Phantom.TasksDispatchTest do + use ExUnit.Case, async: true + + alias Phantom.Session + + @tasks_ext "io.modelcontextprotocol/tasks" + + defmodule Router do + use Phantom.Router, name: "Tasks", vsn: "1.0" + + def get_task("expired", _session), do: {:error, :expired} + def get_task("broken", _session), do: {:error, Phantom.Request.internal_error("broken")} + + def get_task(id, %{assigns: %{owner: "alice"}}) when id in ~w[working input rejecting], + do: {:ok, task(id)} + + def get_task(_task_id, _session), do: {:error, :not_found} + + def update_task(%{id: "rejecting"}, _responses, _session), + do: {:error, Phantom.Request.invalid_params(%{inputResponses: "rejected"})} + + def update_task(task, responses, session) do + send(session.assigns.test_pid, {:update_task, task.id, responses}) + :ok + end + + def cancel_task(task, session) do + send(session.assigns.test_pid, {:cancel_task, task.id}) + :ok + end + + def task("working"), + do: Phantom.Tasks.new(id: "working", status: :working, created_at: ~U[2025-11-25 10:30:00Z]) + + def task(id) when id in ~w[input rejecting] do + Phantom.Tasks.new( + id: id, + status: :input_required, + created_at: ~U[2025-11-25 10:30:00Z], + input_requests: %{ + "name" => %{method: "elicitation/create", params: %{mode: "form", message: "Name?"}}, + "age" => %{method: "elicitation/create", params: %{mode: "form", message: "Age?"}} + } + ) + end + + tool :always_task, description: "Always answers with a task" + tool :input_task, description: "Answers with a task that already needs input" + + def input_task(_params, session), do: {:reply, task("input"), session} + + def always_task(_params, session) do + {:reply, + Phantom.Tasks.new( + id: "task-1", + status: :working, + created_at: ~U[2025-11-25 10:30:00Z], + ttl_ms: 60_000, + poll_interval_ms: 5_000 + ), session} + end + end + + defmodule NoTasksRouter do + use Phantom.Router, name: "NoTasks", vsn: "1.0" + end + + defmodule BadRouter do + use Phantom.Router, name: "Bad", vsn: "1.0" + + def get_task("plain-map", _session), do: {:ok, %{id: "plain-map"}} + def get_task(id, _session), do: {:ok, Phantom.TasksDispatchTest.Router.task(id)} + + def update_task(task, _responses, _session), do: {:ok, task} + end + + defmodule GetOnlyRouter do + use Phantom.Router, name: "GetOnly", vsn: "1.0" + + def get_task(id, _session), do: {:ok, Phantom.TasksDispatchTest.Router.task(id)} + end + + import Phantom.Test + + setup do + Phantom.Test.start(router: Router) + Phantom.Test.start(router: NoTasksRouter) + Phantom.Test.start(router: GetOnlyRouter) + Phantom.Test.start(router: BadRouter) + :ok + end + + defp tasks_capabilities, do: %{extensions: %{@tasks_ext => %{}}} + + defp session(router \\ Router, opts \\ []) do + build_session( + router, + Keyword.merge( + [ + protocol_version: "2026-07-28", + client_capabilities: tasks_capabilities(), + assigns: %{owner: "alice", test_pid: self()} + ], + opts + ) + ) + end + + defp dispatch(session, method, params \\ %{}) do + request = %{build_request(method, params: params) | meta: session.request.meta} + session = %{session | pid: self(), request: request} + session.router.dispatch_method([method, params, request, session]) + end + + describe "Session.tasks_supported?/1" do + test "is true when the request declares the extension" do + assert Session.tasks_supported?(session()) + end + + test "is false when the request does not declare the extension" do + refute Session.tasks_supported?(session(Router, client_capabilities: %{})) + end + + test "is false under a legacy protocol" do + refute Session.tasks_supported?(build_session(Router)) + end + end + + describe "server/discover" do + test "advertises the extension when the router implements get_task/2" do + assert {:reply, %{capabilities: %{extensions: %{@tasks_ext => %{}}}}, _} = + dispatch(session(), "server/discover") + end + + test "does not advertise the extension otherwise" do + assert {:reply, %{capabilities: capabilities}, _} = + dispatch(session(NoTasksRouter), "server/discover") + + refute get_in(capabilities, [:extensions, @tasks_ext]) + end + end + + describe "tools/call returning a task" do + test "responds with a CreateTaskResult" do + assert call_tool(session(), :always_task) == %{ + resultType: "task", + taskId: "task-1", + status: "working", + createdAt: "2025-11-25T10:30:00Z", + lastUpdatedAt: "2025-11-25T10:30:00Z", + ttlMs: 60_000, + pollIntervalMs: 5_000 + } + end + + test "leaves status-specific fields out of the CreateTaskResult" do + result = call_tool(session(), :input_task) + assert %{resultType: "task", taskId: "input", status: "input_required"} = result + refute Map.has_key?(result, :inputRequests) + end + + test "keeps resultType task through result normalization" do + request = %{build_request("tools/call") | meta: session().request.meta} + + assert %{resultType: "task"} = + Phantom.Request.normalize_result(%{resultType: "task", taskId: "t"}, request) + end + + test "is a missing capability error when the request did not declare the extension" do + assert {:jsonrpc_error, error} = + call_tool(session(Router, client_capabilities: %{}), :always_task) + + assert error == %{ + code: -32021, + message: "Missing required client capability", + data: %{requiredCapabilities: %{extensions: %{@tasks_ext => %{}}}} + } + end + + test "is a missing capability error under a legacy protocol" do + assert {:jsonrpc_error, %{code: -32021}} = call_tool(build_session(Router), :always_task) + end + end + + @not_found %{code: -32602, message: "Failed to retrieve task: Task not found"} + + describe "tasks/get" do + test "returns the task" do + assert {:reply, result, _} = dispatch(session(), "tasks/get", %{"taskId" => "working"}) + assert result == Phantom.Tasks.to_json(Router.task("working")) + end + + test "inlines outstanding input requests" do + assert {:reply, %{status: "input_required", inputRequests: %{"name" => _, "age" => _}}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "input"}) + end + + test "is not found when get_task/2 denies the caller" do + session = session(Router, assigns: %{owner: "mallory", test_pid: self()}) + + assert {:error, @not_found, _} = dispatch(session, "tasks/get", %{"taskId" => "working"}) + end + + test "is not found for an unknown task" do + assert {:error, @not_found, _} = dispatch(session(), "tasks/get", %{"taskId" => "nope"}) + end + + test "reports an expired task" do + assert {:error, %{code: -32602, message: "Failed to retrieve task: Task has expired"}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "expired"}) + end + + test "passes through a JSON-RPC error from get_task/2" do + assert {:error, %{code: -32603, message: "broken"}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "broken"}) + end + + test "raises a clear error when get_task/2 returns an unexpected value" do + assert_raise ArgumentError, ~r/get_task\/2/, fn -> + dispatch(session(BadRouter), "tasks/get", %{"taskId" => "plain-map"}) + end + end + + test "is method not found under a legacy protocol" do + request = build_request("tasks/get", params: %{"taskId" => "working"}) + session = %{build_session(Router) | pid: self(), request: request} + + assert {:error, %{code: -32601}, _} = + Router.dispatch_method(["tasks/get", request.params, request, session]) + end + + test "requires a taskId" do + assert {:error, %{code: -32602}, _} = dispatch(session(), "tasks/get", %{}) + end + + test "is a missing capability error when the request did not declare the extension" do + assert {:error, + %{code: -32021, data: %{requiredCapabilities: %{extensions: %{@tasks_ext => %{}}}}}, + _} = + dispatch(session(Router, client_capabilities: %{}), "tasks/get", %{ + "taskId" => "working" + }) + end + + test "is method not found when the router does not implement tasks" do + assert {:error, %{code: -32601}, _} = + dispatch(session(NoTasksRouter), "tasks/get", %{"taskId" => "working"}) + end + + test "other tasks methods are method not found" do + assert {:error, %{code: -32601}, _} = dispatch(session(), "tasks/list", %{}) + end + end + + describe "tasks/update" do + test "passes only outstanding input responses to update_task/3" do + params = %{ + "taskId" => "input", + "inputResponses" => %{ + "name" => %{"action" => "accept", "content" => %{"name" => "Luca"}}, + "unknown" => %{"action" => "accept"} + } + } + + assert {:reply, %{}, _} = dispatch(session(), "tasks/update", params) + + assert_received {:update_task, "input", + %{"name" => %{"action" => "accept", "content" => %{"name" => "Luca"}}} = + responses} + + refute Map.has_key?(responses, "unknown") + end + + test "acknowledges without calling update_task/3 when nothing is outstanding" do + params = %{"taskId" => "working", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert {:reply, %{}, _} = dispatch(session(), "tasks/update", params) + refute_received {:update_task, _, _} + end + + test "passes through an error from update_task/3" do + params = %{ + "taskId" => "rejecting", + "inputResponses" => %{"name" => %{"action" => "accept"}} + } + + assert {:error, %{code: -32602, data: %{inputResponses: "rejected"}}, _} = + dispatch(session(), "tasks/update", params) + end + + test "requires inputResponses to be an object" do + assert {:error, %{code: -32602}, _} = + dispatch(session(), "tasks/update", %{"taskId" => "input", "inputResponses" => []}) + end + + test "is not found for a task the caller cannot see" do + params = %{"taskId" => "nope", "inputResponses" => %{"name" => %{"action" => "accept"}}} + assert {:error, @not_found, _} = dispatch(session(), "tasks/update", params) + end + + test "acknowledges when the router does not implement update_task/3" do + params = %{"taskId" => "input", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert {:reply, %{}, _} = dispatch(session(GetOnlyRouter), "tasks/update", params) + end + + test "raises a clear error when update_task/3 returns an unexpected value" do + params = %{"taskId" => "input", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert_raise ArgumentError, ~r/update_task\/3/, fn -> + dispatch(session(BadRouter), "tasks/update", params) + end + end + end + + describe "tasks/cancel" do + test "calls cancel_task/2 and acknowledges" do + assert {:reply, %{}, _} = dispatch(session(), "tasks/cancel", %{"taskId" => "working"}) + assert_received {:cancel_task, "working"} + end + + test "acknowledges when the router does not implement cancel_task/2" do + assert {:reply, %{}, _} = + dispatch(session(GetOnlyRouter), "tasks/cancel", %{"taskId" => "working"}) + end + + test "is not found for a task the caller cannot see" do + assert {:error, @not_found, _} = dispatch(session(), "tasks/cancel", %{"taskId" => "nope"}) + refute_received {:cancel_task, _} + end + end + + describe "Phantom.Test task helpers" do + test "call_tool/4 declares the extension with tasks: true" do + session(Router, client_capabilities: %{}) + |> call_tool(:always_task, %{}, tasks: true) + |> assert_task(id: "task-1", status: :working) + end + + test "tasks: true requires a stateless session" do + assert_raise ArgumentError, ~r/protocol_version/, fn -> + call_tool(build_session(Router), :always_task, %{}, tasks: true) + end + end + + test "get_task/3 returns the task" do + session(Router, client_capabilities: %{}) + |> get_task("input") + |> assert_task(status: :input_required) + end + + test "get_task/3 returns a JSON-RPC error for an unknown task" do + session() + |> get_task("nope") + |> assert_jsonrpc_error(code: -32602, message: "Failed to retrieve task: Task not found") + end + + test "update_task/4 sends input responses" do + assert update_task(session(), "input", %{name: %{"action" => "accept"}}) == %{} + assert_received {:update_task, "input", %{"name" => %{"action" => "accept"}}} + end + + test "cancel_task/3 cancels" do + assert cancel_task(session(), "working") == %{} + assert_received {:cancel_task, "working"} + end + + test "assert_task/2 fails on a mismatch" do + task = get_task(session(), "working") + + assert_raise ExUnit.AssertionError, ~r/status/, fn -> + assert_task(task, status: :completed) + end + + assert_raise ExUnit.AssertionError, ~r/expected a task/, fn -> + assert_task(%{content: []}) + end + end + end +end diff --git a/test/phantom/tasks_test.exs b/test/phantom/tasks_test.exs new file mode 100644 index 0000000..7e4b4b0 --- /dev/null +++ b/test/phantom/tasks_test.exs @@ -0,0 +1,199 @@ +defmodule Phantom.TasksTest do + use ExUnit.Case, async: true + + alias Phantom.Tasks + + @created_at ~U[2025-11-25 10:30:00Z] + @updated_at ~U[2025-11-25 10:50:00Z] + + defp attrs(overrides \\ []) do + Keyword.merge( + [ + id: "786512e2-9e0d-44bd-8f29-789f320fe840", + status: :working, + created_at: @created_at, + last_updated_at: @updated_at, + ttl_ms: 3_600_000, + poll_interval_ms: 5_000 + ], + overrides + ) + end + + describe "new/1" do + test "builds a working task" do + assert %Tasks{id: "786512e2-9e0d-44bd-8f29-789f320fe840", status: :working} = + Tasks.new(attrs()) + end + + test "accepts a map and a status string, as loaded from a database" do + task = + Tasks.new( + attrs() + |> Map.new() + |> Map.put(:status, "input_required") + |> Map.put(:input_requests, %{"a" => %{method: "elicitation/create", params: %{}}}) + ) + + assert task.status == :input_required + end + + test "defaults last_updated_at to created_at and ttl_ms to unlimited" do + task = Tasks.new(id: "t", status: :working, created_at: @created_at) + assert task.last_updated_at == @created_at + assert task.ttl_ms == nil + end + + test "rejects an unknown status" do + assert_raise ArgumentError, ~r/status/, fn -> Tasks.new(attrs(status: :done)) end + assert_raise ArgumentError, ~r/status/, fn -> Tasks.new(attrs(status: "done")) end + end + + test "requires an id and created_at" do + assert_raise ArgumentError, ~r/:id/, fn -> Tasks.new(attrs(id: nil)) end + assert_raise ArgumentError, ~r/:id/, fn -> Tasks.new(attrs(id: "")) end + assert_raise ArgumentError, ~r/:created_at/, fn -> Tasks.new(attrs(created_at: nil)) end + + assert_raise ArgumentError, ~r/:last_updated_at/, fn -> + Tasks.new(attrs(last_updated_at: "2025-11-25T10:50:00Z")) + end + end + + test "requires a result when completed" do + assert_raise ArgumentError, ~r/:result/, fn -> Tasks.new(attrs(status: :completed)) end + end + + test "requires a JSON-RPC error when failed" do + assert_raise ArgumentError, ~r/:error/, fn -> Tasks.new(attrs(status: :failed)) end + + assert_raise ArgumentError, ~r/:error/, fn -> + Tasks.new(attrs(status: :failed, error: %{message: "no code"})) + end + end + + test "requires input requests when input_required" do + assert_raise ArgumentError, ~r/:input_requests/, fn -> + Tasks.new(attrs(status: :input_required)) + end + + assert_raise ArgumentError, ~r/:input_requests/, fn -> + Tasks.new(attrs(status: :input_required, input_requests: %{})) + end + end + + test "rejects negative or non-integer durations" do + assert_raise ArgumentError, ~r/:ttl_ms/, fn -> Tasks.new(attrs(ttl_ms: -1)) end + + assert_raise ArgumentError, ~r/:poll_interval_ms/, fn -> + Tasks.new(attrs(poll_interval_ms: 1.5)) + end + end + + test "converts Phantom.Elicit input requests" do + elicit = + Phantom.Elicit.form(%{ + message: "Please enter your name.", + requested_schema: [%{name: "name", type: :string, required: true}] + }) + + task = Tasks.new(attrs(status: :input_required, input_requests: %{"name" => elicit})) + + assert %{"name" => %{method: "elicitation/create", params: %{mode: "form"}}} = + task.input_requests + end + end + + describe "encoding a task changed with struct update syntax" do + test "validates it" do + task = Tasks.new(attrs()) + + assert_raise ArgumentError, ~r/:result/, fn -> + Tasks.to_json(%{task | status: :completed}) + end + end + + test "normalizes a status string" do + task = Tasks.new(attrs()) + assert Tasks.to_json(%{task | status: "cancelled"}).status == "cancelled" + assert Tasks.to_create_result(%{task | status: "cancelled"}).status == "cancelled" + end + end + + describe "generate_id/0" do + test "is URL-safe and unique" do + id = Tasks.generate_id() + assert id =~ ~r/^[A-Za-z0-9_-]{43}$/ + refute id == Tasks.generate_id() + end + end + + describe "to_json/1" do + test "encodes a working task as in the spec" do + assert Tasks.to_json(Tasks.new(attrs(status_message: "The operation is now in progress."))) == + %{ + taskId: "786512e2-9e0d-44bd-8f29-789f320fe840", + status: "working", + statusMessage: "The operation is now in progress.", + createdAt: "2025-11-25T10:30:00Z", + lastUpdatedAt: "2025-11-25T10:50:00Z", + ttlMs: 3_600_000, + pollIntervalMs: 5_000 + } + end + + test "keeps ttlMs as null when unlimited and omits a missing poll interval" do + json = Tasks.to_json(Tasks.new(attrs(ttl_ms: nil, poll_interval_ms: nil))) + assert Map.fetch!(json, :ttlMs) == nil + refute Map.has_key?(json, :pollIntervalMs) + end + + test "inlines the input requests when input_required" do + request = %{ + method: "elicitation/create", + params: %{mode: "form", message: "Please enter your name."} + } + + json = + Tasks.to_json( + Tasks.new(attrs(status: :input_required, input_requests: %{"name" => request})) + ) + + assert json.status == "input_required" + assert json.inputRequests == %{"name" => request} + end + + test "inlines the result when completed" do + result = %{content: [%{type: "text", text: "Hello, Luca!"}], isError: false} + json = Tasks.to_json(Tasks.new(attrs(status: :completed, result: result))) + assert json.status == "completed" + assert json.result == result + refute Map.has_key?(json, :inputRequests) + end + + test "inlines the error when failed" do + error = %{code: -32603, message: "API rate limit exceeded"} + json = Tasks.to_json(Tasks.new(attrs(status: :failed, error: error))) + assert json.status == "failed" + assert json.error == error + end + + test "omits status-specific fields for other statuses" do + json = + Tasks.to_json( + Tasks.new( + attrs( + status: :cancelled, + result: %{content: []}, + error: %{code: -32603, message: "x"}, + input_requests: %{"a" => %{method: "elicitation/create", params: %{}}} + ) + ) + ) + + assert json.status == "cancelled" + refute Map.has_key?(json, :result) + refute Map.has_key?(json, :error) + refute Map.has_key?(json, :inputRequests) + end + end +end diff --git a/test/support/app/mcp/router.ex b/test/support/app/mcp/router.ex index aff5d70..16ae983 100644 --- a/test/support/app/mcp/router.ex +++ b/test/support/app/mcp/router.ex @@ -57,6 +57,11 @@ defmodule Test.MCP.Router do def connect(session, _context), do: {:ok, session} + def get_task("task-" <> _ = id, _session), + do: {:ok, Phantom.Tasks.new(id: id, status: :working, created_at: ~U[2025-11-25 10:30:00Z])} + + def get_task(_id, _session), do: {:error, :not_found} + def terminate(session) do if Test.SessionStore.running?(), do: Test.SessionStore.delete(session.id) {:ok, session} @@ -247,6 +252,7 @@ defmodule Test.MCP.Router do tool :really_long_async_tool, AsyncModule, description: "this will notify of progress" tool :timeout_async_tool, AsyncModule, description: "this will timeout!" tool :hanging_tool, description: "Never responds, so only cancellation ends it" + tool :remote_progress_tool, description: "Reports progress from another node, then responds" @audio File.read!(@base <> "/game-over.wav") def audio_tool(_params, session) do @@ -320,6 +326,20 @@ defmodule Test.MCP.Router do def hanging_tool(_params, session), do: {:noreply, session} + # Another node stands in for a job that only has the progress reference. + def remote_progress_tool(_params, session) do + ref = Session.progress_ref(session) + target = Enum.find(Node.list(), node(), &String.starts_with?(Atom.to_string(&1), "node")) + + Node.spawn(target, fn -> + Session.notify_progress(ref, 50, 100, "Halfway on #{node()}") + Process.sleep(200) + Session.respond(session, Tool.text("Done")) + end) + + {:noreply, session} + end + def client_log_tool(params, session) do message = params["message"] || "client-log-test" Phantom.ClientLogger.log(session, :info, %{message: message}, "test") diff --git a/test/support/conformance.ex b/test/support/conformance.ex index ab5188b..4051abf 100644 --- a/test/support/conformance.ex +++ b/test/support/conformance.ex @@ -58,15 +58,36 @@ defmodule Phantom.Test.Conformance do end defp npx(args) do + results = + Path.join(System.tmp_dir!(), "phantom-conformance-#{System.unique_integer([:positive])}") + {output, status} = - System.cmd("npx", ["conformance", "server", "--timeout", "15000" | args], + System.cmd( + "npx", + ["conformance", "server", "--timeout", "15000", "--output-dir", results | args], stderr_to_stdout: true, env: [{"NO_COLOR", "1"}, {"FORCE_COLOR", "0"}] ) + output = if status == 0, do: output, else: output <> failed_checks(results) + File.rm_rf(results) {status, output} end + # The summary names only the scenario, so the report adds each failed check. + defp failed_checks(results) do + failures = + for path <- Path.wildcard(Path.join(results, "*/checks.json")), + check <- path |> File.read!() |> JSON.decode!(), + check["status"] == "FAILURE" do + scenario = path |> Path.dirname() |> Path.basename() + details = if check["details"], do: "\n " <> JSON.encode!(check["details"]), else: "" + " #{scenario} #{check["id"]}: #{check["errorMessage"]}#{details}" + end + + Enum.join(["\nFailed checks (expected failures included):" | failures], "\n") + end + # The CLI takes one baseline file, so the topology's additions are appended # to the revision's entries in a temporary copy. defp baseline(topology, revision) do