# `Phantom.Test`
[🔗](https://github.com/dbernheisel/phantom_mcp/blob/main/lib/phantom/test.ex#L1)

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.

# `assert_client_log_seen`

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`

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

Options: `:message` matches the elicitation message string.

# `assert_jsonrpc_error`

Assert the dispatcher returned a JSON-RPC error.

Options: `:code`, `:message`, `:data`.

# `assert_progress_seen`

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`

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

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

# `assert_resource_blob`

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

When `:mime_type` is provided, asserts on it too.

# `assert_resource_text`

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

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

# `assert_task`

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`

Assert the tool returned audio content.

# `assert_tool_embedded_resource`

Assert the tool returned an embedded resource.

# `assert_tool_error`

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

# `assert_tool_image`

Assert the tool returned image content.

# `assert_tool_resource_link`

Assert the tool returned a resource link.

# `assert_tool_text`

Assert the tool returned a text content payload.

`expected` may be a binary or a regex.

# `build_request`

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

# `build_session`

```elixir
@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`
    - allow-lists, like in `c:Phantom.Router.connect/2`.
  * `: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`

```elixir
@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`

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

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

# `complete_prompt`

```elixir
@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`

```elixir
@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`

```elixir
@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`

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

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

# `flush_notifications`

```elixir
@spec flush_notifications() :: :ok
```

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

# `get_prompt`

```elixir
@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`

```elixir
@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`

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

Dispatch `resources/list` with an optional cursor.

# `read_resource`

```elixir
@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`

```elixir
@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`

```elixir
@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`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*