Noizu.MCP.Test (Noizu MCP v0.1.2)

Copy Markdown View Source

Test helpers for exercising Noizu.MCP.Server modules over an in-memory transport — no sockets, async: true safe.

defmodule MyApp.MCPTest do
  use ExUnit.Case, async: true
  import Noizu.MCP.Test

  setup do
    %{client: connect(MyApp.MCP)}
  end

  test "weather tool", %{client: client} do
    assert {:ok, result} = call_tool(client, "get_weather", %{"location" => "NYC"})
    assert [%{type: :text}] = result.content
    assert_progress(client)
  end
end

The returned client handle is bound to the calling process: server output is delivered to your mailbox and consumed by these helpers. Notifications that arrive while awaiting a response are buffered and visible to assert_notification/3 / assert_progress/2.

Summary

Functions

assert_notification(client, method, match \\ nil, opts \\ [])

Assert the server emitted a notification with method; returns its params. When match (a map) is given, every key/value in it must be present in the notification params.

assert_progress(client, opts \\ [])

Assert the server emitted at least one notifications/progress; returns its params.

await(client, id, opts \\ [])

Await the response for a previously sent request id.

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

Call a tool. Returns {:ok, %ToolResult{}} | {:error, error_map}. Args use string keys.

cancel(client, id, reason \\ nil)

Send notifications/cancelled for an in-flight request id.

complete(client, ref, arg_name, value)

Request completion values. ref is {:prompt, name} or {:resource_template, uri_template}. Returns {:ok, %{values: _, total: _, has_more: _}}.

connect(server, opts \\ [])

Start a session against server (a use Noizu.MCP.Server module) and run the initialize handshake. Starts the server's supervision tree if it is not already running.

deliver_raw(client, binary)

Send a raw wire binary (escape hatch for malformed-input tests).

ensure_server_started(server)

Ensure server's supervision tree is running (shared across tests), detached from the calling process. Safe under concurrent async: true callers — waits until the tree is fully started before returning.

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

Get a prompt. Returns {:ok, %{description: _, messages: [%PromptMessage{}]}}. Args use string keys.

list_prompts(client, opts \\ [])

List all prompts (auto-paginates). Returns {:ok, [%Prompt{}]}.

list_resource_templates(client, opts \\ [])

List all resource templates (auto-paginates).

list_resources(client, opts \\ [])

List all resources (auto-paginates). Returns {:ok, [%Resource{}]}.

list_tools(client, opts \\ [])

List all tools (auto-paginates). Returns {:ok, [%Tool{}]} | {:error, error_map}.

notify(client, method, params \\ nil)

Send a notification to the server.

read_resource(client, uri, opts \\ [])

Read a resource. Returns {:ok, [%ResourceContents{}]} | {:error, error_map}.

refute_notification(client, method, opts \\ [])

Assert no notification with method arrives within opts[:timeout] (default 100ms).

request(client, method, params \\ nil, opts \\ [])

Send a request and await its result. Returns {:ok, result} | {:error, error_map}.

send_request(client, method, params \\ nil)

Send a request and return its id without awaiting the response.

set_log_level(client, level)

Set the MCP log level for this session.

subscribe(client, uri)

Subscribe to resource update notifications for uri.

unsubscribe(client, uri)

Unsubscribe from resource update notifications for uri.

Functions

assert_notification(client, method, match \\ nil, opts \\ [])

@spec assert_notification(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map() | nil,
  keyword()
) :: map() | nil

Assert the server emitted a notification with method; returns its params. When match (a map) is given, every key/value in it must be present in the notification params.

assert_progress(client, opts \\ [])

@spec assert_progress(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  keyword()
) :: map()

Assert the server emitted at least one notifications/progress; returns its params.

await(client, id, opts \\ [])

@spec await(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  integer(),
  keyword()
) :: {:ok, map()} | {:error, map()}

Await the response for a previously sent request id.

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

@spec call_tool(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map(),
  keyword()
) :: {:ok, Noizu.MCP.Types.ToolResult.t()} | {:error, map()}

Call a tool. Returns {:ok, %ToolResult{}} | {:error, error_map}. Args use string keys.

cancel(client, id, reason \\ nil)

