LiteLLM fixes to make it work with cursor/opencode/etc

callbacks.py

# Useful fixes for running gpt-oss-120b on vLLM
#
# Implements required fixes in order to get the following clients
# to work:
# - Cursor
# - Opencode
import json
import os
import time
from typing import Any, AsyncGenerator, Optional, Union

from litellm.integrations.custom_logger import CustomLogger
from litellm.types.utils import CallTypesLiteral


def _convert_responses_input_to_messages(input_items: list) -> list:
    """
    Convert OpenAI Responses API 'input' items to Chat Completions 'messages' format.

    Handles:
      - {"role": "...", "content": "..."}  → kept as-is
      - {"type": "message", "role": "...", "content": ...} → unwrapped
      - {"type": "function_call", ...}     → assistant message with tool_calls
      - {"type": "function_call_output",...} → tool message
    """
    messages = []
    pending_tool_calls: list = []

    for item in input_items:
        if not isinstance(item, dict):
            continue

        item_type = item.get("type", "")

        if item_type == "function_call":
            # Accumulate into the current assistant turn
            pending_tool_calls.append({
                "id": item.get("call_id", f"call_{len(pending_tool_calls)}"),
                "type": "function",
                "function": {
                    "name": item.get("name", ""),
                    "arguments": item.get("arguments", "{}"),
                },
            })

        elif item_type == "function_call_output":
            # Flush pending tool calls as an assistant message first
            if pending_tool_calls:
                messages.append({
                    "role": "assistant",
                    "content": None,
                    "tool_calls": pending_tool_calls,
                })
                pending_tool_calls = []

            output = item.get("output", [])
            if isinstance(output, list):
                parts = []
                for part in output:
                    if isinstance(part, dict):
                        text = (
                            part.get("text")
                            or part.get("output")
                            or ""
                        )
                        if isinstance(text, list):
                            # Nested content list
                            for sub in text:
                                if isinstance(sub, dict):
                                    parts.append(sub.get("text", ""))
                                else:
                                    parts.append(str(sub))
                        elif text:
                            parts.append(str(text))
                    else:
                        parts.append(str(part))
                content = "\n".join(p for p in parts if p)
            elif output is None:
                content = ""
            else:
                content = str(output)

            messages.append({
                "role": "tool",
                "tool_call_id": item.get("call_id", ""),
                "content": content,
            })

        else:
            # Regular message (type="message" wrapper or bare role+content)
            # Flush any pending tool calls
            if pending_tool_calls:
                messages.append({
                    "role": "assistant",
                    "content": None,
                    "tool_calls": pending_tool_calls,
                })
                pending_tool_calls = []

            if "role" in item:
                messages.append(item)
            elif item_type == "message" and "role" in item:
                # Unwrap the type wrapper
                messages.append({k: v for k, v in item.items() if k != "type"})
            # else: skip unknown items silently

    # Flush any remaining tool calls
    if pending_tool_calls:
        messages.append({
            "role": "assistant",
            "content": None,
            "tool_calls": pending_tool_calls,
        })

    return messages


class DefaultReasoningEffort(CustomLogger):
    """
    1. Sets reasoning_effort='medium' by default for gpt-oss-120b (Harmony).
       Callers can override with 'low', 'medium', or 'high'.
       Note: Harmony does not support reasoning_effort='none'.

    2. Fixes a LiteLLM streaming bug where the content: "" initialization
       chunk is dropped, causing "text part not found" errors in streaming
       SDKs (OpenAI SDK helpers, Vercel AI SDK, Mux Coder, etc.).

       The fix: inject content="" into the first assistant chunk whenever
       content is missing. This covers:
         - Reasoning + tool call responses (reasoning_content present)
         - Pure tool call responses (no reasoning_content)
         - Any other response where content is absent from the first chunk

    3. Handles OpenAI Responses API requests that use 'input' instead of
       'messages'. Converts function_call / function_call_output items to
       the standard Chat Completions tool-call format so that vLLM can
       process them.
    """
    async def async_pre_call_hook(
        self,
        user_api_key_dict,
        cache,
        data: dict,
        call_type: CallTypesLiteral,
    ) -> Optional[Union[Exception, str, dict]]:
        # ------------------------------------------------------------------ #
        # Handle OpenAI Responses API format: 'input' instead of 'messages'  #
        # ------------------------------------------------------------------ #
        if "input" in data and not data.get("messages"):
            raw_input = data.pop("input")
            if isinstance(raw_input, list):
                data["messages"] = _convert_responses_input_to_messages(raw_input)
            else:
                # Fallback: treat as-is (shouldn't happen in practice)
                data["messages"] = raw_input

        if "reasoning_effort" not in data:
            data["reasoning_effort"] = "medium"

        # vLLM 0.18 pydantic v2 bug: fields typed as Iterable[...] create lazy
        # ValidatorIterators during union validation, which fail downstream as
        # "Input should be a valid dictionary or instance of Content
        # [input_value=ValidatorIterator]". Normalize ALL messages to plain
        # dicts and flatten text-only list content to strings so vLLM validates
        # against the str branch, not Iterable. Also converts empty content
        # lists to None (proper OpenAI format for assistant+tool_calls messages).
        raw_messages = data.get("messages") or []
        normalized = []
        for msg in raw_messages:
            # Ensure we have a mutable plain dict (handles pydantic models too)
            if hasattr(msg, "model_dump"):
                d = msg.model_dump()
            elif not isinstance(msg, dict):
                try:
                    d = dict(msg)
                except Exception:
                    d = msg
            else:
                d = dict(msg)  # shallow copy so we don't mutate the original

            c = d.get("content")

            # Materialise any lazy iterator that isn't str/list/None
            if c is not None and not isinstance(c, (str, list)):
                try:
                    c = list(c)
                    d["content"] = c
                except Exception:
                    pass

            if isinstance(c, list):
                if not c:
                    # Empty list → None (correct for assistant+tool_calls messages)
                    d["content"] = None
                elif all(isinstance(p, dict) and p.get("type") in ("text", "input_text", "output_text") and "text" in p for p in c):
                    # All-text parts → flatten to plain string
                    d["content"] = "".join(p.get("text", "") for p in c)

            normalized.append(d)

        if normalized:
            data["messages"] = normalized

        return data

    async def async_post_call_streaming_iterator_hook(
        self,
        user_api_key_dict,
        response: Any,
        request_data: dict,
    ) -> AsyncGenerator:
        first = True
        async for chunk in response:
            if first:
                first = False
                try:
                    delta = chunk.choices[0].delta
                    # Inject content="" if the first assistant chunk has no content.
                    # This prevents "text part not found" errors in streaming SDKs
                    # that require a text part to be initialized even for tool-call
                    # or reasoning-only responses.
                    if getattr(delta, "role", None) == "assistant":
                        content = getattr(delta, "content", "__missing__")
                        if content in (None, "__missing__"):
                            delta.content = ""
                except Exception:
                    pass
            yield chunk



# Module-level instances required by LiteLLM's get_instance_fn loader
default_reasoning_effort = DefaultReasoningEffort()