mcp_elixir_sdk

defmodule MCP.Server do
  @moduledoc """
  MCP server implementation.

  A GenServer that handles incoming MCP client requests via a pluggable
  transport. Routes requests to a handler module implementing the
  `MCP.Server.Handler` behaviour.

  ## Usage

      defmodule MyHandler do
        @behaviour MCP.Server.Handler

        @impl true
        def init(_opts), do: {:ok, %{}}

        @impl true
        def handle_list_tools(_cursor, state) do
          tools = [%{"name" => "echo", "description" => "Echoes input"}]
          {:ok, tools, nil, state}
        end

        @impl true
        def handle_call_tool("echo", %{"message" => msg}, state) do
          {:ok, [%{"type" => "text", "text" => msg}], state}
        end
      end

      {:ok, server} = MCP.Server.start_link(
        transport: {MCP.Transport.Stdio, mode: :server},
        handler: {MyHandler, []},
        server_info: %{name: "my_server", version: "1.0.0"}
      )

  ## Options

    * `:transport` — `{module, opts}` transport spec. The server starts the
      transport in its init, setting itself as the owner.
    * `:handler` — `{module, opts}` handler spec. The module must implement
      `MCP.Server.Handler`.
    * `:server_info` — `%Implementation{}` or map with `:name` and `:version`.
    * `:instructions` — optional string instructions for the client.
    * `:request_timeout` — default timeout in ms for server-initiated
      requests (default: 30_000).
  """

  use GenServer

  require Logger

  alias MCP.Protocol

  alias MCP.Protocol.Capabilities.{
    CompletionCapabilities,
    LoggingCapabilities,
    PromptCapabilities,
    ResourceCapabilities,
    ServerCapabilities,
    ToolCapabilities
  }

  alias MCP.Protocol.Error
  alias MCP.Protocol.Messages.{Initialize, Notification, Request, Response}
  alias MCP.Protocol.Methods
  alias MCP.Protocol.Types.Implementation
  alias MCP.Server.ToolContext

  defstruct [
    :handler_module,
    :handler_state,
    :transport_module,
    :transport_pid,
    :client_capabilities,
    :client_info,
    :server_info,
    :capabilities,
    :instructions,
    :status,
    :pending_requests,
    :next_id,
    :request_timeout,
    :log_level,
    :has_async_tools,
    :async_tool_tasks
  ]

  @default_request_timeout 30_000

  # --- Public API ---

  @doc """
  Starts the server GenServer and its transport.
  """
  def start_link(opts) do
    {gen_opts, server_opts} = Keyword.split(opts, [:name])
    GenServer.start_link(__MODULE__, server_opts, gen_opts)
  end

  @doc """
  Returns the transport pid (useful for testing with MockTransport).
  """
  def transport(server) do
    GenServer.call(server, :get_transport)
  end

  @doc """
  Returns the current server status.
  """
  def status(server) do
    GenServer.call(server, :get_status)
  end

  @doc """
  Returns the client capabilities from initialization.
  """
  def client_capabilities(server) do
    GenServer.call(server, :get_client_capabilities)
  end

  @doc """
  Returns the client info from initialization.
  """
  def client_info(server) do
    GenServer.call(server, :get_client_info)
  end

  @doc """
  Sends a `notifications/tools/list_changed` notification to the client.
  """
  def notify_tools_changed(server) do
    GenServer.cast(server, {:notify, Methods.tools_list_changed(), nil})
  end

  @doc """
  Sends a `notifications/resources/list_changed` notification to the client.
  """
  def notify_resources_changed(server) do
    GenServer.cast(server, {:notify, Methods.resources_list_changed(), nil})
  end

  @doc """
  Sends a `notifications/resources/updated` notification for a specific URI.
  """
  def notify_resource_updated(server, uri) do
    GenServer.cast(server, {:notify, Methods.resources_updated(), %{"uri" => uri}})
  end

  @doc """
  Sends a `notifications/prompts/list_changed` notification to the client.
  """
  def notify_prompts_changed(server) do
    GenServer.cast(server, {:notify, Methods.prompts_list_changed(), nil})
  end

  @doc """
  Sends a log message notification to the client.

  Respects the log level set by the client via `logging/setLevel`.
  Messages below the current level are silently dropped.
  """
  def log(server, level, data, logger_name \\ nil) do
    GenServer.cast(server, {:log, level, data, logger_name})
  end

  @doc """
  Sends a progress notification to the client.
  """
  def send_progress(server, progress_token, progress, total \\ nil) do
    params = %{"progressToken" => progress_token, "progress" => progress}
    params = if total, do: Map.put(params, "total", total), else: params
    GenServer.cast(server, {:notify, Methods.progress(), params})
  end

  @doc """
  Sends a `sampling/createMessage` request to the client.

  Returns `{:ok, result}` or `{:error, reason}`.
  Requires the client to have declared sampling capability.
  """
  def request_sampling(server, params, timeout \\ 60_000) do
    GenServer.call(server, {:request_client, Methods.sampling_create_message(), params}, timeout)
  end

  @doc """
  Sends a `roots/list` request to the client.

  Returns `{:ok, result}` or `{:error, reason}`.
  Requires the client to have declared roots capability.
  """
  def request_roots(server, timeout \\ 30_000) do
    GenServer.call(server, {:request_client, Methods.roots_list(), %{}}, timeout)
  end

  @doc """
  Sends an `elicitation/create` request to the client.

  Returns `{:ok, result}` or `{:error, reason}`.
  Requires the client to have declared elicitation capability.
  """
  def request_elicitation(server, params, timeout \\ 60_000) do
    GenServer.call(server, {:request_client, Methods.elicitation_create(), params}, timeout)
  end

  @doc """
  Closes the server and its transport.
  """
  def close(server) do
    GenServer.call(server, :close)
  catch
    :exit, _ -> :ok
  end

  # --- GenServer callbacks ---

  @impl GenServer
  def init(opts) do
    {transport_spec, opts} = Keyword.pop!(opts, :transport)
    {handler_spec, opts} = Keyword.pop!(opts, :handler)

    server_info =
      build_server_info(Keyword.get(opts, :server_info, %{name: "mcp_elixir_sdk", version: "1.0.0"}))

    instructions = Keyword.get(opts, :instructions)
    request_timeout = Keyword.get(opts, :request_timeout, @default_request_timeout)

    {handler_module, handler_opts} = handler_spec
    capabilities = detect_capabilities(handler_module)
    has_async_tools = has_async_tool_handler?(handler_module)

    case handler_module.init(handler_opts) do
      {:ok, handler_state} ->
        case start_transport(transport_spec) do
          {:ok, module, pid} ->
            state = %__MODULE__{
              handler_module: handler_module,
              handler_state: handler_state,
              transport_module: module,
              transport_pid: pid,
              server_info: server_info,
              capabilities: capabilities,
              instructions: instructions,
              status: :waiting,
              pending_requests: %{},
              next_id: 1,
              request_timeout: request_timeout,
              has_async_tools: has_async_tools,
              async_tool_tasks: %{}
            }

            {:ok, state}

          {:error, reason} ->
            {:stop, reason}
        end

      {:error, reason} ->
        {:stop, {:handler_init_failed, reason}}
    end
  end

  @impl GenServer
  def handle_call(:get_transport, _from, state) do
    {:reply, state.transport_pid, state}
  end

  def handle_call(:get_status, _from, state) do
    {:reply, state.status, state}
  end

  def handle_call(:get_client_capabilities, _from, state) do
    {:reply, state.client_capabilities, state}
  end

  def handle_call(:get_client_info, _from, state) do
    {:reply, state.client_info, state}
  end

  def handle_call(:close, _from, state) do
    do_close(state)
  end

  def handle_call({:request_client, method, params}, from, %{status: :ready} = state) do
    {id, state} = next_id(state)
    send_request(state, id, method, params)
    timeout_ref = schedule_timeout(id, state.request_timeout)
    state = put_pending(state, id, from, timeout_ref)
    {:noreply, state}
  end

  def handle_call({:request_client, _method, _params}, _from, %{status: status} = state) do
    {:reply, {:error, {:not_ready, status}}, state}
  end

  # Context calls from async tool handlers
  def handle_call({:context_notify, related_request_id, method, params}, _from, state) do
    send_notification(state, method, params, related_request_id: related_request_id)
    {:reply, :ok, state}
  end

  def handle_call({:context_request, related_request_id, method, params}, from, state) do
    {id, state} = next_id(state)
    send_request(state, id, method, params, related_request_id: related_request_id)
    timeout_ref = schedule_timeout(id, state.request_timeout)
    state = put_pending(state, id, from, timeout_ref)
    {:noreply, state}
  end

  @impl GenServer
  def handle_cast({:notify, method, params}, %{status: :ready} = state) do
    send_notification(state, method, params)
    {:noreply, state}
  end

  def handle_cast({:notify, _method, _params}, state) do
    {:noreply, state}
  end

  def handle_cast({:log, level, data, logger_name}, %{status: :ready} = state) do
    if should_log?(level, state.log_level) do
      params = %{"level" => level, "data" => data}
      params = if logger_name, do: Map.put(params, "logger", logger_name), else: params
      send_notification(state, Methods.logging_message(), params)
    end

    {:noreply, state}
  end

  def handle_cast({:log, _level, _data, _logger_name}, state) do
    {:noreply, state}
  end

  # --- Incoming messages from transport ---

  @impl GenServer
  def handle_info({:mcp_message, message}, state) do
    case Protocol.decode_message(message) do
      {:ok, %Request{} = request} ->
        handle_client_request(request, state)

      {:ok, %Notification{} = notification} ->
        handle_client_notification(notification, state)

      {:ok, %Response{} = response} ->
        handle_client_response(response, state)

      {:error, error} ->
        Logger.warning("MCP Server: failed to decode message: #{inspect(error)}")
        {:noreply, state}
    end
  end

  def handle_info({:mcp_transport_closed, reason}, state) do
    Logger.debug("MCP Server: transport closed: #{inspect(reason)}")

    Enum.each(state.pending_requests, fn {_id, {from, timeout_ref}} ->
      cancel_timeout(timeout_ref)
      GenServer.reply(from, {:error, {:transport_closed, reason}})
    end)

    {:noreply, %{state | status: :closed, pending_requests: %{}}}
  end

  def handle_info({:request_timeout, id}, state) do
    case Map.pop(state.pending_requests, id) do
      {{from, _timeout_ref}, pending} ->
        GenServer.reply(from, {:error, :timeout})
        {:noreply, %{state | pending_requests: pending}}

      {nil, _} ->
        {:noreply, state}
    end
  end

  # Async tool task completion (Task.async sends {ref, result})
  def handle_info({ref, result}, state) when is_reference(ref) do
    case find_async_task(state, ref) do
      {request_id, _task_pid} ->
        Process.demonitor(ref, [:flush])
        state = remove_async_task(state, request_id)
        handle_async_tool_result(request_id, result, state)

      nil ->
        Logger.debug("MCP Server: unexpected ref message: #{inspect(ref)}")
        {:noreply, state}
    end
  end

  # Task DOWN message (process exited abnormally)
  def handle_info({:DOWN, ref, :process, _pid, reason}, state) do
    case find_async_task(state, ref) do
      {request_id, _task_pid} ->
        state = remove_async_task(state, request_id)
        send_error_response(state, request_id, %Error{code: -32_603, message: "Tool execution failed: #{inspect(reason)}"})
        {:noreply, state}

      nil ->
        {:noreply, state}
    end
  end

  def handle_info(msg, state) do
    Logger.debug("MCP Server: unexpected message: #{inspect(msg)}")
    {:noreply, state}
  end

  @impl GenServer
  def terminate(_reason, state) do
    if state.transport_pid && state.status != :closed do
      state.transport_module.close(state.transport_pid)
    end
  catch
    _, _ -> :ok
  end

  # --- Request handling ---

  defp handle_client_request(%Request{id: id, method: "initialize", params: params}, state) do
    handle_initialize(id, params, state)
  end

  defp handle_client_request(%Request{id: id, method: "ping"}, state) do
    send_success_response(state, id, %{})
    {:noreply, state}
  end

  # All other requests require :ready status
  defp handle_client_request(%Request{id: id} = request, %{status: :ready} = state) do
    route_request(request, id, state)
  end

  defp handle_client_request(%Request{id: id}, state) do
    send_error_response(state, id, Error.invalid_request("Server not initialized"))
    {:noreply, state}
  end

  # --- Notification handling ---

  defp handle_client_notification(
         %Notification{method: "notifications/initialized"},
         %{status: :waiting} = state
       ) do
    {:noreply, %{state | status: :ready}}
  end

  defp handle_client_notification(
         %Notification{method: "notifications/cancelled", params: params},
         state
       ) do
    Logger.debug("MCP Server: received cancellation for request #{inspect(params)}")
    {:noreply, state}
  end

  defp handle_client_notification(%Notification{method: method}, state) do
    Logger.debug("MCP Server: unhandled notification: #{method}")
    {:noreply, state}
  end

  # --- Response handling (for server-initiated requests) ---

  defp handle_client_response(%Response{id: id} = response, state) do
    case Map.pop(state.pending_requests, id) do
      {{from, timeout_ref}, pending} ->
        cancel_timeout(timeout_ref)
        state = %{state | pending_requests: pending}

        reply =
          if response.error do
            {:error, response.error}
          else
            {:ok, response.result}
          end

        GenServer.reply(from, reply)
        {:noreply, state}

      {nil, _} ->
        Logger.warning("MCP Server: received response for unknown request id=#{inspect(id)}")
        {:noreply, state}
    end
  end

  # --- Initialization ---

  defp handle_initialize(id, params, %{status: :waiting} = state) do
    init_params = Initialize.Params.from_map(params)

    result =
      Initialize.Result.to_map(%Initialize.Result{
        protocol_version: negotiate_version(init_params.protocol_version),
        capabilities: state.capabilities,
        server_info: state.server_info,
        instructions: state.instructions
      })

    send_success_response(state, id, result)

    state = %{
      state
      | client_capabilities: init_params.capabilities,
        client_info: init_params.client_info
    }

    {:noreply, state}
  end

  defp handle_initialize(id, _params, %{status: :ready} = state) do
    send_error_response(state, id, Error.invalid_request("Already initialized"))
    {:noreply, state}
  end

  defp handle_initialize(id, _params, state) do
    send_error_response(state, id, Error.invalid_request("Invalid server state"))
    {:noreply, state}
  end

  # --- Request routing ---

  defp route_request(%Request{method: "tools/list", params: params}, id, state),
    do: handle_tools_list(id, params, state)

  defp route_request(%Request{method: "tools/call", params: params}, id, state),
    do: handle_tools_call(id, params, state)

  defp route_request(%Request{method: "resources/list", params: params}, id, state),
    do: handle_resources_list(id, params, state)

  defp route_request(%Request{method: "resources/read", params: params}, id, state),
    do: handle_resources_read(id, params, state)

  defp route_request(%Request{method: "resources/subscribe", params: params}, id, state),
    do: handle_resources_subscribe(id, params, state)

  defp route_request(%Request{method: "resources/unsubscribe", params: params}, id, state),
    do: handle_resources_unsubscribe(id, params, state)

  defp route_request(%Request{method: "resources/templates/list", params: params}, id, state),
    do: handle_resources_templates_list(id, params, state)

  defp route_request(%Request{method: "prompts/list", params: params}, id, state),
    do: handle_prompts_list(id, params, state)

  defp route_request(%Request{method: "prompts/get", params: params}, id, state),
    do: handle_prompts_get(id, params, state)

  defp route_request(%Request{method: "completion/complete", params: params}, id, state),
    do: handle_completion(id, params, state)

  defp route_request(%Request{method: "logging/setLevel", params: params}, id, state),
    do: handle_set_log_level(id, params, state)

  defp route_request(%Request{method: method}, id, state) do
    send_error_response(state, id, Error.method_not_found(method))
    {:noreply, state}
  end

  defp handle_tools_list(id, params, state) do
    cursor = get_in_params(params, "cursor")

    case state.handler_module.handle_list_tools(cursor, state.handler_state) do
      {:ok, tools, next_cursor, handler_state} ->
        result = %{"tools" => tools}
        result = if next_cursor, do: Map.put(result, "nextCursor", next_cursor), else: result
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_tools_call(id, params, state) do
    name = Map.get(params || %{}, "name", "")
    arguments = Map.get(params || %{}, "arguments", %{})
    meta = Map.get(params || %{}, "_meta")

    if state.has_async_tools do
      handle_tools_call_async(id, name, arguments, meta, state)
    else
      handle_tools_call_sync(id, name, arguments, state)
    end
  end

  defp handle_tools_call_sync(id, name, arguments, state) do
    case state.handler_module.handle_call_tool(name, arguments, state.handler_state) do
      {:ok, content, handler_state} ->
        send_success_response(state, id, %{"content" => content})
        {:noreply, %{state | handler_state: handler_state}}

      {:ok, content, is_error, handler_state} ->
        result = %{"content" => content}
        result = if is_error, do: Map.put(result, "isError", true), else: result
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_tools_call_async(id, name, arguments, meta, state) do
    server_pid = self()
    handler_module = state.handler_module
    handler_state = state.handler_state

    context = %ToolContext{
      server_pid: server_pid,
      request_id: id,
      meta: meta
    }

    task =
      Task.async(fn ->
        handler_module.handle_call_tool(name, arguments, context, handler_state)
      end)

    async_tasks = Map.put(state.async_tool_tasks, id, {task.ref, task.pid})
    {:noreply, %{state | async_tool_tasks: async_tasks}}
  end

  defp handle_resources_list(id, params, state) do
    cursor = get_in_params(params, "cursor")

    case state.handler_module.handle_list_resources(cursor, state.handler_state) do
      {:ok, resources, next_cursor, handler_state} ->
        result = %{"resources" => resources}
        result = if next_cursor, do: Map.put(result, "nextCursor", next_cursor), else: result
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_resources_read(id, params, state) do
    uri = Map.get(params || %{}, "uri", "")

    case state.handler_module.handle_read_resource(uri, state.handler_state) do
      {:ok, contents, handler_state} ->
        send_success_response(state, id, %{"contents" => contents})
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_resources_subscribe(id, params, state) do
    uri = Map.get(params || %{}, "uri", "")

    case state.handler_module.handle_subscribe(uri, state.handler_state) do
      {:ok, handler_state} ->
        send_success_response(state, id, %{})
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_resources_unsubscribe(id, params, state) do
    uri = Map.get(params || %{}, "uri", "")

    case state.handler_module.handle_unsubscribe(uri, state.handler_state) do
      {:ok, handler_state} ->
        send_success_response(state, id, %{})
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_resources_templates_list(id, params, state) do
    cursor = get_in_params(params, "cursor")

    case state.handler_module.handle_list_resource_templates(cursor, state.handler_state) do
      {:ok, templates, next_cursor, handler_state} ->
        result = %{"resourceTemplates" => templates}
        result = if next_cursor, do: Map.put(result, "nextCursor", next_cursor), else: result
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_prompts_list(id, params, state) do
    cursor = get_in_params(params, "cursor")

    case state.handler_module.handle_list_prompts(cursor, state.handler_state) do
      {:ok, prompts, next_cursor, handler_state} ->
        result = %{"prompts" => prompts}
        result = if next_cursor, do: Map.put(result, "nextCursor", next_cursor), else: result
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_prompts_get(id, params, state) do
    name = Map.get(params || %{}, "name", "")
    arguments = Map.get(params || %{}, "arguments")

    case state.handler_module.handle_get_prompt(name, arguments, state.handler_state) do
      {:ok, result, handler_state} ->
        send_success_response(state, id, result)
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_completion(id, params, state) do
    ref = Map.get(params || %{}, "ref", %{})
    argument = Map.get(params || %{}, "argument", %{})

    case state.handler_module.handle_complete(ref, argument, state.handler_state) do
      {:ok, completion, handler_state} ->
        send_success_response(state, id, %{"completion" => completion})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp handle_set_log_level(id, params, state) do
    level = Map.get(params || %{}, "level", "info")

    case state.handler_module.handle_set_log_level(level, state.handler_state) do
      {:ok, handler_state} ->
        send_success_response(state, id, %{})
        {:noreply, %{state | handler_state: handler_state, log_level: level}}
    end
  end

  # --- Private helpers ---

  defp start_transport({module, opts}) do
    case module.start_link([{:owner, self()} | opts]) do
      {:ok, pid} -> {:ok, module, pid}
      {:error, reason} -> {:error, reason}
    end
  end

  defp build_server_info(%Implementation{} = impl), do: impl

  defp build_server_info(map) when is_map(map) do
    %Implementation{
      name: Map.get(map, :name) || Map.get(map, "name", "mcp_elixir_sdk"),
      version: Map.get(map, :version) || Map.get(map, "version", "1.0.0")
    }
  end

  defp detect_capabilities(handler_module) do
    callbacks = handler_module.__info__(:functions)

    %ServerCapabilities{
      tools: if({:handle_list_tools, 2} in callbacks, do: %ToolCapabilities{list_changed: true}),
      resources: detect_resource_capabilities(callbacks),
      prompts:
        if({:handle_list_prompts, 2} in callbacks, do: %PromptCapabilities{list_changed: true}),
      logging: if({:handle_set_log_level, 2} in callbacks, do: %LoggingCapabilities{}),
      completions: if({:handle_complete, 3} in callbacks, do: %CompletionCapabilities{})
    }
  end

  defp detect_resource_capabilities(callbacks) do
    has_resources = {:handle_list_resources, 2} in callbacks
    has_subscribe = {:handle_subscribe, 2} in callbacks

    if has_resources do
      %ResourceCapabilities{
        list_changed: true,
        subscribe: if(has_subscribe, do: true)
      }
    end
  end

  defp negotiate_version(client_version) do
    supported = Protocol.protocol_version()

    if client_version == supported do
      supported
    else
      supported
    end
  end

  defp next_id(state) do
    {state.next_id, %{state | next_id: state.next_id + 1}}
  end

  defp send_request(state, id, method, params, opts \\ []) do
    message = Request.new(id, method, params)
    encoded = Jason.decode!(Jason.encode!(message))
    send_message_to_transport(state, encoded, opts)
  end

  defp send_notification(state, method, params, opts \\ []) do
    message = Notification.new(method, params)
    encoded = Jason.decode!(Jason.encode!(message))
    send_message_to_transport(state, encoded, opts)
  end

  defp send_success_response(state, id, result, opts \\ []) do
    response = Response.success(id, result)
    encoded = Jason.decode!(Jason.encode!(response))
    send_message_to_transport(state, encoded, opts)
  end

  defp send_error_response(state, id, %Error{} = error, opts \\ []) do
    response = Response.error(id, error)
    encoded = Jason.decode!(Jason.encode!(response))
    send_message_to_transport(state, encoded, opts)
  end

  defp send_message_to_transport(state, message, opts) do
    if opts != [] && function_exported?(state.transport_module, :send_message, 3) do
      state.transport_module.send_message(state.transport_pid, message, opts)
    else
      state.transport_module.send_message(state.transport_pid, message)
    end
  end

  defp put_pending(state, id, from, timeout_ref) do
    %{state | pending_requests: Map.put(state.pending_requests, id, {from, timeout_ref})}
  end

  defp schedule_timeout(id, timeout_ms) do
    Process.send_after(self(), {:request_timeout, id}, timeout_ms)
  end

  defp cancel_timeout(ref) do
    Process.cancel_timer(ref)
  end

  defp get_in_params(nil, _key), do: nil
  defp get_in_params(params, key), do: Map.get(params, key)

  @log_levels ~w(debug info notice warning error critical alert emergency)

  defp should_log?(_level, nil), do: false

  defp should_log?(level, threshold) do
    level_index = Enum.find_index(@log_levels, &(&1 == level)) || 0
    threshold_index = Enum.find_index(@log_levels, &(&1 == threshold)) || 0
    level_index >= threshold_index
  end

  defp has_async_tool_handler?(handler_module) do
    {:handle_call_tool, 4} in handler_module.__info__(:functions)
  end

  defp find_async_task(state, ref) do
    Enum.find_value(state.async_tool_tasks, fn {request_id, {task_ref, task_pid}} ->
      if task_ref == ref, do: {request_id, task_pid}
    end)
  end

  defp remove_async_task(state, request_id) do
    %{state | async_tool_tasks: Map.delete(state.async_tool_tasks, request_id)}
  end

  defp handle_async_tool_result(request_id, result, state) do
    case result do
      {:ok, content, handler_state} ->
        send_success_response(state, request_id, %{"content" => content})
        {:noreply, %{state | handler_state: handler_state}}

      {:ok, content, is_error, handler_state} ->
        result_map = %{"content" => content}
        result_map = if is_error, do: Map.put(result_map, "isError", true), else: result_map
        send_success_response(state, request_id, result_map)
        {:noreply, %{state | handler_state: handler_state}}

      {:error, code, message, handler_state} ->
        send_error_response(state, request_id, %Error{code: code, message: message})
        {:noreply, %{state | handler_state: handler_state}}
    end
  end

  defp do_close(state) do
    if state.transport_pid do
      state.transport_module.close(state.transport_pid)
    end

    {:stop, :normal, :ok, %{state | status: :closed}}
  catch
    _, _ -> {:stop, :normal, :ok, %{state | status: :closed}}
  end
end