ExMCP.Authorization.TokenManager (ex_mcp v0.9.0)

View Source

Manages OAuth tokens with automatic refresh and expiration tracking.

This GenServer maintains tokens, monitors their expiration, and automatically refreshes them before they expire, following MCP authorization best practices.

Features

  • Automatic token refresh before expiration
  • Thread-safe token storage
  • Configurable refresh window
  • Token rotation support

Summary

Types

auth_method()
t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_token(manager)

Gets the current valid access token.

get_token_info(manager)

Gets full token information including metadata.

refresh_now(manager)

Forces an immediate token refresh.

set_token(manager, token_response)

Sets a new token in the manager.

start_link(opts)

Starts a token manager process.

subscribe(manager)

Subscribes to token update notifications.

unsubscribe(manager)

Unsubscribes from token update notifications.

upgrade_scopes(manager, additional_scopes, opts \\ [])

Upgrades the token's scopes by merging additional scopes with the current ones.

Types

auth_method()

@type auth_method() :: :client_secret | :private_key_jwt | :enterprise_idjag

t()

@type t() :: %ExMCP.Authorization.TokenManager{
  access_token: String.t() | nil,
  auth_config: map(),
  auth_method: auth_method() | nil,
  expires_at: DateTime.t() | nil,
  refresh_timer: reference() | nil,
  refresh_token: String.t() | nil,
  scope: String.t() | nil,
  subscribers: MapSet.t(pid()),
  token_type: String.t()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_token(manager)

@spec get_token(GenServer.server()) :: {:ok, String.t()} | {:error, atom()}

Gets the current valid access token.

Returns {:ok, token} or {:error, reason} if token is expired/missing.

get_token_info(manager)

@spec get_token_info(GenServer.server()) :: {:ok, map()} | {:error, atom()}

Gets full token information including metadata.

refresh_now(manager)

@spec refresh_now(GenServer.server()) :: {:ok, map()} | {:error, any()}

Forces an immediate token refresh.

set_token(manager, token_response)

@spec set_token(GenServer.server(), map()) :: :ok

Sets a new token in the manager.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a token manager process.

Options:

  • :auth_config - Authorization configuration (client_id, client_secret, token_endpoint, etc.)
  • :refresh_window - Seconds before expiration to refresh (default: 300)
  • :name - Optional process name

subscribe(manager)

@spec subscribe(GenServer.server()) :: :ok

Subscribes to token update notifications.

Subscribers receive {:token_updated, manager, token_info} messages.

unsubscribe(manager)

@spec unsubscribe(GenServer.server()) :: :ok

Unsubscribes from token update notifications.

upgrade_scopes(manager, additional_scopes, opts \\ [])

@spec upgrade_scopes(GenServer.server(), [String.t()], keyword()) ::
  {:ok, map()} | {:error, atom() | {atom(), [String.t()]}}

Upgrades the token's scopes by merging additional scopes with the current ones.

If a refresh token is available, attempts to refresh with the expanded scope set. If no refresh token is available, returns {:error, :reauthorization_required} with the combined scope list so the caller can initiate a full re-authorization.

Options

  • :timeout - Call timeout in milliseconds (default: 30_000)