"""
OpenAI provider implementation for Praval framework.
Provides integration with OpenAI's Chat Completions API with support
for conversation history, tool calling, and streaming responses.
"""
import json
import os
import re
import threading
from collections import OrderedDict
from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
import openai
from .._metering_runtime import provider_request
from ..core.exceptions import (
HITLConfigurationError,
InterventionRequired,
ProviderError,
)
from ..hitl.runtime import HITLRuntime
from ..model_runtime import (
_nested_unsafe_option_keys,
call_with_retries,
execute_legacy_tool_call,
max_provider_retries,
)
from ..models import (
AudioResponse,
ContentPart,
ModelEvent,
ModelRequest,
ModelResponse,
ProviderCapabilities,
SpeechRequest,
ToolCall,
ToolResult,
ToolSpec,
TranscriptionRequest,
Usage,
)
from .errors import map_provider_exception, sdk_max_retries
from .registry import reasoning_parameters
from .usage import openai_usage
def _redact_secrets(message: str) -> str:
if not message:
return message
secrets = [
os.getenv("OPENAI_API_KEY"),
os.getenv("ANTHROPIC_API_KEY"),
os.getenv("COHERE_API_KEY"),
]
redacted = message
for secret in secrets:
if secret and secret in redacted:
redacted = redacted.replace(secret, "***")
return redacted
[docs]
class OpenAIProvider:
"""
OpenAI provider for LLM interactions.
Handles communication with OpenAI's GPT models through the
Chat Completions API with support for tools and conversation history.
"""
provider_name = "openai"
capabilities = ProviderCapabilities(
chat_completions=True,
responses_api=True,
tools=True,
streaming=True,
native_streaming=True,
tool_streaming=True,
structured_outputs=True,
json_schema_mode="json_schema",
multimodal=True,
image_input=True,
audio_transcription=True,
speech_generation=True,
reasoning=True,
reasoning_effort=True,
embeddings=True,
)
[docs]
def __init__(self, config):
"""
Initialize OpenAI provider.
Args:
config: AgentConfig object with provider settings
Raises:
ProviderError: If OpenAI client initialization fails
"""
self.config = config
self._parameter_lock = threading.RLock()
self._parameter_policies: OrderedDict[
Tuple[str, str], Dict[str, Optional[str]]
] = OrderedDict()
try:
api_key_env = getattr(config, "api_key_env", None) or "OPENAI_API_KEY"
api_key = os.getenv(api_key_env)
if not api_key:
raise ProviderError(f"{api_key_env} environment variable not set")
client_kwargs: Dict[str, Any] = {"api_key": api_key}
if getattr(config, "base_url", None):
client_kwargs["base_url"] = config.base_url
if getattr(config, "timeout", None):
client_kwargs["timeout"] = config.timeout
client_kwargs["max_retries"] = sdk_max_retries(config)
self.client = openai.OpenAI(**client_kwargs)
except Exception as e:
raise ProviderError(
f"Failed to initialize OpenAI client: {self._redact(str(e))}"
) from e
[docs]
def generate(
self,
messages: List[Dict[str, str]],
tools: Optional[List[Dict[str, Any]]] = None,
hitl_context: Optional[Dict[str, Any]] = None,
) -> str:
"""
Generate a response using OpenAI's Chat Completions API.
Args:
messages: Conversation history as list of message dictionaries
tools: Optional list of available tools for function calling
hitl_context: Optional run metadata for HITL gating/resume
Returns:
Generated response as a string
Raises:
ProviderError: If API call fails
"""
try:
call_params = self._base_chat_completion_params(
model=self._model_name(),
messages=messages,
temperature=self.config.temperature,
max_output_tokens=self._max_output_tokens(),
)
if tools:
formatted_tools = self._format_tools_for_openai(tools)
if formatted_tools:
call_params["tools"] = formatted_tools
call_params["tool_choice"] = "auto"
self._apply_chat_completion_model_constraints(call_params)
response = self._create_chat_completion_with_empty_retry(call_params)
if response.choices and response.choices[0].message:
message = response.choices[0].message
if hasattr(message, "tool_calls") and message.tool_calls:
return self._handle_tool_calls(
message.tool_calls,
tools,
messages,
hitl_context=hitl_context,
)
return self._extract_chat_message_text(message)
return ""
except (InterventionRequired, HITLConfigurationError):
raise
except Exception as e:
raise self._mapped_error(
e, f"OpenAI API error: {self._redact(str(e))}"
) from e
[docs]
def invoke(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> ModelResponse:
"""Invoke OpenAI through the provider-neutral adapter surface."""
if self._use_responses_api(request):
return self._invoke_responses(request)
return self._invoke_chat_completions(request, tools=tools)
[docs]
def stream(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> Iterator[ModelEvent]:
"""Stream OpenAI responses as provider-neutral events."""
try:
if self._use_responses_api(request):
yield from self._stream_responses(request)
else:
yield from self._stream_chat_completions(request, tools=tools)
except Exception as e:
message = self._redact(str(e))
yield ModelEvent(type="error", metadata={"message": message})
raise self._mapped_error(e, f"OpenAI streaming error: {message}") from e
def _redact(self, message: str) -> str:
"""Redact known provider keys and the key from ``config.api_key_env``."""
redacted = _redact_secrets(message)
api_key_env = getattr(self.config, "api_key_env", None)
secret = os.getenv(api_key_env) if api_key_env else None
if secret and redacted:
redacted = redacted.replace(secret, "***")
return redacted
[docs]
def map_provider_error(self, exc: BaseException) -> ProviderError:
"""Map an OpenAI SDK exception to a typed, redacted ``ProviderError``."""
return self._mapped_error(exc, None)
def _mapped_error(
self, exc: BaseException, message: Optional[str]
) -> ProviderError:
return map_provider_exception(
exc,
provider=self.provider_name,
model=self._model_name(),
message=message,
redact=self._redact,
)
[docs]
def transcribe(self, request: TranscriptionRequest) -> AudioResponse:
"""Transcribe request-based audio through OpenAI's audio endpoint."""
if not isinstance(request, TranscriptionRequest):
request = TranscriptionRequest.model_validate(request)
file_value, should_close = self._audio_file_value(request)
try:
call_params: Dict[str, Any] = {
"file": file_value,
"model": request.model
or self._provider_config_option(
"transcription_model", "gpt-4o-transcribe"
),
"response_format": request.response_format,
}
if request.language:
call_params["language"] = request.language
if request.prompt:
call_params["prompt"] = request.prompt
if request.temperature is not None:
call_params["temperature"] = request.temperature
if request.timeout is not None:
call_params["timeout"] = request.timeout
self._apply_audio_provider_options(call_params, request.provider_options)
# Praval owns retries (SDK retries are off). An upload can only be
# resent when its stream can be rewound to where it started.
rewind = self._audio_rewind(file_value)
def create_transcription() -> Any:
if rewind is not None:
rewind()
return self.client.audio.transcriptions.create(**call_params)
response = call_with_retries(
"transcribe",
create_transcription,
retries=max_provider_retries(self.config) if rewind else 0,
map_error=lambda exc: self._mapped_error(
exc, f"OpenAI transcription error: {self._redact(str(exc))}"
),
)
text = self._transcription_text(response)
if not text:
raise ProviderError("OpenAI transcription returned no text")
return AudioResponse(
text=text,
provider=self.provider_name,
model=str(call_params["model"]),
format=request.response_format,
raw=response,
metadata=dict(request.metadata),
)
except ProviderError:
raise
except Exception as e:
raise self._mapped_error(
e, f"OpenAI transcription error: {self._redact(str(e))}"
) from e
finally:
if should_close:
file_value.close()
[docs]
def speak(self, request: SpeechRequest) -> AudioResponse:
"""Synthesize request-based speech through OpenAI's audio endpoint."""
if not isinstance(request, SpeechRequest):
request = SpeechRequest.model_validate(request)
if not request.input.strip():
raise ProviderError("Speech input cannot be empty")
if not 0.25 <= request.speed <= 4.0:
raise ProviderError("Speech speed must be between 0.25 and 4.0")
try:
model = request.model or self._provider_config_option(
"speech_model", "tts-1"
)
if request.instructions and model in {"tts-1", "tts-1-hd"}:
raise ProviderError(
f"Speech instructions are not supported by model '{model}'"
)
call_params: Dict[str, Any] = {
"input": request.input,
"model": model,
"voice": request.voice,
"response_format": request.response_format,
"speed": request.speed,
}
if request.instructions:
call_params["instructions"] = request.instructions
if request.timeout is not None:
call_params["timeout"] = request.timeout
self._apply_audio_provider_options(call_params, request.provider_options)
response = call_with_retries(
"speak",
lambda: self.client.audio.speech.create(**call_params),
retries=max_provider_retries(self.config),
map_error=lambda exc: self._mapped_error(
exc, f"OpenAI speech generation error: {self._redact(str(exc))}"
),
)
data = self._speech_bytes(response)
if not data:
raise ProviderError("OpenAI speech generation returned no audio")
return AudioResponse(
data=data,
provider=self.provider_name,
model=str(call_params["model"]),
format=request.response_format,
mime_type=self._speech_mime_type(request.response_format),
raw=response,
metadata=dict(request.metadata),
)
except ProviderError:
raise
except Exception as e:
raise self._mapped_error(
e, f"OpenAI speech generation error: {self._redact(str(e))}"
) from e
[docs]
def close(self) -> None:
"""Close the underlying SDK client when supported."""
close = getattr(self.client, "close", None)
if callable(close):
close()
@staticmethod
def _audio_rewind(file_value: Any) -> Optional[Callable[[], None]]:
"""Return a callable restoring the upload's start, or ``None``.
Bytes need no rewinding. A file object (alone or inside an SDK file
tuple) is rewound to its current position; one that cannot seek
cannot be resent, so it gets ``None`` and is not retried.
"""
stream = file_value
if isinstance(file_value, tuple):
stream = file_value[1] if len(file_value) > 1 else None
if not hasattr(stream, "read"):
return lambda: None
try:
if not stream.seekable():
return None
position = stream.tell()
except (AttributeError, OSError, ValueError):
return None
def rewind() -> None:
stream.seek(position)
return rewind
def _audio_file_value(self, request: TranscriptionRequest) -> Tuple[Any, bool]:
audio = request.audio
if isinstance(audio, (bytes, bytearray, memoryview)):
data = bytes(audio)
if not data:
raise ProviderError("Audio input cannot be empty")
return (
(
request.filename or "audio.wav",
data,
request.mime_type or "audio/wav",
),
False,
)
if isinstance(audio, (str, os.PathLike)):
path = os.fspath(audio)
if not os.path.isfile(path):
raise ProviderError(f"Audio file does not exist: {path}")
return open(path, "rb"), True
if audio is None:
raise ProviderError("Audio input cannot be empty")
if hasattr(audio, "read") or isinstance(audio, tuple):
return audio, False
raise ProviderError(
"Audio input must be bytes, a file path, a file object, "
"or an SDK file tuple"
)
def _provider_config_option(self, name: str, default: str) -> str:
options = getattr(self.config, "provider_options", None) or {}
return str(options.get(name) or default)
def _apply_audio_provider_options(
self,
call_params: Dict[str, Any],
provider_options: Dict[str, Any],
) -> None:
if provider_options.get("stream"):
raise ProviderError(
"Streaming audio is outside the request-based voice API; "
"use a realtime or streaming adapter instead"
)
unsafe = _nested_unsafe_option_keys(provider_options)
if unsafe:
blocked = ", ".join(sorted(set(unsafe)))
raise ProviderError(f"Unsafe provider option(s): {blocked}")
reserved = {
"file",
"input",
"model",
"voice",
"response_format",
"speed",
"instructions",
"language",
"prompt",
"temperature",
"timeout",
"stream",
"transcription_model",
"speech_model",
}
for key, value in provider_options.items():
if key not in reserved:
call_params.setdefault(key, value)
def _transcription_text(self, response: Any) -> str:
if isinstance(response, str):
return response
if isinstance(response, dict):
return str(response.get("text") or "")
value = getattr(response, "text", None)
return value if isinstance(value, str) else ""
def _speech_bytes(self, response: Any) -> bytes:
if isinstance(response, bytes):
return response
if isinstance(response, bytearray):
return bytes(response)
content = getattr(response, "content", None)
if isinstance(content, bytes):
return content
if isinstance(content, bytearray):
return bytes(content)
read = getattr(response, "read", None)
if callable(read):
data = read()
if isinstance(data, bytes):
return data
if isinstance(data, bytearray):
return bytes(data)
return b""
def _speech_mime_type(self, response_format: str) -> str:
return {
"mp3": "audio/mpeg",
"opus": "audio/opus",
"aac": "audio/aac",
"flac": "audio/flac",
"wav": "audio/wav",
"pcm": "audio/pcm",
}.get(response_format.lower(), "application/octet-stream")
def _model_name(self) -> str:
return str(getattr(self.config, "model", None) or "gpt-5.4-mini")
def _max_output_tokens(self) -> int:
return int(
getattr(self.config, "max_output_tokens", None)
or getattr(self.config, "max_tokens", 1000)
)
def _base_chat_completion_params(
self,
*,
model: str,
messages: List[Dict[str, Any]],
temperature: Optional[float],
max_output_tokens: int,
) -> Dict[str, Any]:
call_params: Dict[str, Any] = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_output_tokens,
}
if temperature is None:
call_params.pop("temperature", None)
self._apply_chat_completion_model_constraints(call_params)
return call_params
def _apply_chat_completion_model_constraints(
self,
call_params: Dict[str, Any],
) -> None:
model = str(call_params.get("model") or "")
if not self._uses_max_completion_tokens(model):
return
if "max_completion_tokens" not in call_params:
max_tokens = call_params.pop("max_tokens", None)
if max_tokens is not None:
call_params["max_completion_tokens"] = max_tokens
else:
call_params.pop("max_tokens", None)
for key in ("temperature", "top_p", "top_logprobs", "logprobs"):
call_params.pop(key, None)
if self._gpt_generation(model) >= 6 and call_params.get("tools"):
normalized = self._normalized_model_name(model)
supports_nonreasoning_tools = any(
normalized == name
or re.fullmatch(re.escape(name) + r"-\d{4}-?\d{2}-?\d{2}", normalized)
for name in ("gpt-6-luna", "gpt-6-sol")
)
if (
not supports_nonreasoning_tools
or call_params.get("reasoning_effort", "none") != "none"
):
raise ProviderError(
f"OpenAI model '{model}' requires the Responses API for "
"this tool request; set provider_options={'endpoint': 'responses'}"
)
call_params.setdefault("reasoning_effort", "none")
def _normalized_model_name(self, model: str) -> str:
normalized = model.strip().lower().replace("_", "-")
if ":" in normalized:
normalized = normalized.rsplit(":", 1)[-1]
if "/" in normalized:
normalized = normalized.rsplit("/", 1)[-1]
return normalized
def _gpt_generation(self, model: str) -> int:
match = re.match(r"gpt-(\d+)(?=[.-]|$)", self._normalized_model_name(model))
return int(match.group(1)) if match else 0
def _uses_max_completion_tokens(self, model: str) -> bool:
normalized = self._normalized_model_name(model)
return self._gpt_generation(model) >= 5 or (
normalized.startswith("o")
and len(normalized) > 1
and normalized[1].isdigit()
)
def _create_chat_completion_with_empty_retry(
self,
call_params: Dict[str, Any],
) -> Any:
response = self._create_model_request("chat.completions", call_params)
retry_params = self._empty_text_retry_params(call_params, response)
if retry_params is None:
return response
return self._create_model_request("chat.completions", retry_params)
def _parameter_repair(
self, exc: BaseException, params: Dict[str, Any]
) -> Optional[Dict[str, Optional[str]]]:
"""Repair only explicit sampling rejections or the token-limit rename."""
options = getattr(self.config, "provider_options", {}) or {}
if options.get("parameter_recovery", True) is not True:
return None
if getattr(exc, "status_code", None) != 400:
return None
body = getattr(exc, "body", None)
error = body.get("error", body) if isinstance(body, dict) else {}
if not isinstance(error, dict):
error = {}
parameter = error.get("param") or getattr(exc, "param", None)
message = str(error.get("message") or str(exc))[:2000]
code = error.get("code") or getattr(exc, "code", None)
if code not in {"unsupported_parameter", "unsupported_value"} and not re.search(
r"unsupported (?:parameter|value)", message, re.IGNORECASE
):
return None
if not isinstance(parameter, str):
match = re.search(
r"(?:parameter|value):?\s*['\"]([^'\"]+)['\"]", message, re.IGNORECASE
)
parameter = match.group(1) if match else None
if parameter not in params:
return None
if parameter in {"temperature", "top_p", "top_logprobs", "logprobs"}:
return {parameter: None}
if parameter == "max_tokens" and "max_completion_tokens" in message:
return {"max_tokens": "max_completion_tokens"}
return None
@staticmethod
def _apply_parameter_policy(
params: Dict[str, Any], policy: Dict[str, Optional[str]]
) -> None:
for parameter, replacement in policy.items():
if parameter in params:
value = params.pop(parameter)
if replacement is not None:
params.setdefault(replacement, value)
def _model_request_params(
self, endpoint: str, params: Dict[str, Any]
) -> Dict[str, Any]:
prepared = dict(params)
enabled = prepared.pop("_praval_parameter_recovery", True)
options = getattr(self.config, "provider_options", {}) or {}
if enabled is not True or options.get("parameter_recovery", True) is not True:
return prepared
key = (endpoint, str(params.get("model") or ""))
with self._parameter_lock:
self._apply_parameter_policy(
prepared, self._parameter_policies.get(key, {})
)
return prepared
def _remember_parameter_policy(
self, endpoint: str, params: Dict[str, Any], policy: Dict[str, Optional[str]]
) -> None:
key = (endpoint, str(params.get("model") or ""))
with self._parameter_lock:
self._parameter_policies.setdefault(key, {}).update(policy)
self._parameter_policies.move_to_end(key)
while len(self._parameter_policies) > 128:
self._parameter_policies.popitem(last=False)
def _create_model_request(self, endpoint: str, params: Dict[str, Any]) -> Any:
"""Negotiate once per rejected request, preserving tools and metering."""
prepared = self._model_request_params(endpoint, params)
create = (
self.client.responses.create
if endpoint == "responses"
else self.client.chat.completions.create
)
policy: Optional[Dict[str, Optional[str]]] = None
for attempt in range(2):
try:
with provider_request() as report:
response = create(**prepared)
if report is not None:
report.finish(self._request_usage_response(response))
if policy is not None:
self._remember_parameter_policy(endpoint, params, policy)
return response
except Exception as exc:
policy = (
self._parameter_repair(exc, prepared)
if attempt == 0
and params.get("_praval_parameter_recovery", True) is True
else None
)
if policy is None:
raise
self._apply_parameter_policy(prepared, policy)
raise AssertionError("Unreachable parameter recovery state")
def _request_usage_response(self, response: Any) -> ModelResponse:
return ModelResponse(usage=self._extract_usage(response))
def _stream_model_request(
self, endpoint: str, params: Dict[str, Any]
) -> Iterator[Any]:
"""Recover only before any stream data is emitted, and meter each request."""
prepared = self._model_request_params(endpoint, params)
create = (
self.client.responses.create
if endpoint == "responses"
else self.client.chat.completions.create
)
policy: Optional[Dict[str, Optional[str]]] = None
for attempt in range(2):
opened = False
try:
with provider_request() as report:
chunks = create(**prepared)
opened = True
if policy is not None:
self._remember_parameter_policy(endpoint, params, policy)
usage_response = ModelResponse()
try:
for chunk in chunks:
raw = self._event_value(chunk, "response", None) or chunk
observed = self._request_usage_response(raw)
if observed.usage is not None:
usage_response = observed
yield chunk
except BaseException:
if report is not None:
report.finish(usage_response, status="error")
raise
finally:
close = getattr(chunks, "close", None)
if callable(close):
close()
if report is not None:
report.finish(usage_response)
return
except Exception as exc:
policy = (
self._parameter_repair(exc, prepared)
if attempt == 0
and not opened
and params.get("_praval_parameter_recovery", True) is True
else None
)
if policy is None:
raise
self._apply_parameter_policy(prepared, policy)
def _empty_text_retry_params(
self,
call_params: Dict[str, Any],
response: Any,
) -> Optional[Dict[str, Any]]:
model = str(call_params.get("model") or "")
if not self._uses_max_completion_tokens(model):
return None
finish_reason = self._chat_choice_finish_reason(response)
if finish_reason == "content_filter":
return None
choice = self._first_chat_choice(response)
message = self._chat_choice_message(choice)
if message is not None:
tool_calls = self._event_value(message, "tool_calls", None)
if tool_calls:
return None
if self._extract_chat_message_text(message).strip():
return None
current_tokens = self._chat_completion_token_budget(call_params)
if current_tokens >= 4096:
return None
retry_params = dict(call_params)
retry_params["max_completion_tokens"] = max(current_tokens * 4, 4096)
retry_params.pop("max_tokens", None)
retry_params.pop("temperature", None)
return retry_params
def _chat_completion_token_budget(self, call_params: Dict[str, Any]) -> int:
raw_tokens = call_params.get(
"max_completion_tokens",
call_params.get("max_tokens", self._max_output_tokens()),
)
try:
return int(raw_tokens or self._max_output_tokens())
except (TypeError, ValueError):
return self._max_output_tokens()
def _first_chat_choice(self, response: Any) -> Any:
choices = getattr(response, "choices", None)
if choices is None and isinstance(response, dict):
choices = response.get("choices")
if not choices:
return None
return choices[0]
def _chat_choice_message(self, choice: Any) -> Any:
if choice is None:
return None
return self._event_value(choice, "message", None)
def _chat_choice_finish_reason(self, response: Any) -> Optional[str]:
choice = self._first_chat_choice(response)
if choice is None:
return None
value = self._event_value(choice, "finish_reason", None)
return value if isinstance(value, str) else None
def _extract_chat_message_text(self, message: Any) -> str:
text = self._extract_content_text(self._event_value(message, "content", None))
if text:
return text
return self._extract_content_text(self._event_value(message, "refusal", None))
def _extract_content_text(self, content: Any) -> str:
if content is None:
return ""
if type(content).__module__.startswith("unittest.mock"):
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
return "".join(self._extract_content_part_text(part) for part in content)
return str(content)
def _extract_content_part_text(self, part: Any) -> str:
if part is None:
return ""
if isinstance(part, str):
return part
for key in ("text", "content", "refusal"):
value = (
part.get(key) if isinstance(part, dict) else getattr(part, key, None)
)
if value is not None:
return self._extract_content_text(value)
return ""
def _use_responses_api(self, request: ModelRequest) -> bool:
endpoint = str(
request.provider_options.get("endpoint")
or request.provider_options.get("api")
or ""
).lower()
if endpoint in {"chat.completions", "chat", "chat_completions"}:
return False
return (
endpoint == "responses"
or bool(request.provider_options.get("use_responses", False))
or request.reasoning is not None
or (
self._gpt_generation(request.model or self._model_name()) >= 6
and bool(request.tools)
)
)
def _invoke_chat_completions(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> ModelResponse:
call_params = self._chat_completion_params(request, tools=tools)
response = self._create_chat_completion_with_empty_retry(call_params)
return self._chat_model_response(response, call_params)
def _chat_model_response(
self,
response: Any,
call_params: Dict[str, Any],
) -> ModelResponse:
content = ""
tool_calls: List[ToolCall] = []
assistant_tool_calls: List[Dict[str, Any]] = []
choice = self._first_chat_choice(response)
message = self._chat_choice_message(choice)
if message is not None:
content = self._extract_chat_message_text(message)
raw_tool_calls = self._event_value(message, "tool_calls", None) or []
if not isinstance(raw_tool_calls, (list, tuple)):
raw_tool_calls = []
assistant_tool_calls = self._serialize_tool_calls(list(raw_tool_calls))
tool_calls = [
self._chat_tool_call(tool_call) for tool_call in assistant_tool_calls
]
metadata: Dict[str, Any] = {
"openai_endpoint": "chat.completions",
"assistant_tool_calls": assistant_tool_calls,
}
if assistant_tool_calls:
# The cumulative transcript lets every continuation round resend
# all earlier tool turns, not only the original request.
metadata["openai_chat_messages"] = [
*call_params["messages"],
self._chat_assistant_message(content, assistant_tool_calls),
]
return ModelResponse(
content=content,
provider=self.provider_name,
model=call_params["model"],
tool_calls=tool_calls,
raw=response,
usage=self._extract_usage(response),
finish_reason=self._chat_choice_finish_reason(response),
metadata=metadata,
)
def _chat_assistant_message(
self,
content: str,
assistant_tool_calls: List[Dict[str, Any]],
) -> Dict[str, Any]:
return {
"role": "assistant",
"content": content or None,
"tool_calls": assistant_tool_calls,
}
def _chat_tool_call(self, tool_call: Dict[str, Any]) -> ToolCall:
function = tool_call.get("function") or {}
raw_arguments = function.get("arguments") or "{}"
try:
arguments = json.loads(raw_arguments)
except (TypeError, json.JSONDecodeError, RecursionError):
arguments = {"raw": raw_arguments}
return ToolCall(
id=str(tool_call.get("id") or ""),
name=str(function.get("name") or ""),
arguments=arguments if isinstance(arguments, dict) else {"raw": arguments},
raw=tool_call,
)
def _continue_chat_completions(
self,
request: ModelRequest,
response: ModelResponse,
tool_results: List[ToolResult],
) -> ModelResponse:
assistant_tool_calls = response.metadata.get("assistant_tool_calls")
if not isinstance(assistant_tool_calls, list):
raise ProviderError("OpenAI Chat Completions tool state is missing")
call_params = self._chat_completion_params(request)
transcript = response.metadata.get("openai_chat_messages")
if isinstance(transcript, list):
messages = list(transcript)
else:
# State written by v0.8.3 has no transcript; rebuild one round.
messages = list(call_params["messages"])
messages.append(
self._chat_assistant_message(response.content, assistant_tool_calls)
)
messages.extend(
{
"role": "tool",
"tool_call_id": result.tool_call_id,
"content": result.content,
}
for result in tool_results
)
call_params["messages"] = messages
continued = self._create_chat_completion_with_empty_retry(call_params)
return self._chat_model_response(continued, call_params)
def _continue_responses(
self,
request: ModelRequest,
response: ModelResponse,
tool_results: List[ToolResult],
) -> ModelResponse:
response_id = response.metadata.get("response_id")
if not response_id:
raise ProviderError("OpenAI Responses tool state is missing response_id")
call_params = self._responses_params(request)
call_params["previous_response_id"] = response_id
call_params["input"] = [
{
"type": "function_call_output",
"call_id": result.tool_call_id,
"output": result.content,
}
for result in tool_results
]
continued = self._create_model_request("responses", call_params)
return self._responses_model_response(continued, call_params)
def _invoke_responses(self, request: ModelRequest) -> ModelResponse:
call_params = self._responses_params(request)
response = self._create_model_request("responses", call_params)
return self._responses_model_response(response, call_params)
def _responses_model_response(
self,
response: Any,
call_params: Dict[str, Any],
) -> ModelResponse:
output = self._event_value(response, "output", None) or []
if not isinstance(output, (list, tuple)):
output = []
tool_calls = [
self._tool_call(item)
for item in output
if self._event_value(item, "type", "") == "function_call"
]
return ModelResponse(
content=self._extract_responses_text(response),
provider=self.provider_name,
model=call_params["model"],
tool_calls=tool_calls,
raw=response,
usage=self._extract_usage(response),
metadata={
"openai_endpoint": "responses",
"response_id": self._event_value(response, "id", None),
},
)
def _chat_completion_params(
self,
request: ModelRequest,
*,
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False,
) -> Dict[str, Any]:
if request.provider_options.get("experimental_tools") is not None:
raise ProviderError(
"OpenAI experimental tools require the Responses API endpoint"
)
call_params = self._base_chat_completion_params(
model=request.model or self._model_name(),
messages=[
self._format_message_for_openai(message, responses=False)
for message in request.messages
],
temperature=request.temperature,
max_output_tokens=request.max_output_tokens or self._max_output_tokens(),
)
if tools:
formatted_tools = self._format_tools_for_openai(tools)
if formatted_tools:
call_params["tools"] = formatted_tools
call_params["tool_choice"] = "auto"
elif request.tools:
formatted_tool_specs = self._format_tool_specs_for_openai(request.tools)
if formatted_tool_specs:
call_params["tools"] = formatted_tool_specs
call_params["tool_choice"] = "auto"
if request.response_schema is not None:
call_params["response_format"] = self._openai_response_format(
request.response_schema
)
native_reasoning = reasoning_parameters(request)
effort = native_reasoning.get("effort")
if request.reasoning is not None and request.reasoning.effort is not None:
effort = request.reasoning.effort
if effort is not None:
call_params["reasoning_effort"] = effort
if request.timeout is not None:
call_params["timeout"] = request.timeout
if stream:
call_params["stream"] = True
if request.stream_options:
call_params["stream_options"] = request.stream_options
self._apply_provider_options(call_params, request)
self._apply_chat_completion_model_constraints(call_params)
return call_params
def _responses_params(
self,
request: ModelRequest,
*,
stream: bool = False,
) -> Dict[str, Any]:
call_params: Dict[str, Any] = {
"model": request.model or self._model_name(),
"input": self._responses_input(request),
"temperature": request.temperature,
"max_output_tokens": request.max_output_tokens or self._max_output_tokens(),
}
if request.temperature is None:
call_params.pop("temperature", None)
formatted_tools = self._format_tool_specs_for_responses(request.tools)
formatted_tools.extend(self._experimental_tools(request))
if formatted_tools:
call_params["tools"] = formatted_tools
if request.response_schema is not None:
call_params["text"] = {
"format": self._openai_text_format(request.response_schema)
}
if request.reasoning is not None:
reasoning: Dict[str, Any] = dict(reasoning_parameters(request))
if request.reasoning.effort:
reasoning["effort"] = request.reasoning.effort
if request.reasoning.summary:
reasoning["summary"] = request.reasoning.summary
if request.reasoning.budget_tokens is not None:
reasoning["budget_tokens"] = request.reasoning.budget_tokens
if reasoning:
call_params["reasoning"] = reasoning
if request.timeout is not None:
call_params["timeout"] = request.timeout
if stream:
call_params["stream"] = True
if request.stream_options:
call_params["stream_options"] = request.stream_options
self._apply_provider_options(call_params, request)
if self._uses_max_completion_tokens(str(call_params["model"])) or (
request.reasoning is not None
and (request.reasoning.level is not None or request.reasoning.effort)
):
for key in ("temperature", "top_p", "top_logprobs"):
call_params.pop(key, None)
return call_params
def _apply_provider_options(
self,
call_params: Dict[str, Any],
request: ModelRequest,
) -> None:
reserved = {
"endpoint",
"api",
# A model profile uses this only to select an OpenAI-compatible
# local endpoint. It is not a Chat Completions API argument.
"local_preset",
"use_responses",
"capabilities",
"transcription_model",
"speech_model",
"allow_experimental_tools",
"experimental_tools",
# Client construction option (SDK-internal retries), not a request
# argument.
"max_retries",
"parameter_recovery",
"discover_model",
"required_context_tokens",
}
for key, value in request.provider_options.items():
if key not in reserved:
call_params.setdefault(key, value)
if "parameter_recovery" in request.provider_options:
recovery = request.provider_options["parameter_recovery"]
if not isinstance(recovery, bool):
raise ProviderError("parameter_recovery must be a bool")
call_params["_praval_parameter_recovery"] = recovery
def _experimental_tools(self, request: ModelRequest) -> List[Dict[str, Any]]:
value = request.provider_options.get("experimental_tools")
if value is None:
return []
if request.provider_options.get("allow_experimental_tools") is not True:
raise ProviderError(
"experimental_tools requires allow_experimental_tools=True"
)
if not isinstance(value, list) or not all(
isinstance(tool, dict) for tool in value
):
raise ProviderError("experimental_tools must be a list of tool mappings")
return [dict(tool) for tool in value]
def _responses_input(self, request: ModelRequest) -> List[Dict[str, Any]]:
return [
self._format_message_for_openai(message, responses=True)
for message in request.messages
]
def _format_message_for_openai(
self,
message: Any,
*,
responses: bool,
) -> Dict[str, Any]:
content = self._format_openai_content(message.content, responses=responses)
formatted: Dict[str, Any] = {"role": message.role, "content": content}
if getattr(message, "name", None):
formatted["name"] = message.name
if getattr(message, "tool_call_id", None):
formatted["tool_call_id"] = message.tool_call_id
return formatted
def _format_openai_content(self, content: Any, *, responses: bool) -> Any:
if not isinstance(content, list):
return content
formatted_parts: List[Dict[str, Any]] = []
for item in content:
part = item if isinstance(item, ContentPart) else ContentPart(**item)
if part.type == "text":
if responses:
formatted_parts.append(
{"type": "input_text", "text": part.text or ""}
)
else:
formatted_parts.append({"type": "text", "text": part.text or ""})
elif part.type in {"image_url", "image_base64", "image"}:
image_url = part.url or ""
if part.type == "image_base64":
image_url = (
f"data:{part.mime_type or 'image/png'};base64,{part.data or ''}"
)
if responses:
formatted_parts.append(
{"type": "input_image", "image_url": image_url}
)
else:
formatted_parts.append(
{"type": "image_url", "image_url": {"url": image_url}}
)
else:
raise ProviderError(
f"OpenAI provider cannot serialize content part type: {part.type}"
)
return formatted_parts
def _stream_chat_completions(
self,
request: ModelRequest,
*,
tools: Optional[List[Dict[str, Any]]] = None,
) -> Iterator[ModelEvent]:
call_params = self._chat_completion_params(
request,
tools=tools,
stream=True,
)
chunks = self._stream_model_request("chat.completions", call_params)
content_parts: List[str] = []
request_metadata: Dict[str, Any] = {}
usage: Optional[Usage] = None
finish_reason: Optional[str] = None
for chunk in chunks:
request_metadata.update(self._request_usage_response(chunk).metadata)
usage = self._extract_usage(chunk) or usage
choices = getattr(chunk, "choices", None) or []
if not choices and isinstance(chunk, dict):
choices = chunk.get("choices") or []
if not choices:
continue
choice = choices[0]
delta = self._event_value(choice, "delta", {})
text = self._event_value(delta, "content", None)
if text:
content_parts.append(str(text))
yield ModelEvent(type="delta", delta=str(text))
tool_call_deltas = self._event_value(delta, "tool_calls", None)
if tool_call_deltas:
yield ModelEvent(
type="tool_call_delta",
metadata={"delta": self._serialize_event_value(tool_call_deltas)},
)
finish_reason = self._event_value(choice, "finish_reason", finish_reason)
response = ModelResponse(
content="".join(content_parts),
provider=self.provider_name,
model=call_params["model"],
usage=usage,
finish_reason=finish_reason,
metadata=request_metadata,
)
if usage:
yield ModelEvent(type="usage", usage=usage)
yield ModelEvent(type="final", response=response, usage=usage)
def _stream_responses(self, request: ModelRequest) -> Iterator[ModelEvent]:
call_params = self._responses_params(request, stream=True)
events = self._stream_model_request("responses", call_params)
content_parts: List[str] = []
final_response: Optional[ModelResponse] = None
usage: Optional[Usage] = None
for event in events:
event_type = str(self._event_value(event, "type", ""))
if event_type in {"response.output_text.delta", "output_text.delta"}:
delta = self._event_value(event, "delta", "")
if delta:
content_parts.append(str(delta))
yield ModelEvent(type="delta", delta=str(delta))
elif event_type.endswith("function_call_arguments.delta"):
yield ModelEvent(
type="tool_call_delta",
metadata={"delta": self._event_value(event, "delta", "")},
)
elif event_type in {"response.output_item.done", "output_item.done"}:
item = self._event_value(event, "item", {})
if self._event_value(item, "type", "") == "function_call":
yield ModelEvent(type="tool_call", tool_call=self._tool_call(item))
elif event_type in {"response.completed", "response.done"}:
raw_response = self._event_value(event, "response", event)
usage = self._extract_usage(raw_response) or usage
final_response = ModelResponse(
content=self._extract_responses_text(raw_response)
or "".join(content_parts),
provider=self.provider_name,
model=call_params["model"],
raw=raw_response,
usage=usage,
)
elif event_type in {"response.failed", "response.error", "error"}:
error = self._event_value(event, "error", event)
yield ModelEvent(
type="error",
metadata={"message": self._redact(str(error))},
)
if final_response is None:
final_response = ModelResponse(
content="".join(content_parts),
provider=self.provider_name,
model=call_params["model"],
usage=usage,
)
if final_response.usage:
yield ModelEvent(type="usage", usage=final_response.usage)
yield ModelEvent(
type="final",
response=final_response,
usage=final_response.usage,
)
def _format_tool_specs_for_openai(
self, tool_specs: List[ToolSpec]
) -> List[Dict[str, Any]]:
formatted_tools = []
for spec in tool_specs:
function_def: Dict[str, Any] = {
"name": spec.name,
"description": spec.description,
"parameters": spec.parameters,
}
if spec.strict:
function_def["strict"] = True
formatted_tools.append({"type": "function", "function": function_def})
return formatted_tools
def _format_tool_specs_for_responses(
self, tool_specs: List[ToolSpec]
) -> List[Dict[str, Any]]:
"""Format client functions for the OpenAI Responses API."""
formatted_tools = []
for spec in tool_specs:
function_def: Dict[str, Any] = {
"type": "function",
"name": spec.name,
"description": spec.description,
"parameters": spec.parameters,
}
if spec.strict:
function_def["strict"] = True
formatted_tools.append(function_def)
return formatted_tools
def _openai_response_format(self, config: Any) -> Dict[str, Any]:
return {
"type": "json_schema",
"json_schema": {
"name": config.name or "praval_response",
"schema": config.json_schema or {},
"strict": config.strict,
},
}
def _openai_text_format(self, config: Any) -> Dict[str, Any]:
return {
"type": "json_schema",
"name": config.name or "praval_response",
"schema": config.json_schema or {},
"strict": config.strict,
}
def _extract_responses_text(self, response: Any) -> str:
output_text = getattr(response, "output_text", None)
if output_text is not None:
return str(output_text)
output = getattr(response, "output", None)
if output is None and isinstance(response, dict):
output = response.get("output")
text_parts: List[str] = []
for item in output or []:
content = getattr(item, "content", None)
if content is None and isinstance(item, dict):
content = item.get("content")
for part in content or []:
part_type = getattr(part, "type", None) or (
part.get("type") if isinstance(part, dict) else None
)
if part_type in {"output_text", "text"}:
value = getattr(part, "text", None) or (
part.get("text", "") if isinstance(part, dict) else ""
)
text_parts.append(str(value))
return "".join(text_parts)
def _extract_usage(self, response: Any) -> Optional[Usage]:
return openai_usage(response)
def _event_value(self, value: Any, key: str, default: Any = None) -> Any:
if isinstance(value, dict):
return value.get(key, default)
return getattr(value, key, default)
def _serialize_event_value(self, value: Any) -> Any:
if isinstance(value, (str, int, float, bool)) or value is None:
return value
if isinstance(value, list):
return [self._serialize_event_value(item) for item in value]
if isinstance(value, dict):
return {
str(key): self._serialize_event_value(item)
for key, item in value.items()
}
if hasattr(value, "model_dump"):
return value.model_dump(exclude_none=True)
return {
key: self._serialize_event_value(item)
for key, item in vars(value).items()
if not key.startswith("_")
}
def _tool_call(self, item: Any) -> ToolCall:
raw_args = self._event_value(item, "arguments", {}) or {}
if isinstance(raw_args, str):
try:
arguments = json.loads(raw_args)
except (json.JSONDecodeError, RecursionError):
arguments = {"raw": raw_args}
elif isinstance(raw_args, dict):
arguments = raw_args
else:
arguments = {"raw": raw_args}
return ToolCall(
id=str(
self._event_value(item, "call_id", None)
or self._event_value(item, "id", "")
),
name=str(self._event_value(item, "name", "")),
arguments=arguments,
raw=item,
)
def _format_tools_for_openai(
self, tools: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""Format tools for OpenAI's function calling format."""
formatted_tools = []
for tool in tools:
if "function" not in tool or "description" not in tool:
continue
formatted_tool = {
"type": "function",
"function": {
"name": tool["function"].__name__,
"description": tool["description"],
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
}
if "parameters" in tool:
if tool["parameters"].get("type") == "object":
# Agent.add_tool_spec already supplies JSON Schema.
formatted_tool["function"]["parameters"] = tool["parameters"]
if tool.get("strict"):
formatted_tool["function"]["strict"] = True
formatted_tools.append(formatted_tool)
continue
for param_name, param_info in tool["parameters"].items():
python_type = param_info.get("type", "str")
json_type = self._python_type_to_json_schema(python_type)
formatted_tool["function"]["parameters"]["properties"][
param_name
] = {"type": json_type}
if param_info.get("required", False):
formatted_tool["function"]["parameters"]["required"].append(
param_name
)
formatted_tools.append(formatted_tool)
return formatted_tools
def _python_type_to_json_schema(self, python_type: str) -> str:
"""Convert Python type annotation to JSON schema type."""
type_mapping = {
"str": "string",
"int": "integer",
"float": "number",
"bool": "boolean",
"list": "array",
"dict": "object",
"List": "array",
"Dict": "object",
}
return type_mapping.get(python_type, "string")
def _build_runtime(
self, hitl_context: Optional[Dict[str, Any]]
) -> Optional[HITLRuntime]:
if not hitl_context:
return None
run_id = hitl_context.get("run_id")
agent_name = hitl_context.get("agent_name")
provider_name = hitl_context.get("provider_name")
if not run_id or not agent_name or not provider_name:
return None
return HITLRuntime(
run_id=run_id,
agent_name=agent_name,
provider_name=provider_name,
hitl_enabled=bool(hitl_context.get("enabled", False)),
db_path=hitl_context.get("db_path"),
trace_id=hitl_context.get("trace_id"),
)
def _serialize_tool_calls(self, tool_calls: List[Any]) -> List[Dict[str, Any]]:
serialized: List[Dict[str, Any]] = []
for tool_call in tool_calls:
if isinstance(tool_call, dict):
call_type = tool_call.get("type")
tool_id = tool_call.get("id")
function_obj = tool_call.get("function", {})
function_name = function_obj.get("name")
arguments = function_obj.get("arguments", "{}")
else:
call_type = getattr(tool_call, "type", None)
tool_id = getattr(tool_call, "id", None)
function_obj = getattr(tool_call, "function", None)
function_name = getattr(function_obj, "name", None)
arguments = getattr(function_obj, "arguments", "{}")
if call_type == "function":
serialized.append(
{
"id": tool_id,
"type": "function",
"function": {
"name": function_name,
"arguments": arguments,
},
}
)
return serialized
def _execute_tool_calls(
self,
*,
serialized_tool_calls: List[Dict[str, Any]],
available_tools: List[Dict[str, Any]],
original_messages: List[Dict[str, str]],
hitl_context: Optional[Dict[str, Any]],
start_index: int = 0,
existing_tool_messages: Optional[List[Dict[str, str]]] = None,
resume_intervention: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, str]]:
tool_messages = list(existing_tool_messages or [])
for idx in range(start_index, len(serialized_tool_calls)):
tool_call = serialized_tool_calls[idx]
function_name = tool_call["function"]["name"]
arguments = tool_call["function"]["arguments"]
tool_call_id = tool_call["id"]
continuation_state = None
if not (idx == start_index and resume_intervention is not None):
continuation_state = {
"schema": "openai_tool_v1",
"original_messages": original_messages,
"tool_calls": serialized_tool_calls,
"current_index": idx,
"tool_messages": list(tool_messages),
}
result_content = execute_legacy_tool_call(
hitl_context=hitl_context,
tool_call_id=tool_call_id,
function_name=function_name,
raw_args=arguments,
available_tools=available_tools or [],
continuation_state=continuation_state,
resume_intervention=(
resume_intervention
if idx == start_index and resume_intervention is not None
else None
),
)
tool_messages.append(
{
"role": "tool",
"tool_call_id": tool_call_id,
"content": result_content,
}
)
return tool_messages
def _follow_up_response(
self,
*,
serialized_tool_calls: List[Dict[str, Any]],
original_messages: List[Dict[str, str]],
tool_messages: List[Dict[str, str]],
) -> str:
extended_messages = list(original_messages)
extended_messages.append(
{
"role": "assistant",
"tool_calls": serialized_tool_calls,
}
)
extended_messages.extend(tool_messages)
try:
call_params = self._base_chat_completion_params(
model=self._model_name(),
messages=extended_messages,
temperature=self.config.temperature,
max_output_tokens=self._max_output_tokens(),
)
# The tools have already run: retry only this request.
follow_up_response = call_with_retries(
"follow_up",
lambda: self._create_chat_completion_with_empty_retry(call_params),
retries=max_provider_retries(self.config),
map_error=self.map_provider_error,
)
if follow_up_response.choices and follow_up_response.choices[0].message:
return self._extract_chat_message_text(
follow_up_response.choices[0].message
)
return "No response generated after tool execution"
except Exception:
return "\n".join([msg["content"] for msg in tool_messages])
def _handle_tool_calls(
self,
tool_calls: List[Any],
available_tools: List[Dict[str, Any]],
original_messages: List[Dict[str, str]],
hitl_context: Optional[Dict[str, Any]] = None,
) -> str:
"""Handle OpenAI tool/function calls with optional HITL interception."""
serialized_tool_calls = self._serialize_tool_calls(tool_calls)
tool_messages = self._execute_tool_calls(
serialized_tool_calls=serialized_tool_calls,
available_tools=available_tools or [],
original_messages=original_messages,
hitl_context=hitl_context,
)
return self._follow_up_response(
serialized_tool_calls=serialized_tool_calls,
original_messages=original_messages,
tool_messages=tool_messages,
)