scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

feat(parallel_ai): add Parallel AI as an LLM provider (responses, chat, messages)

Browse source 
Parallel AI is registered as a search-only provider today, so its OpenAI
Responses-compatible endpoint is unreachable through the gateway: routing
parallel_ai/parallel to /v1/responses, /v1/chat/completions or /v1/messages
fails with 'LLM Provider NOT provided'.

Registers parallel_ai as an LLM provider with a native Responses API config.
completion() and /v1/messages are served through the existing responses bridge.
Parallel exposes one model whose performance tier is selected by
reasoning.effort, so cost tracking resolves the priced model from the effective
effort, and parallel-low / -medium / -high aliases pin a tier and bill at its
own published rate.

A server-configured key is never forwarded to a caller-supplied api_base; an
explicit api_key is required when overriding the base, matching the trust model
the search adapter already uses.
This commit is contained in:
James Liounis 2026-08-21 15:51:17 -04:00
parent 66525bee2b
commit 1ed4ae981a
17 changed files with 1022 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/__init__.py
View file

@ -1786,6 +1786,9 @@ if TYPE_CHECKING:
from .llms.manus.responses.transformation import (
ManusResponsesAPIConfig as ManusResponsesAPIConfig,
)
from .llms.parallel_ai.responses.transformation import (
ParallelAIResponsesConfig as ParallelAIResponsesConfig,
)
from .llms.perplexity.responses.transformation import (
PerplexityResponsesConfig as PerplexityResponsesConfig,
)

5
litellm/_lazy_imports_registry.py
View file

