MCPEx.Transport.Test (MCPEx v0.1.0)

Test implementation of the MCP transport for integration testing.

This transport operates fully in memory and connects to an MCPEx.Transport.TestServer instance to simulate the server side of the protocol.

Features

  • In-memory communication with no external dependencies
  • Controllable message flow and timing
  • Observable message exchange for testing
  • Configurable error scenarios
  • Complete protocol simulation

Usage 

# Create a test server
server = MCPEx.Transport.TestServer.new(
  capabilities: [:resources, :tools]
)

# Create a test transport connected to this server
{:ok, transport} = MCPEx.Transport.Test.start_link(server: server)

# Register a message receiver (optional)
MCPEx.Transport.Test.register_message_receiver(transport, self())

# Send a message and get response
{:ok, response} = MCPEx.Transport.send(transport, json_encoded_message)

Summary

Types

t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

connected?(transport)

Checks if the transport is connected to the server.

get_log(transport)

Gets the current message log for debugging/testing purposes.

new(options)

Creates a new test transport instance without starting a process.

record_session(fun)

Records a session of message exchanges for later replay.

register_message_receiver(transport, receiver)

Registers a process to receive message notifications.

replay_session(recording)

Replays a previously recorded session.

send_message(transport, message)

Sends a message to the server and waits for a response.

start_link(options)

Starts a new test transport as a linked process.

Types

t()

@type t() :: %MCPEx.Transport.Test{
  client_ref: reference() | nil,
  config: map(),
  connected: boolean(),
  log: list(),
  message_receiver: pid() | nil,
  next_id: integer(),
  pending_requests: map(),
  server: MCPEx.Transport.TestServer.t() | nil
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

connected?(transport)

@spec connected?(pid()) :: boolean()

Checks if the transport is connected to the server.

Parameters

  • transport - The transport process

Returns

  • true or false

get_log(transport)

@spec get_log(pid()) :: list()

Gets the current message log for debugging/testing purposes.

Parameters

  • transport - The transport process

Returns

  • List of log entries

new(options)

@spec new(keyword()) :: t()

Creates a new test transport instance without starting a process.

Options

  • :server - The MCPEx.Transport.TestServer instance to connect to (required)
  • :config - Additional configuration options

Returns

  • A new TestTransport struct

record_session(fun)

@spec record_session((-> any())) :: {:ok, term()} | {:error, term()}

Records a session of message exchanges for later replay.

Parameters

  • fun - Function to execute during recording

Returns

  • {:ok, recording} - The recorded session
  • {:error, reason} - Failed to record the session

register_message_receiver(transport, receiver)

@spec register_message_receiver(pid(), pid()) :: :ok

Registers a process to receive message notifications.

The registered process will receive {:message, message} tuples for all messages received from the server.

Parameters

  • transport - The transport process
  • receiver - The process to receive notifications

Returns

  • :ok

replay_session(recording)

@spec replay_session(term()) :: {:ok, pid()} | {:error, term()}

Replays a previously recorded session.

Parameters

  • recording - Recording from record_session

Returns

  • {:ok, transport} - Transport for the replayed session
  • {:error, reason} - Failed to replay the session

send_message(transport, message)

@spec send_message(pid(), map() | String.t()) :: {:ok, map()} | {:error, term()}

Sends a message to the server and waits for a response.

Parameters

  • transport - The transport process
  • message - The message to send, either as a map or JSON string

Returns

  • {:ok, response} - The response from the server
  • {:error, reason} - Failed to send the message or receive a response

start_link(options)

@spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}

Starts a new test transport as a linked process.

Options

  • :server - The MCPEx.Transport.TestServer instance to connect to (required)
  • :auto_connect - Whether to automatically connect to the server (default: true)
  • :config - Additional configuration options

Returns

  • {:ok, pid} - The transport was started successfully
  • {:error, reason} - Failed to start the transport