@spec cancel(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  integer(),
  String.t() | nil
) :: :ok

Send notifications/cancelled for an in-flight request id.

complete(client, ref, arg_name, value)

@spec complete(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  tuple(),
  String.t(),
  String.t()
) :: {:ok, map()} | {:error, map()}

Request completion values. ref is {:prompt, name} or {:resource_template, uri_template}. Returns {:ok, %{values: _, total: _, has_more: _}}.

connect(server, opts \\ [])

@spec connect(
  module(),
  keyword()
) :: %Noizu.MCP.Test.Client{
  capabilities: term(),
  counter: term(),
  instructions: term(),
  ref: term(),
  server: term(),
  server_info: term(),
  session: term()
}

Start a session against server (a use Noizu.MCP.Server module) and run the initialize handshake. Starts the server's supervision tree if it is not already running.

Options: :client_info (map), :client_capabilities (wire-format map), :protocol_version.

deliver_raw(client, binary)

@spec deliver_raw(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  binary()
) :: :ok

Send a raw wire binary (escape hatch for malformed-input tests).

ensure_server_started(server)

@spec ensure_server_started(module()) :: :ok

Ensure server's supervision tree is running (shared across tests), detached from the calling process. Safe under concurrent async: true callers — waits until the tree is fully started before returning.

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

@spec get_prompt(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map(),
  keyword()
) :: {:ok, map()} | {:error, map()}

Get a prompt. Returns {:ok, %{description: _, messages: [%PromptMessage{}]}}. Args use string keys.

list_prompts(client, opts \\ [])

@spec list_prompts(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  keyword()
) :: {:ok, [Noizu.MCP.Types.Prompt.t()]} | {:error, map()}

List all prompts (auto-paginates). Returns {:ok, [%Prompt{}]}.

list_resource_templates(client, opts \\ [])

@spec list_resource_templates(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  keyword()
) :: {:ok, [Noizu.MCP.Types.ResourceTemplate.t()]} | {:error, map()}

List all resource templates (auto-paginates).

list_resources(client, opts \\ [])

@spec list_resources(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  keyword()
) :: {:ok, [Noizu.MCP.Types.Resource.t()]} | {:error, map()}

List all resources (auto-paginates). Returns {:ok, [%Resource{}]}.

list_tools(client, opts \\ [])

@spec list_tools(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  keyword()
) :: {:ok, [Noizu.MCP.Types.Tool.t()]} | {:error, map()}

List all tools (auto-paginates). Returns {:ok, [%Tool{}]} | {:error, error_map}.

notify(client, method, params \\ nil)

@spec notify(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map() | nil
) :: :ok

Send a notification to the server.

read_resource(client, uri, opts \\ [])

@spec read_resource(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  keyword()
) :: {:ok, [Noizu.MCP.Types.ResourceContents.t()]} | {:error, map()}

Read a resource. Returns {:ok, [%ResourceContents{}]} | {:error, error_map}.

refute_notification(client, method, opts \\ [])

@spec refute_notification(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  keyword()
) :: :ok

Assert no notification with method arrives within opts[:timeout] (default 100ms).

request(client, method, params \\ nil, opts \\ [])

@spec request(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map() | nil,
  keyword()
) :: {:ok, map()} | {:error, map()}

Send a request and await its result. Returns {:ok, result} | {:error, error_map}.

send_request(client, method, params \\ nil)

@spec send_request(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t(),
  map() | nil
) :: integer()

Send a request and return its id without awaiting the response.

set_log_level(client, level)

@spec set_log_level(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  atom() | String.t()
) :: {:ok, map()} | {:error, map()}

Set the MCP log level for this session.

subscribe(client, uri)

@spec subscribe(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t()
) :: {:ok, map()} | {:error, map()}

Subscribe to resource update notifications for uri.

unsubscribe(client, uri)

@spec unsubscribe(
  %Noizu.MCP.Test.Client{
    capabilities: term(),
    counter: term(),
    instructions: term(),
    ref: term(),
    server: term(),
    server_info: term(),
    session: term()
  },
  String.t()
) :: {:ok, map()} | {:error, map()}

Unsubscribe from resource update notifications for uri.