# `ReqLLM.Provider`
[🔗](https://github.com/agentjido/req_llm/blob/main/lib/req_llm/provider.ex#L1)

Behavior for LLM provider implementations.

Providers implement this behavior to handle model-specific request configuration,
body encoding, response parsing, and usage extraction. Each provider is a Req plugin
that uses the standard Req request/response pipeline.

## Provider Responsibilities

- **Request Preparation**: Configure operation-specific requests via `prepare_request/4`
- **Request Configuration**: Set headers, base URLs, authentication via `attach/3`
- **Body Encoding**: Transform Context to provider-specific JSON via `encode_body/1`
- **Body Construction**: Build request body maps via `build_body/1` (optional)
- **Response Parsing**: Decode API responses via `decode_response/1`
- **Usage Extraction**: Parse usage/cost data via `extract_usage/2` (optional)
- **Streaming Configuration**: Build HTTP requests via `attach_stream/4` or provide
  canonical in-process streams via `attach_in_process_stream/3`

## Implementation Pattern

Providers use `use ReqLLM.Provider` to define their configuration and implement
the required callbacks as Req pipeline steps.

## Examples

    defmodule MyProvider do
      use ReqLLM.Provider,
        id: :myprovider,
        default_base_url: "https://api.example.com/v1",
        default_env_key: "MYPROVIDER_API_KEY"

      @impl ReqLLM.Provider
      def prepare_request(operation, model, messages, opts) do
        with {:ok, request} <- Req.new(base_url: base_url()),
             request <- add_auth_headers(request),
             request <- add_operation_specific_config(request, operation) do
          {:ok, request}
        end
      end

      @impl ReqLLM.Provider
      def attach(request, model, opts) do
        request
        |> add_auth_headers()
        |> Req.Request.append_request_steps(llm_encode_body: &encode_body/1)
        |> Req.Request.append_response_steps(llm_decode_response: &decode_response/1)
      end

      @impl ReqLLM.Provider
      def attach_stream(model, context, opts, _finch_name) do
        operation = Keyword.get(opts, :operation, :chat)
        processed_opts = ReqLLM.Provider.Options.process!(__MODULE__, operation, model, Keyword.merge(opts, stream: true, context: context))
        
        with {:ok, req} <- prepare_request(operation, model, context, processed_opts),
             req <- attach(req, model, processed_opts),
             {req, _resp} <- encode_body(req) do
          url = URI.to_string(req.url)
          headers = req.headers |> Enum.map(fn {k, [v | _]} -> {k, v} end)
          body = req.body
          
          finch_request = Finch.build(:post, url, headers, body)
          {:ok, finch_request}
        end
      end

      def encode_body(request) do
        # Transform request.options[:context] to provider JSON
      end

      def decode_response({req, resp}) do
        # Parse response body and return {req, updated_resp}
      end
    end

# `operation`

```elixir
@type operation() :: :chat | :embed | :moderate | atom()
```

# `attach`

```elixir
@callback attach(Req.Request.t(), LLMDB.Model.t(), keyword()) :: Req.Request.t()
```

Attaches provider-specific configuration to a Req request.

This callback configures the request for the specific provider by setting up
authentication, base URLs, and registering request/response pipeline steps.

## Parameters

  * `request` - The Req.Request struct to configure
  * `model` - The ReqLLM.Model struct with model specification
  * `opts` - Additional options (messages, tools, streaming, etc.)

## Returns

  * `Req.Request.t()` - The configured request with pipeline steps attached

# `attach_in_process_stream`
*optional* 

```elixir
@callback attach_in_process_stream(
  LLMDB.Model.t(),
  ReqLLM.Context.t(),
  keyword()
) ::
  {:ok, Enumerable.t() | ReqLLM.Provider.InProcessStream.t()} | {:error, term()}
```

Builds a provider-owned stream of canonical chunks.

This callback is used when `stream_transport/2` returns `:in_process`. The
returned enumerable must emit `ReqLLM.StreamChunk` structs. It can emit
`{:error, reason}` to fail the stream.

Return an enumerable directly when task termination is enough for cleanup.
Return `ReqLLM.Provider.InProcessStream` when the provider also needs an
explicit callback for cancellation or timeouts.

## Examples

    def stream_transport(_model, _opts), do: :in_process

    def attach_in_process_stream(model, context, opts) do
      {:ok, canonical_chunk_stream(model, context, opts)}
    end

    def attach_in_process_stream(model, context, opts) do
      subscription = subscribe(model, context, opts)

      {:ok,
       ReqLLM.Provider.InProcessStream.new(subscription,
         cancel: fn -> cancel_subscription(subscription) end
       )}
    end

# `attach_stream`
*optional* 

```elixir
@callback attach_stream(
  LLMDB.Model.t(),
  ReqLLM.Context.t(),
  keyword(),
  atom()
) :: {:ok, Finch.Request.t()} | {:error, Exception.t()}
```

Build complete Finch request for streaming operations.

This callback creates a complete Finch.Request struct for streaming operations,
allowing providers to specify their streaming endpoint, headers, and request body
format. This consolidates streaming request preparation into a single callback.

## Parameters

  * `model` - The ReqLLM.Model struct
  * `context` - The Context with messages to stream
  * `opts` - Additional options (temperature, max_tokens, etc.)
  * `finch_name` - Finch process name for connection pooling

## Returns

  * `{:ok, Finch.Request.t()}` - Successfully built streaming request
  * `{:error, Exception.t()}` - Request building error

## Examples

    def attach_stream(model, context, opts, _finch_name) do
      url = "https://api.openai.com/v1/chat/completions"
      api_key = ReqLLM.Keys.get!(model, opts)
      headers = [
        {"Authorization", "Bearer " <> api_key},
        {"Content-Type", "application/json"}
      ]
      
      body = Jason.encode!(%{
        model: model.id,
        messages: encode_messages(context.messages),
        stream: true
      })
      
      request = Finch.build(:post, url, headers, body)
      {:ok, request}
    end

    # Anthropic with different endpoint and headers
    def attach_stream(model, context, opts, _finch_name) do
      url = "https://api.anthropic.com/v1/messages"
      api_key = ReqLLM.Keys.get!(model, opts)
      headers = [
        {"Authorization", "Bearer " <> api_key},
        {"Content-Type", "application/json"},
        {"anthropic-version", "2023-06-01"}
      ]
      
      body = Jason.encode!(%{
        model: model.id,
        messages: encode_anthropic_messages(context),
        stream: true
      })
      
      request = Finch.build(:post, url, headers, body)
      {:ok, request}
    end

# `build_body`
*optional* 

```elixir
@callback build_body(Req.Request.t()) :: map()
```

# `credential_missing?`
*optional* 

```elixir
@callback credential_missing?(Exception.t()) :: boolean()
```

Checks if an exception indicates missing credentials for this provider.

This optional callback allows providers to identify credential-related errors
so the fixture system can fall back to existing fixtures during recording when
credentials are unavailable.

## Parameters

  * `exception` - The exception to check

## Returns

  * `true` - Exception indicates missing credentials
  * `false` - Exception is not credential-related

## Examples

    # Google provider checking for missing API key
    def credential_missing?(%ReqLLM.Error.Invalid.Parameter{parameter: param}) do
      String.contains?(param, "api_key") and
        String.contains?(param, "GOOGLE_API_KEY")
    end
    def credential_missing?(_), do: false

# `decode_response`

```elixir
@callback decode_response({Req.Request.t(), Req.Response.t()}) ::
  {Req.Request.t(), Req.Response.t() | Exception.t()}
```

Decodes provider API response.

This callback is typically used as a Req response step that transforms the
raw API response into a standardized format for ReqLLM consumption.

## Parameters

  * `request_response` - Tuple of {Req.Request.t(), Req.Response.t()}

## Returns

  * `{Req.Request.t(), Req.Response.t() | Exception.t()}` - Decoded response or error

# `decode_stream_event`
*optional* 

```elixir
@callback decode_stream_event(map(), LLMDB.Model.t()) :: [ReqLLM.StreamChunk.t()]
```

Decode provider streaming event to list of StreamChunk structs for streaming responses.

This is called by ReqLLM.StreamServer during real-time streaming to convert
provider-specific streaming events into canonical StreamChunk structures. For terminal
events (like "[DONE]"), providers should return metadata chunks with usage
information and finish reasons.

Different providers use different streaming protocols:
- **OpenAI, Anthropic, Google**: Server-Sent Events (SSE) - `event` is typically `%{data: ...}`
- **AWS Bedrock**: AWS EventStream (binary) - `event` is the decoded JSON payload

## Parameters

  * `event` - The streaming event data (typically a map)
  * `model` - The ReqLLM.Model struct

## Returns

  * `[ReqLLM.StreamChunk.t()]` - List of decoded stream chunks (may be empty)

## Terminal Metadata

For terminal streaming events, providers should return metadata chunks:

    # Final usage and completion metadata
    ReqLLM.StreamChunk.meta(%{
      usage: %{input_tokens: 10, output_tokens: 25},
      finish_reason: :stop,
      terminal?: true
    })

## Examples

    def decode_stream_event(%{data: %{"choices" => [%{"delta" => delta}]}}, _model) do
      case delta do
        %{"content" => content} when content != "" ->
          [ReqLLM.StreamChunk.text(content)]
        _ ->
          []
      end
    end

    # Handle terminal [DONE] event
    def decode_stream_event(%{data: "[DONE]"}, _model) do
      # Provider should have accumulated usage data
      [ReqLLM.StreamChunk.meta(%{terminal?: true})]
    end

# `decode_stream_event`
*optional* 

```elixir
@callback decode_stream_event(map(), LLMDB.Model.t(), any()) ::
  {[ReqLLM.StreamChunk.t()], any()}
```

Decode streaming event with provider-specific state (optional, alternative to decode_stream_event/2).

This stateful variant of `decode_stream_event/2` allows providers to maintain state
across streaming chunks. Use this when your provider needs to accumulate data or
track parsing state across multiple streaming events.

If both `decode_stream_event/3` and `decode_stream_event/2` are defined, the 3-arity
version takes precedence during streaming.

## Parameters

  * `event` - Parsed streaming event map (SSE events have `:event`, `:data`, etc.)
  * `model` - The ReqLLM.Model struct
  * `provider_state` - Current provider state from `init_stream_state/1`

## Returns

`{chunks, new_provider_state}` where:
  * `chunks` - List of StreamChunk structs to emit
  * `new_provider_state` - Updated state for next event

## Examples

    def decode_stream_event(event, model, state) do
      chunks = ReqLLM.Provider.Defaults.default_decode_stream_event(event, model)
      
      Enum.reduce(chunks, {[], state}, fn chunk, {acc, st} ->
        case chunk.type do
          :content ->
            {emitted, new_st} = transform_content(st, chunk.text)
            {acc ++ emitted, new_st}
          _ ->
            {acc ++ [chunk], st}
        end
      end)
    end

# `default_env_key`
*optional* 

```elixir
@callback default_env_key() :: String.t()
```

Returns the default environment variable name for API authentication.

This callback provides the fallback environment variable name when the
provider metadata doesn't specify one. Generated automatically by the
DSL if `default_env_key` is provided.

## Returns

  * `String.t()` - Environment variable name (e.g., "ANTHROPIC_API_KEY")

# `encode_body`

```elixir
@callback encode_body(Req.Request.t()) :: Req.Request.t()
```

Encodes request body for provider API.

This callback is typically used as a Req request step that transforms the
request options (especially `:context`) into the provider-specific JSON body.

## Parameters

  * `request` - The Req.Request struct with options to encode

## Returns

  * `Req.Request.t()` - Request with encoded body

# `extract_usage`
*optional* 

```elixir
@callback extract_usage(term(), LLMDB.Model.t() | nil) :: {:ok, map()} | {:error, term()}
```

Extracts usage/cost metadata from response body (optional).

This callback is called by `ReqLLM.Step.Usage` if the provider module
exports this function. It allows custom usage extraction beyond the
standard formats.

## Parameters

  * `body` - The response body (typically a map)
  * `model` - The ReqLLM.Model struct (may be nil)

## Returns

  * `{:ok, map()}` - Usage metadata map with keys like `:input`, `:output`
  * `{:error, term()}` - Extraction error

# `flush_stream_state`
*optional* 

```elixir
@callback flush_stream_state(LLMDB.Model.t(), any()) :: {[ReqLLM.StreamChunk.t()], any()}
```

Flush buffered provider state when stream ends (optional).

This callback is invoked when the stream completes, allowing providers to emit
any buffered content that hasn't been sent yet. This is useful for stateful
transformations that may hold partial data waiting for completion signals.

## Parameters

  * `model` - The ReqLLM.Model struct
  * `provider_state` - Final provider state from last `decode_stream_event/3`

## Returns

`{chunks, new_provider_state}` where:
  * `chunks` - List of final StreamChunk structs to emit
  * `new_provider_state` - Updated state (often with buffer cleared)

## Examples

    def flush_stream_state(_model, %{buffer: ""} = state) do
      {[], state}
    end

    def flush_stream_state(_model, %{mode: :thinking, buffer: text} = state) do
      {[ReqLLM.StreamChunk.thinking(text)], %{state | buffer: ""}}
    end

# `init_stream_state`
*optional* 

```elixir
@callback init_stream_state(LLMDB.Model.t()) :: any()
```

Initialize provider-specific streaming state (optional).

This callback allows providers to set up stateful transformations for streaming
responses. The returned state will be threaded through `decode_stream_event/3` calls
and passed to `flush_stream_state/2` when the stream ends.

## Parameters

  * `model` - The ReqLLM.Model struct

## Returns

Provider-specific state of any type. Commonly a map with transformation state.

## Examples

    # Initialize state for <think> tag parsing
    def init_stream_state(_model) do
      %{mode: :text, buffer: ""}
    end

# `normalize_model_id`
*optional* 

```elixir
@callback normalize_model_id(String.t()) :: String.t()
```

Normalizes a model ID for metadata lookup (optional).

This callback allows providers to normalize model identifiers before looking up
metadata in the provider's model registry. Useful when providers have multiple
aliases or formats for the same underlying model (e.g., regional inference profiles).

## Parameters

  * `model_id` - The model identifier string to normalize

## Returns

  * `String.t()` - The normalized model ID for metadata lookup

## Examples

    # AWS Bedrock inference profiles - strip region prefix for metadata lookup
    def normalize_model_id("us.anthropic.claude-3-sonnet"), do: "anthropic.claude-3-sonnet"
    def normalize_model_id("eu.meta.llama-3"), do: "meta.llama-3"
    def normalize_model_id(model_id), do: model_id

If this callback is not implemented, the model ID is used as-is for metadata lookup.

# `oauth_provider_id`
*optional* 

```elixir
@callback oauth_provider_id() :: String.t()
```

Returns the provider key used inside oauth/auth JSON files.

This is optional and defaults to the provider atom as a string.

# `parse_stream_protocol`
*optional* 

```elixir
@callback parse_stream_protocol(binary(), term()) ::
  {:ok, [map()], term()} | {:incomplete, term()} | {:error, term()}
```

Parse raw binary stream protocol data into events.

This callback allows providers to implement custom streaming protocols (SSE, AWS Event Stream,
etc.) while maintaining a consistent interface. The default implementation uses SSE parsing.

## Parameters

  * `chunk` - Raw binary chunk from the HTTP stream
  * `state` - Opaque parser state from previous incomplete chunks

## Returns

  * `{:ok, events, state}` - Successfully parsed events with updated parser state
  * `{:incomplete, state}` - Need more data, return updated parser state
  * `{:error, reason}` - Parse error

## Examples

    # Default SSE implementation (provided automatically)
    def parse_stream_protocol(chunk, state) do
      {events, new_state} = ReqLLM.Streaming.SSE.accumulate_and_parse(chunk, state)
      {:ok, events, new_state}
    end

    # Custom binary protocol (e.g., AWS Event Stream)
    def parse_stream_protocol(chunk, state) do
      data = (state || "") <> chunk
      case ReqLLM.AWSEventStream.parse_binary(data) do
        {:ok, events, rest} -> {:ok, events, rest}
        {:incomplete, data} -> {:incomplete, data}
        {:error, reason} -> {:error, reason}
      end
    end

# `prepare_request`

```elixir
@callback prepare_request(operation(), LLMDB.Model.t() | term(), term(), keyword()) ::
  {:ok, Req.Request.t()} | {:error, Exception.t()}
```

Prepares a new request for a specific operation type.

This callback creates and configures a new Req request from scratch for the
given operation, model, and parameters. It should handle all operation-specific
configuration including authentication, headers, and base URLs.

## Parameters

  * `operation` - The type of operation (:chat, :embed, :moderate, etc.)
  * `model` - The ReqLLM.Model struct or model identifier
  * `data` - Operation-specific data (messages for chat, text for embed, etc.)
  * `opts` - Additional options (stream, temperature, etc.)
    - For `:object` operations, opts includes `:compiled_schema` with the schema definition

## Returns

  * `{:ok, Req.Request.t()}` - Successfully configured request
  * `{:error, Exception.t()}` - Configuration error (using Splode exception types)

## Examples

    # Chat operation
    def prepare_request(:chat, model, messages, opts) do
      {:ok, request} = Req.new(base_url: "https://api.anthropic.com")
      request = add_auth_headers(request)
      request = put_in(request.options[:json], %{
        model: model.name,
        messages: messages,
        stream: opts[:stream] || false
      })
      {:ok, request}
    end

    # Object generation with schema
    def prepare_request(:object, model, context, opts) do
      compiled_schema = Keyword.fetch!(opts, :compiled_schema)
      # Use compiled_schema.schema for tool definitions
      prepare_request(:chat, model, context, updated_opts)
    end

    # Embedding operation  
    def prepare_request(:embed, model, text, opts) do
      {:ok, request} = Req.new(base_url: "https://api.anthropic.com/v1/embed")
      {:ok, add_auth_headers(request)}
    end

# `refresh_oauth_credentials`
*optional* 

```elixir
@callback refresh_oauth_credentials(map(), keyword()) ::
  {:ok, map()} | {:error, String.t()}
```

Refreshes provider OAuth credentials loaded from an oauth/auth JSON file.

Returns a map that includes updated `access`, `refresh`, and `expires` fields.

# `stream_protocol_parser`
*optional* 

```elixir
@callback stream_protocol_parser(LLMDB.Model.t(), keyword()) ::
  (binary(), term() -&gt;
     {:ok, [map()], term()} | {:incomplete, term()} | {:error, term()})
  | nil
```

Picks the stream protocol parser for a request, in the shape of
`parse_stream_protocol/2`. `nil` falls back to `parse_stream_protocol/2`.

# `stream_transport`
*optional* 

```elixir
@callback stream_transport(LLMDB.Model.t(), keyword()) :: :http | :websocket | :in_process
```

Selects the provider streaming transport.

The default transport is `:http`. Providers can return `:websocket` for the
OpenAI WebSocket path or `:in_process` when they produce canonical
`ReqLLM.StreamChunk` values without a network request.

# `thinking_constraints`
*optional* 

```elixir
@callback thinking_constraints() ::
  %{required_temperature: float(), min_max_tokens: pos_integer()} | :none
```

Returns thinking/reasoning constraints for models with extended thinking capability.

Some providers (e.g., AWS Bedrock, Google Vertex AI) have platform-specific requirements
when extended thinking is enabled:
- Fixed temperature value (typically 1.0)
- Minimum max_tokens to accommodate thinking budget

Returns a map with:
- `:required_temperature` - Temperature that must be used (float)
- `:min_max_tokens` - Minimum max_tokens value (integer)

Returns `:none` if no special constraints apply.

## Examples

    # Provider with thinking constraints
    def thinking_constraints do
      %{required_temperature: 1.0, min_max_tokens: 4001}
    end

    # Provider without thinking constraints
    def thinking_constraints, do: :none

# `tool_call_id_policy`
*optional* 

```elixir
@callback tool_call_id_policy(operation(), LLMDB.Model.t() | map(), keyword() | map()) ::
  map() | keyword()
```

Returns tool call ID compatibility policy for this provider (optional).

Providers can enforce tool call ID constraints when a context built on one
provider is sent to a different provider.

# `translate_options`
*optional* 

```elixir
@callback translate_options(operation(), LLMDB.Model.t(), keyword()) ::
  {keyword(), [String.t()]}
```

Translates canonical options to provider-specific parameters (optional).

This callback allows providers to modify option keys and values before
they are sent to the API. Useful for handling parameter name differences
and model-specific restrictions.

## Parameters

  * `operation` - The operation type (:chat, :embed, etc.)
  * `model` - The ReqLLM.Model struct
  * `opts` - Canonical options after validation

## Returns

  * `{translated_opts, warnings}` - Tuple of translated options and warning messages

## Examples

    # OpenAI o1 models need max_completion_tokens instead of max_tokens
    def translate_options(:chat, %Model{model: <<"o1", _::binary>>}, opts) do
      {opts, warnings} = translate_max_tokens(opts)
      {opts, warnings}
    end

    # Drop unsupported parameters with warnings
    def translate_options(:chat, %Model{model: <<"o1", _::binary>>}, opts) do
      results = [
        translate_rename(opts, :max_tokens, :max_completion_tokens),
        translate_drop(opts, :temperature, "OpenAI o1 models do not support :temperature")
      ]
      translate_combine_warnings(results)
    end

# `get!`

```elixir
@spec get!(atom()) :: module()
```

Registry function with bang syntax (raises on error).

# `parse_stream_protocol`

Default implementation of parse_stream_protocol using SSE parsing.

Providers can override this to implement custom streaming protocols.

---

*Consult [api-reference.md](api-reference.md) for complete listing*