Phantom.Test (phantom_mcp v0.7.3)

Copy Markdown View Source

Test helpers for asserting on tools, prompts, and resources defined in a Phantom.Router.

Drives router handlers directly without booting an HTTP transport, so unit-level tests can assert on the final response of a tool/prompt/ resource regardless of whether the handler returned synchronously ({:reply, _, _}) or asynchronously ({:noreply, _} + Session.respond/2 from a Task).

Example

defmodule MyApp.MCP.RouterTest do
  use ExUnit.Case, async: true
  import Phantom.Test

  setup do
    Phantom.Test.start(router: MyApp.MCP.Router)
    {:ok, session: build_session(MyApp.MCP.Router)}
  end

  test "echo tool", %{session: session} do
    result = call_tool(session, :echo_tool, %{message: "hi"})
    assert_tool_text(result, "hi")
  end

  test "tool that elicits info", %{session: session} do
    expect_elicit(fn _elicit ->
      {:ok, %{"action" => "accept", "content" => %{"name" => "Joe"}}}
    end)

    result = call_tool(session, :elicit_tool, %{})
    assert_tool_text(result, ~r/Joe/)
  end
end

PubSub

PubSub-driven features (Phantom.Tracker, resource subscriptions, list-changed notifications across nodes) are opt-in. Pass :pubsub to start/1 to enable them:

Phantom.Test.start(router: MyApp.MCP.Router, pubsub: MyApp.PubSub)

Side-channel assertions (assert_progress_seen/1, assert_client_log_seen/1) work without PubSub: they read messages cast directly to the test process.

Summary

Functions

assert_client_log_seen(opts \\ [])

Assert that a client log was emitted during the most recent dispatch.

assert_elicitation_required(result, opts \\ [])

Assert the dispatcher returned an :elicitation_required JSON-RPC error (code -32042).

assert_jsonrpc_error(result, opts)

Assert the dispatcher returned a JSON-RPC error.

assert_progress_seen(opts \\ [])

Assert that a progress notification was emitted during the most recent dispatch.

assert_prompt_message(result, opts)

Assert a get_prompt/3 result contains a message matching opts.

assert_resource_blob(result, opts \\ [])

Assert a read_resource/3 result contains a base64-encoded blob.

assert_resource_text(result, expected)

Assert a read_resource/3 result contains text content matching expected.

assert_task(result, opts \\ [])

Assert the result is a task, from a tool that answered with one or from get_task/3.

assert_tool_audio(result, opts \\ [])

Assert the tool returned audio content.

assert_tool_embedded_resource(result, opts \\ [])

Assert the tool returned an embedded resource.

assert_tool_error(result, expected)

Assert the tool returned an error response (isError: true) with the given message.

assert_tool_image(result, opts \\ [])

Assert the tool returned image content.

assert_tool_resource_link(result, opts \\ [])

Assert the tool returned a resource link.

assert_tool_text(result, expected)

Assert the tool returned a text content payload.

build_request(method, opts \\ [])

Build a fresh Phantom.Request struct with a random id and the given method.

build_session(router, opts \\ [])

Build a Phantom.Session for use with the blocking dispatchers below.

call_tool(session, name, args \\ %{}, opts \\ [])

Call a tool by name and block until the final response arrives.

cancel_task(session, task_id, opts \\ [])

Dispatch tasks/cancel for a task. See get_task/3.

complete_prompt(session, prompt_name, arg, value, opts \\ [])

Run prompt completion. Returns the dispatcher's reply map (under :completion) or {:jsonrpc_error, error_map}.

complete_resource(session, uri_template, arg, value, opts \\ [])

Run resource completion against a URI template. See complete_prompt/5.

expect_elicit(fun)

Register a responder for form-mode elicitations.

expect_elicit_url(fun)

Register a responder for url-mode elicitations. See expect_elicit/1.

flush_notifications()

Drain progress notifications and client logs cast to the test mailbox.

get_prompt(session, name, args \\ %{}, opts \\ [])

Get a prompt by name. Blocks until the final response arrives.

get_task(session, task_id, opts \\ [])

Dispatch tasks/get for a task. Declares the Tasks extension on the request, so the session must be built with :protocol_version.

list_resources(session, cursor \\ nil, opts \\ [])

Dispatch resources/list with an optional cursor.

read_resource(session, name, path_params \\ [], opts \\ [])

Read a resource by name and path params; blocks until the final response.

refute_client_log_seen()

Refute that any client log was emitted.

refute_progress_seen()

Refute that any progress notification was emitted.

start(opts)

Register the router with Phantom.Cache and, optionally, start Phantom.Tracker against the given :pubsub.

update_task(session, task_id, input_responses, opts \\ [])

Dispatch tasks/update with responses to a task's input requests. See get_task/3.

Functions

assert_client_log_seen(opts \\ [])

Assert that a client log was emitted during the most recent dispatch.

Options: :level (atom), :domain (string), :data (map for partial match).

assert_elicitation_required(result, opts \\ [])

Assert the dispatcher returned an :elicitation_required JSON-RPC error (code -32042).

Options: :message matches the elicitation message string.

assert_jsonrpc_error(result, opts)

Assert the dispatcher returned a JSON-RPC error.

Options: :code, :message, :data.

assert_progress_seen(opts \\ [])

Assert that a progress notification was emitted during the most recent dispatch.

Options:

  • :progress - exact value
  • :total - exact total
  • :steps - exact number of progress notifications observed

Drains any matching messages from the mailbox.

assert_prompt_message(result, opts)

Assert a get_prompt/3 result contains a message matching opts.

Options: :role, :type, :text (binary or regex).

