feat(fireworks): Responses API for Fireworks AI

pyproject.toml

@@ -49,7 +49,7 @@ cerebras = [
 ]
 
 fireworks = [
-  "fireworks-ai",
+  "fireworks-ai>=0.17.16",
   "instructor",
 ]
 

src/any_llm/providers/fireworks/fireworks.py

@@ -7,11 +7,20 @@
     msg = "fireworks-ai is not installed. Please install it with `pip install any-llm-sdk[fireworks]`"
     raise ImportError(msg) from exc
 
+from openai import OpenAI, Stream
 from pydantic import BaseModel
 
 from any_llm.provider import Provider
 from any_llm.providers.fireworks.utils import _create_openai_chunk_from_fireworks_chunk
-from any_llm.types.completion import ChatCompletion, ChatCompletionChunk, ChatCompletionMessage, Choice, CompletionUsage
+from any_llm.types.completion import (
+    ChatCompletion,
+    ChatCompletionChunk,
+    ChatCompletionMessage,
+    Choice,
+    CompletionUsage,
+    Reasoning,
+)
+from any_llm.types.responses import Response, ResponseStreamEvent
 
 
 class FireworksProvider(Provider):
@@ -21,7 +30,7 @@ class FireworksProvider(Provider):
 
     SUPPORTS_COMPLETION_STREAMING = True
     SUPPORTS_COMPLETION = True
-    SUPPORTS_RESPONSES = False
+    SUPPORTS_RESPONSES = True
     SUPPORTS_COMPLETION_REASONING = False
     SUPPORTS_EMBEDDING = False
 
@@ -95,3 +104,27 @@ def completion(
             choices=choices_out,
             usage=usage,
         )
+
+    def responses(self, model: str, input_data: Any, **kwargs: Any) -> Response | Iterator[ResponseStreamEvent]:
+        """Call Fireworks Responses API and normalize into ChatCompletion/Chunks."""
+        client = OpenAI(
+            base_url="https://api.fireworks.ai/inference/v1",
+            api_key=self.config.api_key,
+        )
+        response = client.responses.create(
+            model=model,
+            input=input_data,
+            **kwargs,
+        )
+        if not isinstance(response, Response | Stream):
+            err_msg = f"Responses API returned an unexpected type: {type(response)}"
+            raise ValueError(err_msg)
+        if isinstance(response, Response) and not isinstance(response, Stream):
+            # See https://fireworks.ai/blog/response-api for details about Fireworks Responses API support
+            reasoning = response.output[-1].content[0].text.split("</think>")[-1]  # type: ignore[union-attr,index]
+            if reasoning:
+                reasoning = reasoning.strip()
+                response.output[-1].content[0].text = response.output[-1].content[0].text.split("</think>")[0]  # type: ignore[union-attr,index]
+            response.reasoning = Reasoning(content=reasoning) if reasoning else None  # type: ignore[assignment]
+
+        return response

src/any_llm/types/responses.py

@@ -1,5 +1,8 @@
 from openai.types.responses import Response as OpenAIResponse
+from openai.types.responses import ResponseOutputMessage as OpenAIResponseOutputMessage
 from openai.types.responses import ResponseStreamEvent as OpenAIResponseStreamEvent
 
 Response = OpenAIResponse
 ResponseStreamEvent = OpenAIResponseStreamEvent
+
+ResponseOutputMessage = OpenAIResponseOutputMessage

tests/conftest.py

@@ -6,10 +6,16 @@
 
 
 @pytest.fixture
-def provider_completion_reasoning_model_map() -> dict[ProviderName, str]:
+def provider_reasoning_model_map() -> dict[ProviderName, str]:
     return {
         ProviderName.MISTRAL: "magistral-small-latest",
         ProviderName.GROQ: "openai/gpt-oss-20b",
+        ProviderName.FIREWORKS: "accounts/fireworks/models/deepseek-r1",
+        ProviderName.OPENAI: "gpt-5-nano",
+        ProviderName.MISTRAL: "magistral-small-latest",
+        ProviderName.XAI: "grok-3-mini-latest",
+        ProviderName.OLLAMA: "gpt-oss:20b",
+        ProviderName.LMSTUDIO: "openai/gpt-oss-20b",  # You must have LM Studio running and the server enabled
     }
 
 

tests/integration/test_reasoning.py

@@ -6,18 +6,21 @@
 
 from any_llm import ProviderName, completion
 from any_llm.exceptions import MissingApiKeyError
+from any_llm.provider import ProviderFactory
 from any_llm.types.completion import ChatCompletion
 
 
 def test_reasoning_providers(
     provider: ProviderName,
-    provider_completion_reasoning_model_map: dict[ProviderName, str],
+    provider_reasoning_model_map: dict[ProviderName, str],
     provider_extra_kwargs_map: dict[ProviderName, dict[str, Any]],
 ) -> None:
     """Test that all supported providers can be loaded successfully."""
-    model_id = provider_completion_reasoning_model_map.get(provider, None)
-    if not model_id:
-        pytest.skip(f"{provider.value} does not yet test reasoning, skipping")
+    providers_metadata = ProviderFactory.get_all_provider_metadata()
+    provider_metadata = next(metadata for metadata in providers_metadata if metadata["provider_key"] == provider.value)
+    if not provider_metadata["reasoning"]:
+        pytest.skip(f"{provider.value} does not support completion reasoning, skipping")
+    model_id = provider_reasoning_model_map[provider]
     extra_kwargs = provider_extra_kwargs_map.get(provider, {})
     try:
         result = completion(

tests/integration/test_responses.py

@@ -12,18 +12,22 @@
 
 def test_responses(
     provider: ProviderName,
-    provider_model_map: dict[ProviderName, str],
+    provider_reasoning_model_map: dict[ProviderName, str],
     provider_extra_kwargs_map: dict[ProviderName, dict[str, Any]],
 ) -> None:
     """Test that all supported providers can be loaded successfully."""
     providers_metadata = ProviderFactory.get_all_provider_metadata()
     provider_metadata = next(metadata for metadata in providers_metadata if metadata["provider_key"] == provider.value)
     if not provider_metadata["responses"]:
         pytest.skip(f"{provider.value} does not support responses, skipping")
-    model_id = provider_model_map[provider]
+    model_id = provider_reasoning_model_map[provider]
     extra_kwargs = provider_extra_kwargs_map.get(provider, {})
     try:
-        result = responses(f"{provider.value}/{model_id}", **extra_kwargs, input_data="Hello")
+        result = responses(
+            f"{provider.value}/{model_id}",
+            **extra_kwargs,
+            input_data="What's the capital of France? Please think step by step.",
+        )
     except MissingApiKeyError:
         pytest.skip(f"{provider.value} API key not provided, skipping")
     except (httpx.HTTPStatusError, httpx.ConnectError, APIConnectionError):