Converters API¶

@classmethod
def create_conversion_context(cls, **options: Any) -> ConversionContext:
    """Create a conversion context for non-streaming conversions.

    Args:
        **options: Initial options to populate in the context
            (e.g., ``output_format="rest"``).

    Returns:
        A new ConversionContext instance.
    """
    return ConversionContext(options=dict(options) if options else {})

@classmethod
def create_stream_context(cls) -> StreamContext:
    """Create a stream context appropriate for this converter.

    Subclasses may override to return a provider-specific context
    subclass with additional state fields.

    Returns:
        A new StreamContext instance.
    """
    return StreamContext()

def message_from_provider(
    self,
    provider_message: Any,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> Message | ExtensionItem:
    """将provider消息转换为IR格式（便利方法）
    Convert provider message to IR format (convenience method)

    Args:
        provider_message: Provider格式的消息
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        IR格式的消息
    """
    result = self.messages_from_provider(
        [provider_message], context=context, **kwargs
    )
    return result[0] if result else cast(Message, {})

def message_to_provider(
    self,
    message: Message | ExtensionItem,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[Any, list[str]]:
    """将单个消息转换为provider格式（便利方法）
    Convert single message to provider format (convenience method)

    Args:
        message: IR格式的单个消息
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        Tuple[转换后的消息, 警告信息列表]
    """
    result, warnings = self.messages_to_provider(
        [message], context=context, **kwargs
    )
    return result[0] if result else None, warnings

@abstractmethod
def messages_from_provider(
    self,
    provider_messages: list[Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> list[Message | ExtensionItem]:
    """将provider消息转换为IR消息列表
    Convert provider messages to IR message list

    Args:
        provider_messages: Provider格式的消息列表
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        IR格式的消息列表
    """
    pass

@abstractmethod
def messages_to_provider(
    self,
    messages: Sequence[Message | ExtensionItem],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[list[Any], list[str]]:
    """将消息列表转换为provider消息格式
    Convert message list to provider message format

    这个方法通常会委托给message_ops_class来处理。
    This method typically delegates to message_ops_class for processing.

    Args:
        messages: IR格式的消息列表（可包含扩展项）
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        Tuple[转换后的消息列表, 警告信息列表]
    """
    pass

@abstractmethod
def request_from_provider(
    self,
    provider_request: dict[str, Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> IRRequest:
    """将provider请求转换为IRRequest
    Convert provider request to IRRequest

    Subclass helper: call ``self._convert_tools_from_p(tools)`` to convert
    provider tool definitions to IR format.

    Args:
        provider_request: Provider格式的请求
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        IR格式的请求
    """
    pass

@abstractmethod
def request_to_provider(
    self,
    ir_request: IRRequest,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[dict[str, Any], list[str]]:
    """将IRRequest转换为provider请求参数
    Convert IRRequest to provider request parameters

    这是最高层的转换方法，会调用各个功能域的ops类来完成转换：
    - 使用message_ops处理messages字段
    - 使用config_ops处理generation、stream等配置字段
    - 使用tool_ops处理tools、tool_choice等工具字段

    This is the highest-level conversion method that calls ops classes from various functional domains:
    - Uses message_ops to handle messages field
    - Uses config_ops to handle generation, stream and other config fields
    - Uses tool_ops to handle tools, tool_choice and other tool fields

    Subclass helper: call ``self._apply_tool_config(ir_request, result, ctx)``
    to handle the tools / tool_choice / tool_config fields.

    Args:
        ir_request: IR格式的完整请求
        context: Optional conversion context for carrying warnings,
            options, and metadata through the pipeline.
        **kwargs: 额外参数

    Returns:
        Tuple[转换后的请求参数, 警告信息列表]
    """
    pass

@abstractmethod
def response_from_provider(
    self,
    provider_response: dict[str, Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> IRResponse:
    """将provider响应转换为IRResponse
    Convert provider response to IRResponse

    Subclass helper: call ``self._build_ir_usage(p_usage)`` to convert
    provider usage to IR format.

    Args:
        provider_response: Provider格式的响应
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        IR格式的响应
    """
    pass

@abstractmethod
def response_to_provider(
    self,
    ir_response: IRResponse,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """将IRResponse转换为provider响应
    Convert IRResponse to provider response

    Subclass helper: call ``self._build_provider_usage(ir_usage)`` to convert
    IR usage to provider format.

    Args:
        ir_response: IR格式的响应
        context: Optional conversion context.
        **kwargs: 额外参数

    Returns:
        Provider格式的响应
    """
    pass

@abstractmethod
def stream_response_from_provider(
    self,
    chunk: dict[str, Any],
    context: StreamContext | None = None,
) -> list[IRStreamEvent]:
    """Convert a provider-native stream chunk to a list of IR stream events.

    A single provider chunk may produce zero or more IR events depending on
    the provider's SSE protocol.  For example, a chunk that carries both a
    text delta and a finish reason would yield two events.

    Args:
        chunk: Provider-native stream chunk (dict or SDK object that will
            be normalized internally by each concrete converter).
        context: Optional stream context for stateful conversions.
            When provided, converters may emit lifecycle events
            (StreamStart/End, ContentBlockStart/End) and track
            cross-chunk state.

    Returns:
        List of IR stream events extracted from the chunk.
    """
    pass

def stream_response_to_provider(
    self,
    event: IRStreamEvent,
    context: StreamContext | None = None,
) -> dict[str, Any] | list[dict[str, Any]]:
    """Convert an IR stream event to provider-native stream chunk(s).

    Uses ``_TO_P_DISPATCH`` to route each event type to its handler,
    then applies ``_post_process_to_provider`` for any provider-specific
    decoration of the result.

    Subclasses that need pre-dispatch logic (e.g., context upgrades)
    may override this method, perform their pre-processing, and call
    ``super().stream_response_to_provider(event, context)``.

    Args:
        event: IR stream event to convert.
        context: Optional stream context for stateful conversions.

    Returns:
        A single provider-native stream chunk dict, or a list of chunk
        dicts when the event maps to multiple provider-level messages.
    """
    handler_name = self._TO_P_DISPATCH.get(event.get("type", ""))
    if handler_name is None:
        return {}
    result = getattr(self, handler_name)(event, context)
    return self._post_process_to_provider(result, event, context)

def messages_from_provider(
    self,
    provider_messages: list[Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> list[Message | ExtensionItem]:
    """Convert OpenAI Chat messages to IR message list.

    Delegates to message_ops.

    Args:
        provider_messages: OpenAI Chat messages.

    Returns:
        IR messages.
    """
    return self.message_ops.p_messages_to_ir(provider_messages, **kwargs)

def messages_to_provider(
    self,
    messages: Sequence[Message | ExtensionItem],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[list[Any], list[str]]:
    """Convert IR message list to OpenAI Chat message format.

    Delegates to message_ops.

    Args:
        messages: IR messages (may contain ExtensionItems).

    Returns:
        Tuple of (converted messages, warnings).
    """
    return self.message_ops.ir_messages_to_p(messages, **kwargs)

def request_from_provider(
    self,
    provider_request: dict[str, Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> IRRequest:
    """Convert OpenAI Chat Completions request to IRRequest.

    Args:
        provider_request: OpenAI request dict (or SDK object).

    Returns:
        IR request.
    """
    provider_request = self._normalize(provider_request)

    ir_request: dict[str, Any] = {
        "model": provider_request.get("model", ""),
        "messages": [],
    }

    # 1. Messages - separate system messages as system_instruction
    messages = provider_request.get("messages", [])
    ir_messages, system_text = self._extract_system_and_messages(messages)
    ir_request["messages"] = ir_messages
    if system_text:
        ir_request["system_instruction"] = system_text

    # 2. Tools
    tools = provider_request.get("tools")
    if tools:
        ir_request["tools"] = self._convert_tools_from_p(tools)

    # 3. Tool choice
    tool_choice = provider_request.get("tool_choice")
    if tool_choice is not None:
        ir_request["tool_choice"] = self.tool_ops.p_tool_choice_to_ir(tool_choice)

    # 4. Tool config (parallel_tool_calls)
    parallel_tool_calls = provider_request.get("parallel_tool_calls")
    if parallel_tool_calls is not None:
        ir_request["tool_config"] = self.tool_ops.p_tool_config_to_ir(
            {"parallel_tool_calls": parallel_tool_calls}
        )

    # 5. Generation config
    gen_config = self.config_ops.p_generation_config_to_ir(provider_request)
    if gen_config:
        ir_request["generation"] = gen_config

    # 6. Response format
    resp_format = provider_request.get("response_format")
    if resp_format:
        ir_request["response_format"] = self.config_ops.p_response_format_to_ir(
            resp_format
        )

    # 7. Reasoning config
    reasoning = self.config_ops.p_reasoning_config_to_ir(provider_request)
    if reasoning:
        ir_request["reasoning"] = reasoning

    # 8. Stream config
    stream = provider_request.get("stream")
    stream_options = provider_request.get("stream_options")
    if stream is not None or stream_options:
        ir_request["stream"] = self.config_ops.p_stream_config_to_ir(
            {"stream": stream, "stream_options": stream_options}
        )

    # 9. Cache config
    cache_fields = {}
    if "prompt_cache_key" in provider_request:
        cache_fields["prompt_cache_key"] = provider_request["prompt_cache_key"]
    if "prompt_cache_retention" in provider_request:
        cache_fields["prompt_cache_retention"] = provider_request[
            "prompt_cache_retention"
        ]
    if cache_fields:
        ir_request["cache"] = self.config_ops.p_cache_config_to_ir(cache_fields)

    return self._validate_ir_request(ir_request)

def request_to_provider(
    self,
    ir_request: IRRequest,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[dict[str, Any], list[str]]:
    """Convert IRRequest to OpenAI Chat Completions request parameters.

    Orchestrates all Ops classes to build the complete provider request.

    Args:
        ir_request: IR request.

    Returns:
        Tuple of (provider request dict, warnings list).
    """
    ctx = context if context is not None else ConversionContext()
    result: dict[str, Any] = {"model": ir_request["model"]}

    # 1. System instruction → system message
    messages: list[dict[str, Any]] = []
    system_instruction = ir_request.get("system_instruction")
    if system_instruction:
        messages.append({"role": "system", "content": system_instruction})

    # 2. Messages — fix orphaned tool_calls at IR level before conversion.
    # OpenAI Chat API strictly requires every tool_call_id to have a
    # matching role:tool response.  Other providers (Anthropic, Google)
    # are lenient, so cross-format conversions may carry orphaned
    # tool_calls from interrupted sessions.
    ir_messages = fix_orphaned_tool_calls_ir(ir_request.get("messages", []))
    ctx.warnings.extend(strip_orphaned_tool_config(ir_request))
    converted_msgs, msg_warnings = self.message_ops.ir_messages_to_p(ir_messages)
    messages.extend(converted_msgs)
    ctx.warnings.extend(msg_warnings)
    result["messages"] = messages

    # 3-5. Tools + tool_choice + tool_config
    self._apply_tool_config(ir_request, result, ctx)

    # 6. Generation config
    gen_config = ir_request.get("generation")
    if gen_config:
        gen_fields = self.config_ops.ir_generation_config_to_p(gen_config)
        result.update(gen_fields)

    # 7. Response format
    resp_format = ir_request.get("response_format")
    if resp_format:
        rf_fields = self.config_ops.ir_response_format_to_p(resp_format)
        result.update(rf_fields)

    # 8. Stream config
    stream = ir_request.get("stream")
    if stream:
        stream_fields = self.config_ops.ir_stream_config_to_p(stream)
        result.update(stream_fields)

    # 9. Reasoning config
    reasoning = ir_request.get("reasoning")
    if reasoning:
        reasoning_fields = self.config_ops.ir_reasoning_config_to_p(reasoning)
        result.update(reasoning_fields)

    # 10. Cache config
    cache = ir_request.get("cache")
    if cache:
        cache_fields = self.config_ops.ir_cache_config_to_p(cache)
        result.update(cache_fields)

    # 11. Provider extensions (pass-through)
    extensions = ir_request.get("provider_extensions")
    if extensions:
        result.update(extensions)

    return result, ctx.warnings

def response_from_provider(
    self,
    provider_response: dict[str, Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> IRResponse:
    """Convert OpenAI Chat Completions response to IRResponse.

    Args:
        provider_response: OpenAI response dict (or SDK object).

    Returns:
        IR response.
    """
    provider_response = self._normalize(provider_response)

    choices = []
    for p_choice in provider_response.get("choices", []):
        message = self.message_ops._p_message_to_ir(
            p_choice.get("message", p_choice.get("delta", {}))
        )

        finish_reason_val = p_choice.get("finish_reason")

        choice_info: dict[str, Any] = {
            "index": p_choice.get("index", 0),
            "message": message,
            "finish_reason": {
                "reason": OPENAI_CHAT_REASON_FROM_PROVIDER.get(
                    finish_reason_val, "stop"
                )
            },
        }

        if "logprobs" in p_choice:
            choice_info["logprobs"] = p_choice["logprobs"]

        choices.append(choice_info)

    ir_response: dict[str, Any] = {
        "id": provider_response.get("id", ""),
        "object": "response",
        "created": provider_response.get("created", 0),
        "model": provider_response.get("model", ""),
        "choices": choices,
    }

    # Usage
    p_usage = provider_response.get("usage")
    if p_usage:
        ir_response["usage"] = self._build_ir_usage(p_usage)

    if provider_response.get("service_tier") is not None:
        ir_response["service_tier"] = provider_response["service_tier"]

    if provider_response.get("system_fingerprint") is not None:
        ir_response["system_fingerprint"] = provider_response["system_fingerprint"]

    return self._validate_ir_response(ir_response)

def response_to_provider(
    self,
    ir_response: IRResponse,
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Convert IRResponse to OpenAI Chat Completions response.

    Args:
        ir_response: IR response.

    Returns:
        OpenAI response dict.
    """
    provider_response: dict[str, Any] = {
        "id": ir_response.get("id", ""),
        "object": "chat.completion",
        "created": ir_response.get("created", 0),
        "model": ir_response.get("model", ""),
        "choices": [],
    }

    for choice in ir_response.get("choices", []):
        openai_choice = self._build_choice_to_provider(choice)  # ty: ignore[invalid-argument-type]
        if openai_choice is not None:
            provider_response["choices"].append(openai_choice)

    # Usage
    ir_usage = ir_response.get("usage")
    if ir_usage:
        provider_response["usage"] = self._build_provider_usage(ir_usage)  # ty: ignore[invalid-argument-type]

    if "service_tier" in ir_response:
        provider_response["service_tier"] = ir_response["service_tier"]

    if "system_fingerprint" in ir_response:
        provider_response["system_fingerprint"] = ir_response["system_fingerprint"]

    return provider_response

def stream_response_from_provider(
    self,
    chunk: dict[str, Any],
    context: StreamContext | None = None,
) -> list[IRStreamEvent]:
    """Convert an OpenAI SSE chunk to IR stream events.

    A single chunk may produce multiple events (e.g., text delta + finish).

    When a ``context`` is provided, lifecycle events (``StreamStartEvent``,
    ``StreamEndEvent``) are emitted and cross-chunk state is tracked.
    Without a context the behaviour is identical to the previous
    implementation (backward compatible).

    Args:
        chunk: OpenAI SSE chunk dict (or SDK object).
        context: Optional stream context for stateful conversions.

    Returns:
        List of IR stream events extracted from the chunk.
    """
    chunk = self._normalize(chunk)
    events: list[IRStreamEvent] = []

    if context is not None and not context.is_started:
        self._handle_stream_start_from_p(chunk, context, events)

    choices = chunk.get("choices", [])
    for p_choice in choices:
        self._handle_choice_from_p(p_choice, context, events)

    usage = chunk.get("usage")
    if usage:
        self._handle_usage_from_p(usage, events)

    self._handle_stream_end_from_p(choices, events, usage, context)

    return events

@classmethod
def create_stream_context(cls) -> OpenAIResponsesStreamContext:
    """Create a stream context with Responses API specific state."""
    return OpenAIResponsesStreamContext()

def messages_from_provider(
    self,
    provider_messages: list[Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> list[Message | ExtensionItem]:
    """Convert OpenAI Responses items to IR message list.

    Delegates to message_ops.

    Args:
        provider_messages: OpenAI Responses items.

    Returns:
        IR messages.
    """
    return self.message_ops.p_messages_to_ir(provider_messages, **kwargs)

def messages_to_provider(
    self,
    messages: Sequence[Message | ExtensionItem],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> tuple[list[Any], list[str]]:
    """Convert IR message list to OpenAI Responses input items.

    Delegates to message_ops.

    Args:
        messages: IR messages (may contain ExtensionItems).

    Returns:
        Tuple of (converted items, warnings).
    """
    return self.message_ops.ir_messages_to_p(messages, **kwargs)

def request_from_provider(
    self,
    provider_request: dict[str, Any],
    *,
    context: ConversionContext | None = None,
    **kwargs: Any,
) -> IRRequest:
    """Convert OpenAI Responses API request to IRRequest.

    Args:
        provider_request: OpenAI Responses request dict (or SDK object).

    Returns:
        IR request.
    """
    provider_request = self._normalize(provider_request)

    ir_request: dict[str, Any] = {
        "model": provider_request.get("model", ""),
        "messages": [],
    }

    # 1. Instructions → system_instruction
    instructions = provider_request.get("instructions")
    if instructions:
        ir_request["system_instruction"] = instructions

    # 2. Input items → messages
    input_items = provider_request.get("input", [])
    if isinstance(input_items, str):
        input_items = [
            {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": input_items}],
            }
        ]
    if isinstance(input_items, list):
        ir_messages = self.message_ops.p_messages_to_ir(input_items)
        ir_request["messages"] = ir_messages

    # 3. Tools
    tools = provider_request.get("tools")
    if tools:
        ir_tools = self._convert_tools_from_p(tools)
        if ir_tools:
            ir_request["tools"] = ir_tools

    # 4-5. Tool choice + tool config
    self._convert_tool_config_from_p(provider_request, ir_request)

    # 6. Generation config
    gen_config = self.config_ops.p_generation_config_to_ir(provider_request)
    if gen_config:
        ir_request["generation"] = gen_config

    # 7. Response format (text field in Responses API)
    text_format = provider_request.get("text")
    if text_format:
        ir_request["response_format"] = self.config_ops.p_response_format_to_ir(
            text_format
        )

    # 8. Reasoning config
    reasoning = self.config_ops.p_reasoning_config_to_ir(provider_request)
    if reasoning:
        ir_request["reasoning"] = reasoning

    # 9. Stream config
    stream = provider_request.get("stream")
    stream_options = provider_request.get("stream_options")
    if stream is not None or stream_options:
        ir_request["stream"] = self.config_ops.p_stream_config_to_ir(
            {"stream": stream, "stream_options": stream_options}
        )

    # 10. Cache config
    self._convert_cache_from_p(provider_request, ir_request)

    # 11. Provider extensions (passthrough fields like allowed_tools)
    allowed_tools = provider_request.get("allowed_tools")
    if allowed_tools is not None:
        ir_request.setdefault("provider_extensions", {})["allowed_tools"] = (
            allowed_tools
        )

    # Preserve mode: capture request fields for echo-back in response
    ctx = context if context is not None else ConversionContext()
    if ctx.metadata_mode == "preserve":
        echo = {
            k: v
            for k, v in provider_request.items()
            if k in RESPONSES_PRESERVE_FIELDS
        }
        if echo:
            ctx.store_request_echo(echo)

    return self._validate_ir_request(ir_request)