ExMCP.Helpers (ex_mcp v0.9.2)

View Source

Helper macros and functions for common MCP patterns in ExMCP.

This module provides convenient macros and utilities to reduce boilerplate and improve developer experience when working with MCP clients and servers.

Features

  • Connection management macros with automatic cleanup
  • Tool calling with pattern matching and error handling
  • Resource reading with automatic parsing
  • Batch operation helpers with timeout management
  • Testing utilities for MCP development

Usage

use ExMCP.Helpers

# Use helper macros
with_mcp_client "http://localhost:8080" do
  tools = list_tools!()
  result = call_tool!("calculator", %{op: "add", a: 1, b: 2})
end

Summary

Functions

Imports helper macros into the calling module.

Executes batch operations with error handling.

Calls a tool with error handling. Raises on failure.

Implements retry logic with exponential backoff.

Formats an error message for display.

Gets a prompt with error handling. Raises on failure.

Gets client status with error handling. Raises on failure.

Lists prompts with error handling. Raises on failure.

Lists resources with error handling. Raises on failure.

Lists tools with error handling. Raises on failure.

Measures execution time of an operation.

Reads a resource with error handling. Raises on failure.

Retries an operation with exponential backoff.

Safely executes an operation with timeout.

Validates tool arguments against schema.

Establishes an MCP connection and executes a block with automatic cleanup.

Creates a testing context with a mock MCP server.

Functions

__using__(opts)

(macro)

Imports helper macros into the calling module.

batch_execute!(operations, opts \\ [])

(macro)

Executes batch operations with error handling.

Examples

results = batch_execute!([
  {:call_tool, "greet", %{name: "Alice"}},
  {:call_tool, "greet", %{name: "Bob"}},
  {:list_resources, %{}}
])

call_tool!(tool_name, args \\ quote do %{} end, opts \\ [])

(macro)

Calls a tool with error handling. Raises on failure.

Examples

result = call_tool!("calculator", %{op: "add", a: 1, b: 2})

# With timeout
result = call_tool!("slow_tool", %{data: "..."}, timeout: 30_000)

do_retry(fun, opts \\ [])

@spec do_retry(
  (-> any()),
  keyword()
) :: any()

Implements retry logic with exponential backoff.

format_error_message(error)

@spec format_error_message(map() | any()) :: String.t()

Formats an error message for display.

get_prompt!(prompt_name, args \\ quote do %{} end, opts \\ [])

(macro)

Gets a prompt with error handling. Raises on failure.

get_status!()

(macro)

Gets client status with error handling. Raises on failure.

list_prompts!(opts \\ [])

(macro)

Lists prompts with error handling. Raises on failure.

list_resources!(opts \\ [])

(macro)

Lists resources with error handling. Raises on failure.

list_tools!(opts \\ [])

(macro)

Lists tools with error handling. Raises on failure.

Must be used within a with_mcp_client block or with an explicit client.

measure(list)

(macro)

Measures execution time of an operation.

Examples

{result, time_ms} = measure do
  call_tool!("slow_operation", %{})
end

IO.puts("Operation took #{time_ms}ms")

read_resource!(uri, opts \\ [])

(macro)

Reads a resource with error handling. Raises on failure.

Examples

content = read_resource!("file://data.txt")

# With JSON parsing
data = read_resource!("file://config.json", parse_json: true)

retry(opts \\ [], list)

(macro)

Retries an operation with exponential backoff.

Examples

result = retry max_attempts: 3, base_delay: 1000 do
  call_tool!("unreliable_tool", %{})
end

safe_execute(fun, timeout \\ 5000)

@spec safe_execute((-> any()), timeout()) :: {:ok, any()} | {:error, :timeout}

Safely executes an operation with timeout.

validate_tool_args(args, schema)

@spec validate_tool_args(map(), map()) :: :ok | {:error, String.t()}

Validates tool arguments against schema.

with_mcp_client(connection_spec, opts \\ [], list)

(macro)

Establishes an MCP connection and executes a block with automatic cleanup.

Examples

with_mcp_client "http://localhost:8080" do
  tools = list_tools!()
  IO.inspect(tools)
end

# With options
with_mcp_client "http://localhost:8080", timeout: 10_000 do
  result = call_tool!("slow_operation", %{})
end

# With fallback
with_mcp_client ["http://primary:8080", "http://backup:8080"] do
  status = get_status!()
end

with_mock_server(opts \\ [], list)

(macro)

Creates a testing context with a mock MCP server.

Examples

with_mock_server tools: [%{name: "test_tool"}] do
  tools = list_tools!()
  assert length(tools) == 1
end