@ -238,6 +238,7 @@ LLM_CONFIG_NAMES: Final = (
"HostedVLLMResponsesAPIConfig",
"VolcEngineResponsesAPIConfig",
"PerplexityResponsesConfig",
"ParallelAIResponsesConfig",
"DatabricksResponsesAPIConfig",
"OpenRouterResponsesAPIConfig",
"BedrockMantleResponsesAPIConfig",
@ -961,6 +962,10 @@ _LLM_CONFIGS_IMPORT_MAP: Final = {
".llms.perplexity.responses.transformation",
"PerplexityResponsesConfig",
),
"ParallelAIResponsesConfig": (
".llms.parallel_ai.responses.transformation",
"ParallelAIResponsesConfig",
),
"DatabricksResponsesAPIConfig": (
".llms.databricks.responses.transformation",
"DatabricksResponsesAPIConfig",

3
litellm/constants.py
View file

@ -562,6 +562,7 @@ LITELLM_CHAT_PROVIDERS: Final = [
"ollama_chat",
"deepinfra",
"perplexity",
"parallel_ai",
"mistral",
"groq",
"gigachat",
@ -741,6 +742,7 @@ DEFAULT_CHAT_COMPLETION_PARAM_VALUES: Final = {
openai_compatible_endpoints: Final[list] = [
"api.perplexity.ai",
"api.parallel.ai",
"api.endpoints.anyscale.com/v1",
"api.deepinfra.com/v1/openai",
"api.mistral.ai/v1",
@ -801,6 +803,7 @@ openai_compatible_providers: Final[list] = [
"tencent",
"deepinfra",
"perplexity",
"parallel_ai",
"xinference",
"xai",
"zai",

32
litellm/cost_calculator.py
View file

@ -2,7 +2,9 @@
## File for 'response_cost' calculation in Logging
import logging
import time
from collections.abc import Mapping
from functools import lru_cache
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from httpx import Response
@ -179,6 +181,7 @@ _SEARCH_CALL_TYPES: Final = frozenset(
)
_AREALTIME_CALL_TYPE: Final = CallTypes.arealtime.value
EMPTY_OPTIONAL_PARAMS: Final[Mapping[str, object]] = MappingProxyType({})
_MCP_CALL_TYPE: Final = CallTypes.call_mcp_tool.value
@ -872,6 +875,20 @@ def _normalize_service_tier(service_tier: object) -> str | None:
return service_tier
def _parallel_ai_response_pricing_model(model: str, optional_params: Mapping[str, object] | None) -> str | None:
from litellm.llms.parallel_ai.responses.cost_calculator import (
is_parallel_ai_response_model,
parallel_ai_response_pricing_model,
)
if not is_parallel_ai_response_model(model):
return None
return parallel_ai_response_pricing_model(
model=model,
optional_params=optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS,
)
def _extract_service_tier(source: object) -> str | None:
"""Read a raw ``service_tier`` off a response body or usage object, dict or pydantic model alike."""
if isinstance(source, BaseModel):
@ -1228,12 +1245,25 @@ def completion_cost(
service_tier = _normalize_service_tier(service_tier)
provider_for_cost: Final = _get_provider_for_cost_calc(
model=model,
custom_llm_provider=custom_llm_provider,
)
pricing_base_model: Final = (
_parallel_ai_response_pricing_model(
model=model or "parallel",
optional_params=optional_params,
)
if base_model is None and custom_pricing is not True and provider_for_cost == LlmProviders.PARALLEL_AI.value
else base_model
)
selected_model: Final = _select_model_name_for_cost_calc(
model=model,
completion_response=completion_response,
custom_llm_provider=custom_llm_provider,
custom_pricing=custom_pricing,
base_model=base_model,
base_model=pricing_base_model,
router_model_id=router_model_id,
)

15
litellm/litellm_core_utils/get_llm_provider_logic.py
View file

@ -233,6 +233,14 @@ def get_llm_provider(
if endpoint == "api.perplexity.ai":
custom_llm_provider = "perplexity"
dynamic_api_key = get_secret_str("PERPLEXITYAI_API_KEY")
elif endpoint == "api.parallel.ai":
from litellm.llms.parallel_ai.common_utils import (
resolve_parallel_ai_credentials,
)
custom_llm_provider = "parallel_ai" # rebind-ok: function returns custom_llm_provider
resolved = resolve_parallel_ai_credentials(api_base=api_base, api_key=api_key)
api_base, dynamic_api_key = resolved # rebind-ok: function returns api_base
elif endpoint == "api.endpoints.anyscale.com/v1":
custom_llm_provider = "anyscale"
dynamic_api_key = get_secret_str("ANYSCALE_API_KEY")
@ -550,6 +558,13 @@ def _get_openai_compatible_provider_info(
api_base,
dynamic_api_key,
) = litellm.PerplexityChatConfig()._get_openai_compatible_provider_info(api_base, api_key)
elif custom_llm_provider == "parallel_ai":
from litellm.llms.parallel_ai.common_utils import resolve_parallel_ai_credentials
# parallel_ai serves the OpenAI Responses-compatible API at https://api.parallel.ai/v1/responses
api_base, dynamic_api_key = resolve_parallel_ai_credentials( # rebind-ok: function returns api_base
api_base=api_base, api_key=api_key
)
elif custom_llm_provider == "aiohttp_openai":
return model, "aiohttp_openai", api_key, api_base
elif custom_llm_provider == "anyscale":

52
litellm/llms/parallel_ai/common_utils.py Normal file
View file

@ -0,0 +1,52 @@
"""
Shared credential and endpoint resolution for the Parallel AI provider.
"""
from typing import Final
from urllib.parse import urlsplit
from litellm.secret_managers.main import get_secret_str
PARALLEL_AI_API_BASE: Final = "https://api.parallel.ai"
def _origin(url: str) -> tuple[str, str]:
"""Scheme and host of a base URL; a scheme-less base is read as https."""
normalized: Final = url if "://" in url else f"https://{url}"
split: Final = urlsplit(normalized)
return split.scheme.lower(), split.netloc.lower()
def _is_trusted_api_base(caller_api_base: str, env_api_base: str | None) -> bool:
"""Whether a caller-supplied base names an origin the server key may be sent to.
Compares scheme as well as host: matching the host alone would accept
``http://api.parallel.ai`` and put the server key on the wire in plaintext.
"""
trusted: Final = frozenset(_origin(base) for base in (PARALLEL_AI_API_BASE, env_api_base) if base)
scheme, host = _origin(caller_api_base)
return bool(host) and (scheme, host) in trusted
def resolve_parallel_ai_credentials(api_base: str | None, api_key: str | None) -> tuple[str, str | None]:
"""
Resolve the effective (api_base, api_key) pair for a Parallel AI LLM request.
A server-managed key (from env) is only used when the request targets the
provider default or the operator's own PARALLEL_AI_API_BASE override; a
caller-supplied base must bring its own key, so the server credential is
never forwarded to a caller-chosen host.
"""
env_api_base: Final = get_secret_str("PARALLEL_AI_API_BASE")
resolved_api_base: Final = api_base or env_api_base or PARALLEL_AI_API_BASE
if api_key:
return resolved_api_base, api_key
server_api_key: Final = get_secret_str("PARALLEL_AI_API_KEY") or get_secret_str("PARALLEL_API_KEY")
if server_api_key and api_base and not _is_trusted_api_base(api_base, env_api_base):
raise ValueError(
f"Refusing to send the server-configured Parallel AI key to the caller-supplied "
f"api_base '{api_base}'. Pass an explicit api_key when overriding api_base."
)
return resolved_api_base, server_api_key

7
litellm/llms/parallel_ai/responses/__init__.py Normal file
View file

@ -0,0 +1,7 @@
"""
Parallel AI Responses API module.
"""
from litellm.llms.parallel_ai.responses.transformation import ParallelAIResponsesConfig
__all__ = ("ParallelAIResponsesConfig",)

41
litellm/llms/parallel_ai/responses/cost_calculator.py Normal file
View file

@ -0,0 +1,41 @@
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from pydantic import TypeAdapter, ValidationError
PARALLEL_AI_DEFAULT_REASONING_EFFORT: Final[str] = "medium"
PARALLEL_AI_EFFORT_TIER_MODELS: Final[Mapping[str, str]] = MappingProxyType(
{
"parallel-low": "low",
"parallel-medium": "medium",
"parallel-high": "high",
}
)
PARALLEL_AI_REASONING_EFFORTS: Final[frozenset[str]] = frozenset(PARALLEL_AI_EFFORT_TIER_MODELS.values())
REASONING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
def is_parallel_ai_response_model(model: str) -> bool:
model_without_provider: Final[str] = model.removeprefix("parallel_ai/")
return model_without_provider == "parallel" or model_without_provider in PARALLEL_AI_EFFORT_TIER_MODELS
def _reasoning_effort(optional_params: Mapping[str, object]) -> str:
try:
reasoning: Final = REASONING_ADAPTER.validate_python(optional_params.get("reasoning"))
except ValidationError:
return PARALLEL_AI_DEFAULT_REASONING_EFFORT
effort: Final[object] = reasoning.get("effort")
if isinstance(effort, str) and effort in PARALLEL_AI_REASONING_EFFORTS:
return effort
return PARALLEL_AI_DEFAULT_REASONING_EFFORT
def parallel_ai_response_pricing_model(model: str, optional_params: Mapping[str, object]) -> str:
"""Return the cost-map model matching the effective Responses reasoning effort."""
model_without_provider: Final[str] = model.removeprefix("parallel_ai/")
effort: Final[str] = PARALLEL_AI_EFFORT_TIER_MODELS.get(model_without_provider) or _reasoning_effort(
optional_params
)
return f"parallel_ai/parallel-{effort}"

120
litellm/llms/parallel_ai/responses/transformation.py Normal file
View file

@ -0,0 +1,120 @@
"""
Parallel AI Responses API, an OpenAI Responses-compatible web-research endpoint.
Provider quirks:
- single `parallel` model; the performance tier is selected via `reasoning.effort` (low/medium/high),
or with the `parallel-low` / `parallel-medium` / `parallel-high` aliases, which pin the tier and
carry per-tier pricing in the cost map
- no `tools` param; web grounding is built in
Ref: https://docs.parallel.ai/responses-api/responses-quickstart
"""
from typing import Final
import httpx
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.llms.parallel_ai.common_utils import resolve_parallel_ai_credentials
from litellm.llms.parallel_ai.responses.cost_calculator import (
PARALLEL_AI_EFFORT_TIER_MODELS as EFFORT_TIER_MODELS,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import ResponseInputParam, ResponsesAPIOptionalRequestParams, ResponsesAPIResponse
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
class ParallelAIResponsesConfig(OpenAIResponsesAPIConfig):
def get_supported_openai_params(self, model: str) -> list: # mutable-ok: BaseResponsesAPIConfig contract
"""Ref: https://docs.parallel.ai/responses-api/responses-quickstart"""
return [ # mutable-ok: callers concatenate with other lists
"stream",
"reasoning",
"instructions",
"text",
"previous_response_id",
]
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.PARALLEL_AI
def map_openai_params(
self,
response_api_optional_params: ResponsesAPIOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict: # mutable-ok: BaseResponsesAPIConfig contract
"""Parallel rejects unknown Responses params (e.g. `tools`), so unsupported keys are filtered out."""
supported: Final = frozenset(self.get_supported_openai_params(model))
return { # mutable-ok: request payload, merged downstream
key: value for key, value in response_api_optional_params.items() if key in supported
}
def validate_environment( # mutable-ok: BaseResponsesAPIConfig contract
self,
headers: dict, # mutable-ok: BaseResponsesAPIConfig contract
model: str,
litellm_params: GenericLiteLLMParams | None,
) -> dict: # mutable-ok: BaseResponsesAPIConfig contract
resolved_params: Final = litellm_params or GenericLiteLLMParams()
_, api_key = resolve_parallel_ai_credentials(api_base=resolved_params.api_base, api_key=resolved_params.api_key)
if api_key:
headers["Authorization"] = f"Bearer {api_key}" # rebind-ok: stamping the caller's headers is the contract
return headers
def get_complete_url(self, api_base: str | None, litellm_params: dict) -> str: # mutable-ok: base contract
resolved_api_base: Final = api_base or get_secret_str("PARALLEL_AI_API_BASE") or "https://api.parallel.ai"
trimmed: Final = resolved_api_base.rstrip("/")
if trimmed.endswith("/v1/responses"):
return trimmed
return f"{trimmed.removesuffix('/v1')}/v1/responses"
def transform_responses_api_request( # mutable-ok: BaseResponsesAPIConfig contract
self,
model: str,
input: str | ResponseInputParam,
response_api_optional_request_params: dict, # mutable-ok: BaseResponsesAPIConfig contract
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: BaseResponsesAPIConfig contract
) -> dict: # mutable-ok: BaseResponsesAPIConfig contract
"""The tier aliases pin `reasoning.effort` so each alias bills at its cost map entry."""
effort: Final = EFFORT_TIER_MODELS.get(model)
if effort is None:
return super().transform_responses_api_request(
model=model,
input=input,
response_api_optional_request_params=response_api_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)
return super().transform_responses_api_request(
model="parallel",
input=input,
response_api_optional_request_params={ # mutable-ok: request payload
**response_api_optional_request_params,
"reasoning": {"effort": effort}, # mutable-ok: request payload
},
litellm_params=litellm_params,
headers=headers,
)
def transform_response_api_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse:
"""The API reports model `parallel` for every tier; cost tracking prices by response model, so the alias is restored."""
response: Final = super().transform_response_api_response(
model=model, raw_response=raw_response, logging_obj=logging_obj
)
if model in EFFORT_TIER_MODELS:
response.model = model
return response
def supports_native_websocket(self) -> bool:
"""Parallel AI does not support native WebSocket for the Responses API"""
return False

36
litellm/model_prices_and_context_window_backup.json
View file

@ -35282,6 +35282,42 @@
"output_cost_per_token": 1.25e-07,
"source": "https://cloud.google.com/vertex-ai/generative-ai/docs/learn/models#foundation_models"
},
"parallel_ai/parallel": {
"input_cost_per_request": 0.05,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-high": {
"input_cost_per_request": 0.25,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-low": {
"input_cost_per_request": 0.01,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-medium": {
"input_cost_per_request": 0.05,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/search": {
"input_cost_per_query": 0.004,
"litellm_provider": "parallel_ai",

1
litellm/types/utils.py
View file

@ -3797,6 +3797,7 @@ class LlmProviders(str, Enum):
CURSOR = "cursor"
BEDROCK_MANTLE = "bedrock_mantle"
GDC = "gdc"
PARALLEL_AI = "parallel_ai"
# Create a set of all provider values for quick lookup

7
litellm/utils.py
View file

@ -6334,6 +6334,11 @@ def validate_environment(
keys_in_environment = True
else:
missing_keys.append("PERPLEXITYAI_API_KEY")
elif custom_llm_provider == "parallel_ai":
if "PARALLEL_AI_API_KEY" in os.environ or "PARALLEL_API_KEY" in os.environ:
keys_in_environment = True
else:
missing_keys.append("PARALLEL_AI_API_KEY")
elif custom_llm_provider == "voyage":
if "VOYAGE_API_KEY" in os.environ:
keys_in_environment = True
@ -8487,6 +8492,8 @@ class ProviderConfigManager:
return litellm.ManusResponsesAPIConfig()
elif litellm.LlmProviders.PERPLEXITY == provider:
return litellm.PerplexityResponsesConfig()
elif litellm.LlmProviders.PARALLEL_AI == provider:
return litellm.ParallelAIResponsesConfig()
elif litellm.LlmProviders.DATABRICKS == provider:
# Databricks Responses API is only compatible with OpenAI GPT models
if model and "gpt" in model.lower():

36
model_prices_and_context_window.json
View file

@ -35282,6 +35282,42 @@
"output_cost_per_token": 1.25e-07,
"source": "https://cloud.google.com/vertex-ai/generative-ai/docs/learn/models#foundation_models"
},
"parallel_ai/parallel": {
"input_cost_per_request": 0.05,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-high": {
"input_cost_per_request": 0.25,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-low": {
"input_cost_per_request": 0.01,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/parallel-medium": {
"input_cost_per_request": 0.05,
"input_cost_per_token": 0.0,
"litellm_provider": "parallel_ai",
"mode": "responses",
"output_cost_per_token": 0.0,
"supports_reasoning": true,
"supports_response_schema": true
},
"parallel_ai/search": {
"input_cost_per_query": 0.004,
"litellm_provider": "parallel_ai",

6
provider_endpoints_support.json
View file

@ -1934,9 +1934,9 @@
"display_name": "Parallel AI (`parallel_ai`)",
"url": "https://docs.litellm.ai/docs/search/parallel_ai",
"endpoints": {
"chat_completions": false,
"messages": false,
"responses": false,
"chat_completions": true,
"messages": true,
"responses": true,
"embeddings": false,
"image_generations": false,
"audio_transcriptions": false,

401
tests/test_litellm/llms/parallel_ai/responses/test_parallel_ai_responses_transformation.py Normal file
View file

@ -0,0 +1,401 @@
"""
Tests for Parallel AI Responses API transformation.
Source: litellm/llms/parallel_ai/responses/transformation.py
"""
import json
import pytest
from litellm.llms.parallel_ai.responses.cost_calculator import parallel_ai_response_pricing_model
from litellm.llms.parallel_ai.responses.transformation import ParallelAIResponsesConfig
from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams, ResponsesAPIResponse
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
def _parallel_response_body(response_id: str = "resp_test") -> dict[str, object]:
return {
"id": response_id,
"object": "response",
"created_at": 1700000000,
"model": "parallel",
"status": "completed",
"output": [
{
"type": "message",
"id": "msg_1",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "grounded answer", "annotations": []}],
}
],
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
"usage": {
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens_details": {"reasoning_tokens": 0},
},
}
class TestParallelAIResponsesConfig:
def test_provider_config_manager_returns_parallel_config(self):
config = ProviderConfigManager.get_provider_responses_api_config(
provider=LlmProviders.PARALLEL_AI, model="parallel"
)
assert isinstance(config, ParallelAIResponsesConfig)
@pytest.mark.parametrize(
"api_base,expected",
[
(None, "https://api.parallel.ai/v1/responses"),
("https://api.parallel.ai", "https://api.parallel.ai/v1/responses"),
("https://api.parallel.ai/", "https://api.parallel.ai/v1/responses"),
("https://api.parallel.ai/v1", "https://api.parallel.ai/v1/responses"),
("https://proxy.example.com/v1/responses", "https://proxy.example.com/v1/responses"),
],
)
def test_get_complete_url(self, api_base, expected, monkeypatch):
monkeypatch.delenv("PARALLEL_AI_API_BASE", raising=False)
config = ParallelAIResponsesConfig()
assert config.get_complete_url(api_base=api_base, litellm_params={}) == expected
def test_validate_environment_sets_bearer_from_env(self, monkeypatch):
monkeypatch.delenv("PARALLEL_AI_API_KEY", raising=False)
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
config = ParallelAIResponsesConfig()
headers = config.validate_environment(headers={}, model="parallel", litellm_params=None)
assert headers["Authorization"] == "Bearer pk-test"
def test_validate_environment_prefers_explicit_key(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
config = ParallelAIResponsesConfig()
headers = config.validate_environment(
headers={},
model="parallel",
litellm_params=GenericLiteLLMParams(api_key="pk-explicit"),
)
assert headers["Authorization"] == "Bearer pk-explicit"
def test_reasoning_effort_is_supported(self):
config = ParallelAIResponsesConfig()
supported = config.get_supported_openai_params(model="parallel")
assert "reasoning" in supported
assert "instructions" in supported
assert "previous_response_id" in supported
assert "tools" not in supported
def test_unsupported_params_dropped_with_drop_params(self):
from litellm.responses.utils import ResponsesAPIRequestUtils
config = ParallelAIResponsesConfig()
params = ResponsesAPIOptionalRequestParams(
reasoning={"effort": "high"},
tools=[{"type": "web_search"}],
temperature=0.5,
)
mapped = ResponsesAPIRequestUtils.get_optional_params_responses_api(
model="parallel",
responses_api_provider_config=config,
response_api_optional_params=params,
drop_params=True,
)
assert mapped["reasoning"] == {"effort": "high"}
assert "tools" not in mapped
assert "temperature" not in mapped
def test_unsupported_params_raise_without_drop_params(self):
import litellm as litellm_module
from litellm.responses.utils import ResponsesAPIRequestUtils
config = ParallelAIResponsesConfig()
params = ResponsesAPIOptionalRequestParams(tools=[{"type": "web_search"}])
with pytest.raises(litellm_module.UnsupportedParamsError):
ResponsesAPIRequestUtils.get_optional_params_responses_api(
model="parallel",
responses_api_provider_config=config,
response_api_optional_params=params,
drop_params=False,
)
def test_transform_request_sends_parallel_model(self):
config = ParallelAIResponsesConfig()
request = config.transform_responses_api_request(
model="parallel",
input="What is the latest AI news?",
response_api_optional_request_params={"reasoning": {"effort": "low"}},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request["model"] == "parallel"
assert request["input"] == "What is the latest AI news?"
assert request["reasoning"] == {"effort": "low"}
def test_no_native_websocket(self):
assert ParallelAIResponsesConfig().supports_native_websocket() is False
def test_untrusted_api_base_refuses_server_key(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-server-secret")
monkeypatch.delenv("PARALLEL_AI_API_BASE", raising=False)
config = ParallelAIResponsesConfig()
with pytest.raises(ValueError, match="Refusing to send"):
config.validate_environment(
headers={},
model="parallel",
litellm_params=GenericLiteLLMParams(api_base="https://attacker.example.com"),
)
def test_untrusted_api_base_with_explicit_key_is_allowed(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-server-secret")
config = ParallelAIResponsesConfig()
headers = config.validate_environment(
headers={},
model="parallel",
litellm_params=GenericLiteLLMParams(api_base="https://proxy.example.com", api_key="pk-caller"),
)
assert headers["Authorization"] == "Bearer pk-caller"
class TestParallelAICompletionBridge:
@pytest.mark.respx()
def test_completion_routes_through_responses_api(self, respx_mock, monkeypatch):
"""litellm.completion() on a mode=responses model must bridge to /v1/responses."""
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
import litellm as litellm_module
litellm_module.model_cost = litellm_module.get_model_cost_map(url="")
respx_mock.post("https://api.parallel.ai/v1/responses").respond(json=_parallel_response_body())
response = litellm_module.completion(
model="parallel_ai/parallel",
messages=[{"role": "user", "content": "question"}],
)
assert response.choices[0].message.content == "grounded answer"
class TestParallelAIEffortTierAliases:
@pytest.mark.parametrize(
"alias,effort",
[("parallel-low", "low"), ("parallel-medium", "medium"), ("parallel-high", "high")],
)
def test_alias_pins_model_and_effort(self, alias, effort):
config = ParallelAIResponsesConfig()
request = config.transform_responses_api_request(
model=alias,
input="question",
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request["model"] == "parallel"
assert request["reasoning"] == {"effort": effort}
def test_alias_effort_wins_over_explicit_reasoning(self):
config = ParallelAIResponsesConfig()
request = config.transform_responses_api_request(
model="parallel-high",
input="question",
response_api_optional_request_params={"reasoning": {"effort": "low"}},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request["reasoning"] == {"effort": "high"}
def test_plain_model_passes_reasoning_through(self):
config = ParallelAIResponsesConfig()
request = config.transform_responses_api_request(
model="parallel",
input="question",
response_api_optional_request_params={"reasoning": {"effort": "low"}},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request["model"] == "parallel"
assert request["reasoning"] == {"effort": "low"}
def test_alias_restored_on_response_model_for_cost_tracking(self):
from unittest.mock import MagicMock
config = ParallelAIResponsesConfig()
raw = MagicMock()
raw.json.return_value = {
"id": "resp_1",
"object": "response",
"created_at": 1700000000,
"model": "parallel",
"status": "completed",
"output": [],
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
}
response = config.transform_response_api_response(
model="parallel-low", raw_response=raw, logging_obj=MagicMock()
)
assert response.model == "parallel-low"
raw.json.return_value["model"] = "parallel"
response_plain = config.transform_response_api_response(
model="parallel", raw_response=raw, logging_obj=MagicMock()
)
assert response_plain.model == "parallel"
class TestParallelAIReasoningEffortPricing:
@pytest.mark.parametrize(
"model,optional_params,expected_model",
[
("parallel", {"reasoning": {"effort": "low"}}, "parallel_ai/parallel-low"),
("parallel", {}, "parallel_ai/parallel-medium"),
("parallel", {"reasoning": {"effort": "high"}}, "parallel_ai/parallel-high"),
("parallel_ai/parallel", {"reasoning": {"effort": "low"}}, "parallel_ai/parallel-low"),
("parallel-high", {"reasoning": {"effort": "low"}}, "parallel_ai/parallel-high"),
],
)
def test_pricing_model_uses_effective_reasoning_effort(self, model, optional_params, expected_model):
assert parallel_ai_response_pricing_model(model=model, optional_params=optional_params) == expected_model
@pytest.mark.parametrize(
"model,optional_params,expected_cost",
[
("parallel_ai/parallel", {"reasoning": {"effort": "low"}}, 0.01),
("parallel_ai/parallel", {}, 0.05),
("parallel_ai/parallel", {"reasoning": {"effort": "high"}}, 0.25),
("parallel_ai/parallel-high", {"reasoning": {"effort": "low"}}, 0.25),
],
)
def test_completion_cost_uses_effective_reasoning_effort(self, model, optional_params, expected_cost, monkeypatch):
import litellm as litellm_module
monkeypatch.setattr(litellm_module, "model_cost", litellm_module.get_model_cost_map(url=""))
response = ResponsesAPIResponse(
id="resp_cost",
created_at=1700000000,
model="parallel",
output=[],
usage={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
)
cost = litellm_module.completion_cost(
completion_response=response,
model=model,
optional_params=optional_params,
call_type="responses",
)
assert cost == pytest.approx(expected_cost)
def test_explicit_base_model_remains_authoritative(self, monkeypatch):
import litellm as litellm_module
monkeypatch.setattr(litellm_module, "model_cost", litellm_module.get_model_cost_map(url=""))
response = ResponsesAPIResponse(
id="resp_custom_base",
created_at=1700000000,
model="parallel",
output=[],
usage={"input_tokens": 0, "output_tokens": 0, "total_tokens": 0},
)
cost = litellm_module.completion_cost(
completion_response=response,
model="parallel_ai/parallel",
optional_params={"reasoning": {"effort": "high"}},
base_model="parallel_ai/parallel-low",
call_type="responses",
)
assert cost == pytest.approx(0.01)
@pytest.mark.respx()
@pytest.mark.parametrize(
"reasoning,expected_cost",
[
({"effort": "low"}, 0.01),
(None, 0.05),
({"effort": "high"}, 0.25),
],
)
def test_responses_records_effort_aware_cost(self, reasoning, expected_cost, respx_mock, monkeypatch):
import litellm as litellm_module
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
monkeypatch.setattr(litellm_module, "model_cost", litellm_module.get_model_cost_map(url=""))
route = respx_mock.post("https://api.parallel.ai/v1/responses").respond(json=_parallel_response_body())
response = litellm_module.responses(
model="parallel_ai/parallel",
input="question",
reasoning=reasoning,
)
assert response.model == "parallel"
assert response._hidden_params["response_cost"] == pytest.approx(expected_cost)
request_body = json.loads(route.calls.last.request.content)
if reasoning is None:
assert "reasoning" not in request_body
else:
assert request_body["reasoning"] == reasoning
@pytest.mark.respx()
def test_streaming_response_records_high_effort_cost(self, respx_mock, monkeypatch):
import litellm as litellm_module
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
monkeypatch.setattr(litellm_module, "model_cost", litellm_module.get_model_cost_map(url=""))
monkeypatch.setattr(litellm_module, "include_cost_in_streaming_usage", True)
completed_event = {
"type": "response.completed",
"response": _parallel_response_body(response_id="resp_stream"),
}
stream_body = f"data: {json.dumps(completed_event)}\n\ndata: [DONE]\n\n".encode()
respx_mock.post("https://api.parallel.ai/v1/responses").respond(
content=stream_body,
headers={"content-type": "text/event-stream"},
)
stream = litellm_module.responses(
model="parallel_ai/parallel",
input="question",
reasoning={"effort": "high"},
stream=True,
)
events = list(stream)
assert len(events) == 1
assert events[0].response.model == "parallel"
assert events[0].response.usage.cost == pytest.approx(0.25)
@pytest.mark.respx()
def test_completion_bridge_records_high_effort_cost(self, respx_mock, monkeypatch):
import litellm as litellm_module
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
monkeypatch.setattr(litellm_module, "model_cost", litellm_module.get_model_cost_map(url=""))
route = respx_mock.post("https://api.parallel.ai/v1/responses").respond(json=_parallel_response_body())
response = litellm_module.completion(
model="parallel_ai/parallel",
messages=[{"role": "user", "content": "question"}],
reasoning={"effort": "high"},
)
assert response.choices[0].message.content == "grounded answer"
assert response._hidden_params["response_cost"] == pytest.approx(0.25)
request_body = json.loads(route.calls.last.request.content)
assert request_body["reasoning"] == {"effort": "high"}

123
tests/test_litellm/llms/parallel_ai/test_common_utils.py Normal file
View file

@ -0,0 +1,123 @@
"""
Tests for Parallel AI credential and provider resolution.
Source: litellm/llms/parallel_ai/common_utils.py
"""
import pytest
import litellm
from litellm.llms.parallel_ai.common_utils import resolve_parallel_ai_credentials
class TestResolveParallelAICredentials:
@pytest.fixture(autouse=True)
def _clean_env(self, monkeypatch):
monkeypatch.delenv("PARALLEL_AI_API_BASE", raising=False)
monkeypatch.delenv("PARALLEL_AI_API_KEY", raising=False)
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
def test_defaults(self, monkeypatch):
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
api_base, api_key = resolve_parallel_ai_credentials(api_base=None, api_key=None)
assert api_base == "https://api.parallel.ai"
assert api_key == "pk-test"
def test_prefers_parallel_ai_key(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-primary")
monkeypatch.setenv("PARALLEL_API_KEY", "pk-fallback")
_, api_key = resolve_parallel_ai_credentials(api_base=None, api_key=None)
assert api_key == "pk-primary"
def test_plaintext_provider_host_is_not_trusted(self, monkeypatch):
"""A host-only trust check would send the server key over plaintext HTTP."""
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
with pytest.raises(ValueError, match="Refusing to send"):
resolve_parallel_ai_credentials(api_base="http://api.parallel.ai", api_key=None)
def test_https_provider_host_is_trusted(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
api_base, api_key = resolve_parallel_ai_credentials(api_base="https://api.parallel.ai", api_key=None)
assert api_base == "https://api.parallel.ai"
assert api_key == "pk-env"
def test_operator_plaintext_override_is_trusted(self, monkeypatch):
"""The operator's own PARALLEL_AI_API_BASE is trusted at the scheme it names."""
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
monkeypatch.setenv("PARALLEL_AI_API_BASE", "http://parallel-proxy.internal")
_, api_key = resolve_parallel_ai_credentials(api_base="http://parallel-proxy.internal", api_key=None)
assert api_key == "pk-env"
def test_explicit_args_win(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
api_base, api_key = resolve_parallel_ai_credentials(api_base="https://proxy.example.com", api_key="pk-explicit")
assert api_base == "https://proxy.example.com"
assert api_key == "pk-explicit"
def test_untrusted_api_base_refuses_server_key(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-server-secret")
with pytest.raises(ValueError, match="Refusing to send"):
resolve_parallel_ai_credentials(api_base="https://attacker.example.com", api_key=None)
def test_operator_env_base_is_trusted(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-server-secret")
monkeypatch.setenv("PARALLEL_AI_API_BASE", "https://proxy.internal.example.com")
api_base, api_key = resolve_parallel_ai_credentials(api_base="https://proxy.internal.example.com", api_key=None)
assert api_base == "https://proxy.internal.example.com"
assert api_key == "pk-server-secret"
def test_untrusted_base_without_server_key_returns_none(self):
api_base, api_key = resolve_parallel_ai_credentials(api_base="https://proxy.example.com", api_key=None)
assert api_base == "https://proxy.example.com"
assert api_key is None
class TestParallelAIProviderWiring:
def test_get_llm_provider_routes_parallel_ai(self, monkeypatch):
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
model, provider, api_key, api_base = litellm.get_llm_provider("parallel_ai/parallel")
assert model == "parallel"
assert provider == "parallel_ai"
assert api_key == "pk-test"
assert api_base == "https://api.parallel.ai"
def test_get_llm_provider_detects_parallel_api_base(self, monkeypatch):
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
model, provider, api_key, api_base = litellm.get_llm_provider(
model="parallel", api_base="https://api.parallel.ai"
)
assert provider == "parallel_ai"
assert api_key == "pk-test"
def test_validate_environment_reports_missing_key(self, monkeypatch):
monkeypatch.delenv("PARALLEL_AI_API_KEY", raising=False)
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
result = litellm.validate_environment(model="parallel_ai/parallel")
assert result["keys_in_environment"] is False
assert "PARALLEL_AI_API_KEY" in result["missing_keys"]
def test_validate_environment_accepts_either_key_name(self, monkeypatch):
monkeypatch.delenv("PARALLEL_AI_API_KEY", raising=False)
monkeypatch.setenv("PARALLEL_API_KEY", "pk-test")
result = litellm.validate_environment(model="parallel_ai/parallel")
assert result["keys_in_environment"] is True
def test_api_base_detection_keeps_explicit_caller_key(self, monkeypatch):
monkeypatch.setenv("PARALLEL_AI_API_KEY", "pk-env")
model, provider, api_key, api_base = litellm.get_llm_provider(
model="parallel", api_base="https://api.parallel.ai", api_key="pk-explicit"
)
assert provider == "parallel_ai"
assert api_key == "pk-explicit"

138
tests/test_litellm/llms/parallel_ai/test_parallel_ai_responses_gateway.py Normal file
View file

@ -0,0 +1,138 @@
"""Gateway coverage for Parallel AI's Responses integration."""
from __future__ import annotations
from collections.abc import Iterator
from typing import Final
from unittest.mock import AsyncMock
import httpx
import pytest
from fastapi.testclient import TestClient
import litellm
from litellm import Router
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.proxy import proxy_server
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
PARALLEL_RESPONSES_URL: Final = "https://api.parallel.ai/v1/responses"
@pytest.fixture
def client() -> TestClient:
return TestClient(proxy_server.app, raise_server_exceptions=False)
@pytest.fixture
def auth_as() -> Iterator[None]:
async def _authorized_request() -> UserAPIKeyAuth:
return UserAPIKeyAuth(
api_key="hashed-sk-test",
user_id="parallel-test-user",
)
previous: Final = proxy_server.app.dependency_overrides.get(user_api_key_auth)
proxy_server.app.dependency_overrides[user_api_key_auth] = _authorized_request
try:
yield
finally:
if previous is None:
proxy_server.app.dependency_overrides.pop(user_api_key_auth, None)
else:
proxy_server.app.dependency_overrides[user_api_key_auth] = previous
def _parallel_response_body() -> dict[str, object]:
return {
"id": "resp_parallel_gateway",
"object": "response",
"created_at": 1700000000,
"model": "parallel",
"status": "completed",
"output": [
{
"type": "message",
"id": "msg_parallel_gateway",
"status": "completed",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": "Parallel grounded answer",
"annotations": [],
}
],
}
],
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
"usage": {
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens_details": {"reasoning_tokens": 0},
},
}
def _parallel_router() -> Router:
return Router(
model_list=[
{
"model_name": "parallel-gateway",
"litellm_params": {
"model": "parallel_ai/parallel",
"api_key": "parallel-responses-key",
"use_in_pass_through": True,
},
}
],
num_retries=0,
)
def _mock_async_post(
monkeypatch,
*,
url: str,
response_body: dict[str, object],
) -> AsyncMock:
response = httpx.Response(
status_code=200,
json=response_body,
request=httpx.Request("POST", url),
)
mock_post = AsyncMock(return_value=response)
monkeypatch.setattr(AsyncHTTPHandler, "post", mock_post)
return mock_post
def test_parallel_responses_gateway_route(client, auth_as, monkeypatch):
"""The primary LLM route reaches Parallel's native Responses endpoint."""
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
monkeypatch.setattr(proxy_server, "llm_router", _parallel_router())
mock_post = _mock_async_post(
monkeypatch,
url=PARALLEL_RESPONSES_URL,
response_body=_parallel_response_body(),
)
response = client.post(
"/v1/responses",
json={"model": "parallel-gateway", "input": "Research Parallel"},
)
assert response.status_code == 200, response.text
assert response.json()["output"][0]["content"][0]["text"] == "Parallel grounded answer"
request_kwargs = mock_post.await_args.kwargs
assert request_kwargs["url"] == PARALLEL_RESPONSES_URL
assert request_kwargs["headers"]["Authorization"] == "Bearer parallel-responses-key"
assert request_kwargs["json"] == {
"model": "parallel",
"input": "Research Parallel",
}