assert_resource_blob(result, opts \\ [])

Assert a read_resource/3 result contains a base64-encoded blob.

When :mime_type is provided, asserts on it too.

assert_resource_text(result, expected)

Assert a read_resource/3 result contains text content matching expected.

Accepts both {:ok, uri, content} and a raw content map.

assert_task(result, opts \\ [])

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).

assert_tool_audio(result, opts \\ [])

Assert the tool returned audio content.

assert_tool_embedded_resource(result, opts \\ [])

Assert the tool returned an embedded resource.

assert_tool_error(result, expected)

Assert the tool returned an error response (isError: true) with the given message.

assert_tool_image(result, opts \\ [])

Assert the tool returned image content.

assert_tool_resource_link(result, opts \\ [])

Assert the tool returned a resource link.

assert_tool_text(result, expected)

Assert the tool returned a text content payload.

expected may be a binary or a regex.

build_request(method, opts \\ [])

Build a fresh Phantom.Request struct with a random id and the given method.

build_session(router, opts \\ [])

@spec build_session(
  module(),
  keyword()
) :: Phantom.Session.t()

Build a Phantom.Session for use with the blocking dispatchers below.

The session's :pid is wired during dispatch — you do not need to set it. Options:

  • :id - session id (default: a fresh UUIDv7)
  • :assigns - session assigns (default: %{})
  • :allowed_tools / :allowed_prompts / :allowed_resource_templates
  • :pubsub - attach a pubsub server to the session.
  • :client_capabilities - override the default capabilities map. Defaults to %{elicitation: %{"url" => %{}}, sampling: %{}, roots: %{}} so both form-mode and url-mode elicitation are supported in tests.
  • :protocol_version - a stateless protocol version such as "2026-07-28". Requests then carry the per-request _meta that protocol requires. Omit it to use the initialize-based protocol.
  • :log_level - with :protocol_version, the client log level each request asks for (default: none, so no client logs are sent).

call_tool(session, name, args \\ %{}, opts \\ [])

@spec call_tool(
  Phantom.Session.t(),
  atom() | String.t(),
  map() | keyword(),
  keyword()
) :: any()

Call a tool by name and block until the final response arrives.

Returns the unwrapped result payload (the value the MCP client would see under "result") for {:reply, _, _} and Session.respond/2 paths, or {:jsonrpc_error, error_map} if the dispatcher returned an error tuple (e.g. validation failure, {:elicitation_required, _}).

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.

cancel_task(session, task_id, opts \\ [])

@spec cancel_task(Phantom.Session.t(), String.t(), keyword()) :: any()

Dispatch tasks/cancel for a task. See get_task/3.

complete_prompt(session, prompt_name, arg, value, opts \\ [])

@spec complete_prompt(
  Phantom.Session.t(),
  atom() | String.t(),
  String.t(),
  String.t(),
  keyword()
) ::
  any()

Run prompt completion. Returns the dispatcher's reply map (under :completion) or {:jsonrpc_error, error_map}.

complete_resource(session, uri_template, arg, value, opts \\ [])

@spec complete_resource(
  Phantom.Session.t(),
  String.t(),
  String.t(),
  String.t(),
  keyword()
) :: any()

Run resource completion against a URI template. See complete_prompt/5.

expect_elicit(fun)

@spec expect_elicit((Phantom.Elicit.t() -> any())) :: :ok

Register a responder for form-mode elicitations.

The responder receives a Phantom.Elicit struct and must return {:ok, response_map} (e.g. {:ok, %{"action" => "accept", "content" => %{...}}}), :error, :not_supported, or :timeout.

Stored in the calling test process; ExUnit isolates per test.

expect_elicit_url(fun)

@spec expect_elicit_url((Phantom.Elicit.t() -> any())) :: :ok

Register a responder for url-mode elicitations. See expect_elicit/1.

flush_notifications()

@spec flush_notifications() :: :ok

Drain progress notifications and client logs cast to the test mailbox.

get_prompt(session, name, args \\ %{}, opts \\ [])

@spec get_prompt(
  Phantom.Session.t(),
  atom() | String.t(),
  map() | keyword(),
  keyword()
) :: any()

Get a prompt by name. Blocks until the final response arrives.

get_task(session, task_id, opts \\ [])

@spec get_task(Phantom.Session.t(), String.t(), keyword()) :: any()

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}.

list_resources(session, cursor \\ nil, opts \\ [])

@spec list_resources(Phantom.Session.t(), String.t() | nil, keyword()) :: any()

Dispatch resources/list with an optional cursor.

read_resource(session, name, path_params \\ [], opts \\ [])

@spec read_resource(
  Phantom.Session.t(),
  atom() | String.t(),
  keyword() | map(),
  keyword()
) :: any()

Read a resource by name and path params; blocks until the final response.

Returns {:ok, uri, content} on success, or {:jsonrpc_error, error_map} if dispatch returned an error tuple.

Path params accept atom or string keys.

refute_client_log_seen()

Refute that any client log was emitted.

refute_progress_seen()

Refute that any progress notification was emitted.

start(opts)

@spec start(keyword()) :: :ok

Register the router with Phantom.Cache and, optionally, start Phantom.Tracker against the given :pubsub.

Options:

  • :router - required. The MCP router module.
  • :pubsub - optional. Phoenix.PubSub server name. If supplied, ensures Phantom.Tracker is running and bound to it.

update_task(session, task_id, input_responses, opts \\ [])

@spec update_task(Phantom.Session.t(), String.t(), map() | keyword(), keyword()) ::
  any()

Dispatch tasks/update with responses to a task's input requests. See get_task/3.