vllm.entrypoints.openai.responses.utils

Functions:

• build_response_output_items –

  Build the output items of one generation.

• construct_chat_messages_with_tool_call –

  Build chat messages from response items.

• convert_tool_responses_to_completions_format –

  Convert a flat Responses tool schema:

• decode_custom_tool_input –

  Payload of a completed shim call; bare text is passed through.

• decode_custom_tool_input_prefix –

  Longest decodable payload prefix of partial shim arguments, so that

• extend_custom_tool_input_prefix –

  :func:decode_custom_tool_input_prefix of arguments, decoding only

• extract_tool_types –

  Extracts the tool types from the given tools.

• response_item_status –

  Status of the output item that ended a generation with finish_reason.

• reuse_streamed_item_ids –

  Give rebuilt output items the ids already streamed for the same message.

• should_continue_final_message –

  Determine if the last input message is a partial assistant message

_construct_message_from_response_item(item, prev_msg=None)

Returns a new message or None. If None, prev_msg might be updated. If prev_msg is None, a new message is always returned.

Source code in vllm/entrypoints/openai/responses/utils.py

def _construct_message_from_response_item(
    item: ResponseInputOutputItem,
    prev_msg: ChatCompletionMessageParam | None = None,
) -> ChatCompletionMessageParam | None:
    """Returns a new message or None. If `None`, `prev_msg` might be updated.
    If `prev_msg` is `None`, a new message is always returned.
    """
    prev_assistant_msg = (
        prev_msg if prev_msg and prev_msg.get("role") == "assistant" else None
    )

    tool_call: ChatCompletionMessageToolCallParam | None = None
    if isinstance(item, ResponseFunctionToolCall):
        tool_name = item.name
        if item.namespace:
            tool_name = flat_namespace_tool_name(item.namespace, item.name)
        tool_call = ChatCompletionMessageToolCallParam(
            id=item.call_id,
            function=FunctionCallTool(
                name=tool_name,
                arguments=item.arguments,
            ),
            type="function",
        )
    elif _item_type(item) == "custom_tool_call":
        call = _item_dict(item)
        tool_call = ChatCompletionMessageToolCallParam(
            id=call["call_id"],
            function=FunctionCallTool(
                name=call["name"],
                arguments=encode_custom_tool_input(call.get("input") or ""),
            ),
            type="function",
        )
    if tool_call is not None:
        if prev_assistant_msg:
            tool_calls = prev_assistant_msg.get("tool_calls")
            if tool_calls is None:
                prev_assistant_msg["tool_calls"] = [tool_call]
                return None
            if isinstance(tool_calls, list):
                tool_calls.append(tool_call)
                return None
            if isinstance(tool_calls, Iterable) and not isinstance(
                tool_calls, (dict, str)
            ):
                tool_calls = list(tool_calls)
                tool_calls.append(tool_call)
                prev_assistant_msg["tool_calls"] = tool_calls
                return None
            logger.warning(
                "Previous assistant message has unknown tool_calls format. "
                "Tool call merging is skipped and a new assistant message is created. "
                "Call %s",
                tool_call["id"],
            )
        return ChatCompletionAssistantMessageParam(
            role="assistant",
            tool_calls=[tool_call],
        )
    if isinstance(item, ResponseReasoningItem):
        reasoning = ""
        if item.encrypted_content:
            raise VLLMValidationError(
                "Encrypted content is not supported.",
                parameter="input",
            )
        elif item.content and len(item.content) >= 1:
            reasoning = item.content[0].text
        elif len(item.summary) >= 1:
            reasoning = item.summary[0].text
            logger.warning(
                "Using summary text as reasoning content for item %s. "
                "Please use content instead of summary for "
                "reasoning items.",
                item.id,
            )

        if prev_assistant_msg:
            previous_reasoning = prev_assistant_msg.get("reasoning")
            if previous_reasoning is None:
                prev_assistant_msg["reasoning"] = reasoning
                return None
        return {
            "role": "assistant",
            "reasoning": reasoning,
        }
    elif isinstance(item, ResponseOutputMessage):
        output_text = item.content[0].text
        if prev_assistant_msg:
            previous_content = prev_assistant_msg.get("content")
            if previous_content is None:
                prev_assistant_msg["content"] = output_text
                return None
        return {
            "role": "assistant",
            "content": output_text,
        }
    elif isinstance(item, ResponseFunctionToolCallOutputItem):
        return ChatCompletionToolMessageParam(
            role="tool",
            content=item.output,
            tool_call_id=item.call_id,
        )
    elif _item_type(item) in ("function_call_output", "custom_tool_call_output"):
        output = _item_dict(item)
        return ChatCompletionToolMessageParam(
            role="tool",
            content=output.get("output"),
            tool_call_id=output.get("call_id"),
        )
    elif isinstance(item, dict) and item.get("role") == "assistant":
        content = item.get("content")
        text: str | None = None
        if isinstance(content, str):
            text = content
        elif isinstance(content, list) and content:
            text = content[0].get("text")
        if text is not None:
            if prev_assistant_msg:
                previous_content = prev_assistant_msg.get("content")
                if previous_content is None:
                    prev_assistant_msg["content"] = text
                    return None
            return {"role": "assistant", "content": text}
    if isinstance(item, dict) and "role" in item:
        return item  # type: ignore[return-value]
    item_type = item.get("type") if isinstance(item, dict) else item.type
    raise VLLMValidationError(
        f"Unsupported input item type: {item_type}",
        parameter="input",
    )

build_response_output_items(reasoning, content, tool_calls, logprobs=None, tools=None, finish_reason=None)

Build the output items of one generation.

Only the last item can have been cut short, so finish_reason decides that item's status and every earlier item is completed.

Source code in vllm/entrypoints/openai/responses/utils.py

def build_response_output_items(
    reasoning: str | None,
    content: str | None,
    tool_calls: list[FunctionCall] | None,
    logprobs: list[Logprob] | None = None,
    tools: list[Tool] | None = None,
    finish_reason: str | None = None,
) -> list[ResponseOutputItem]:
    """Build the output items of one generation.

    Only the last item can have been cut short, so ``finish_reason`` decides
    that item's status and every earlier item is ``completed``.
    """
    outputs: list[ResponseOutputItem] = []
    tool_call_name_map = build_responses_tool_call_name_map(tools)
    custom_names = custom_tool_names(tools)
    last_item_status = response_item_status(finish_reason)

    if reasoning:
        outputs.append(
            ResponseReasoningItem(
                id=f"rs_{random_uuid()}",
                summary=[],
                type="reasoning",
                content=[
                    ResponseReasoningTextContent(text=reasoning, type="reasoning_text")
                ],
                status=None,
            )
        )

    if content:
        outputs.append(
            ResponseOutputMessage(
                id=f"msg_{random_uuid()}",
                content=[
                    ResponseOutputText(
                        text=content,
                        annotations=[],
                        type="output_text",
                        logprobs=logprobs,
                    )
                ],
                role="assistant",
                status="completed" if tool_calls else last_item_status,
                type="message",
            )
        )

    if tool_calls:
        for idx, tool_call in enumerate(tool_calls):
            is_last = idx == len(tool_calls) - 1
            call_id = tool_call.id or make_tool_call_id(
                func_name=tool_call.name, idx=idx
            )
            if tool_call.name in custom_names:
                outputs.append(
                    ResponseCustomToolCall(
                        id=f"ctc_{random_uuid()}",
                        call_id=call_id,
                        type="custom_tool_call",
                        name=tool_call.name,
                        input=decode_custom_tool_input(tool_call.arguments),
                    )
                )
                continue
            call_name = resolve_responses_tool_call_name(
                tool_call.name, tool_call_name_map=tool_call_name_map
            )
            outputs.append(
                ResponseFunctionToolCall(
                    id=f"fc_{random_uuid()}",
                    call_id=call_id,
                    type="function_call",
                    status=last_item_status if is_last else "completed",
                    name=call_name.name,
                    namespace=call_name.namespace,
                    arguments=tool_call.arguments,
                )
            )

    return outputs

construct_chat_messages_with_tool_call(input_messages)

Build chat messages from response items.

Some chat messages span multiple response items (e.g., reasoning + tool calls).

Source code in vllm/entrypoints/openai/responses/utils.py

def construct_chat_messages_with_tool_call(
    input_messages: list[ResponseInputOutputItem],
) -> list[ChatCompletionMessageParam]:
    """Build chat messages from response items.

    Some chat messages span multiple response items (e.g., reasoning + tool calls).
    """
    messages: list[ChatCompletionMessageParam] = []
    for item in input_messages:
        message = _construct_message_from_response_item(
            item, prev_msg=messages[-1] if messages else None
        )
        if message is not None:
            messages.append(message)

    return messages

convert_tool_responses_to_completions_format(tool)

Convert a flat Responses tool schema

{"type": "function", "name": "...", "description": "...", "parameters": {...}}

into a Chat Completions tool param for chat-template rendering.

Source code in vllm/entrypoints/openai/responses/utils.py

def convert_tool_responses_to_completions_format(
    tool: dict,
) -> ChatCompletionToolsParam:
    """Convert a flat Responses tool schema:
        {"type": "function", "name": "...", "description": "...", "parameters": {...}}
    into a Chat Completions tool param for chat-template rendering.
    """
    return ChatCompletionToolsParam(
        type="function",
        function=FunctionDefinition.model_validate(
            {k: v for k, v in tool.items() if k != "type"}
        ),
    )

decode_custom_tool_input(arguments)

Payload of a completed shim call; bare text is passed through.

Source code in vllm/entrypoints/openai/responses/utils.py

def decode_custom_tool_input(arguments: str) -> str:
    """Payload of a completed shim call; bare text is passed through."""
    try:
        payload = _custom_tool_payload(json.loads(arguments))
    except ValueError:
        return arguments
    return arguments if payload is None else payload

decode_custom_tool_input_prefix(arguments)

Longest decodable payload prefix of partial shim arguments, so that streamed custom_tool_call_input deltas stay a prefix of the final input.

Source code in vllm/entrypoints/openai/responses/utils.py

def decode_custom_tool_input_prefix(arguments: str) -> str:
    """Longest decodable payload prefix of partial shim arguments, so that
    streamed ``custom_tool_call_input`` deltas stay a prefix of the final input."""
    try:
        parsed = partial_json_parser.loads(arguments, Allow.STR | Allow.OBJ)
    except (ValueError, IndexError, AssertionError):
        return ""
    payload = _custom_tool_payload(parsed) or ""
    if payload and "\ud800" <= payload[-1] <= "\udbff":
        payload = payload[:-1]
    return payload

extend_custom_tool_input_prefix(arguments, payload, end)

:func:decode_custom_tool_input_prefix of arguments, decoding only arguments[end:]. payload and end come from the previous call on a prefix of arguments ("" and 0 at first). Returns None when the arguments do not start as :func:encode_custom_tool_input writes them, or once the payload string closes; only a full decode is exact then.

Source code in vllm/entrypoints/openai/responses/utils.py

def extend_custom_tool_input_prefix(
    arguments: str, payload: str, end: int
) -> tuple[str, int] | None:
    """:func:`decode_custom_tool_input_prefix` of *arguments*, decoding only
    ``arguments[end:]``. *payload* and *end* come from the previous call on a
    prefix of *arguments* (``""`` and 0 at first). Returns None when the
    arguments do not start as :func:`encode_custom_tool_input` writes them,
    or once the payload string closes; only a full decode is exact then."""
    if not end:
        if not arguments.startswith(_CUSTOM_TOOL_INPUT_HEAD):
            return ("", 0) if _CUSTOM_TOOL_INPUT_HEAD.startswith(arguments) else None
        end = len(_CUSTOM_TOOL_INPUT_HEAD)
    # partial_json_parser strips trailing whitespace before it completes the
    # string. Decode up to the last complete escape, holding back a high
    # surrogate so that a pair is never split between calls.
    tail = arguments[end:].rstrip()
    i = cut = 0
    high_end = -1
    while i < len(tail):
        if tail[i] == "\\":
            if tail[i + 1 : i + 2] == "u":
                if "d800" <= tail[i + 2 : i + 6].lower() <= "dbff":
                    high_end = i + 6
                i += 6
            else:
                i += 2
            if i > len(tail):
                break
        else:
            i += 1
        cut = i
    if cut == high_end:
        cut -= 6
    try:
        chunk = json.loads(f'"{tail[:cut]}"')
    except ValueError:
        return None
    if chunk and "\ud800" <= chunk[-1] <= "\udbff":
        return None
    return payload + chunk, end + cut

extract_tool_types(tools)

Extracts the tool types from the given tools.

Source code in vllm/entrypoints/openai/responses/utils.py

def extract_tool_types(tools: list[Tool]) -> set[str]:
    """Extracts the tool types from the given tools."""
    tool_types: set[str] = set()
    for tool in tools:
        if tool.type == "mcp":
            # Allow the MCP Tool type to enable built in tools if the
            # server_label is allowlisted in
            # envs.VLLM_GPT_OSS_SYSTEM_TOOL_MCP_LABELS
            if tool.server_label in envs.VLLM_GPT_OSS_SYSTEM_TOOL_MCP_LABELS:
                tool_types.add(tool.server_label)
        else:
            tool_types.add(tool.type)
    return tool_types

response_item_status(finish_reason)

Status of the output item that ended a generation with finish_reason.

Source code in vllm/entrypoints/openai/responses/utils.py

def response_item_status(
    finish_reason: str | None,
) -> Literal["completed", "incomplete"]:
    """Status of the output item that ended a generation with ``finish_reason``."""
    return "incomplete" if finish_reason == "length" else "completed"

reuse_streamed_item_ids(items, streamed_items)

Give rebuilt output items the ids already streamed for the same message.

Items are paired by type in order; an item without a streamed counterpart keeps its own id.

Source code in vllm/entrypoints/openai/responses/utils.py

def reuse_streamed_item_ids(
    items: list[ResponseOutputItem],
    streamed_items: list[ResponseOutputItem],
) -> None:
    """Give rebuilt output items the ids already streamed for the same message.

    Items are paired by type in order; an item without a streamed counterpart
    keeps its own id.
    """
    remaining = list(streamed_items)
    for item in items:
        for i, streamed in enumerate(remaining):
            if streamed.type != item.type:
                continue
            item.id = streamed.id
            if isinstance(item, ResponseFunctionToolCall):
                assert isinstance(streamed, ResponseFunctionToolCall)
                item.call_id = streamed.call_id
            del remaining[i]
            break

should_continue_final_message(request_input)

Determine if the last input message is a partial assistant message that should be continued rather than starting a new generation.

This enables partial message completion similar to Anthropic's Messages API, where users can provide an incomplete assistant message and have the model continue from where it left off.

A message is considered partial if: 1. It's a ResponseOutputMessage or ResponseReasoningItem 2. Its status is "in_progress" or "incomplete"

Parameters:

• request_input

  (str | list[ResponseInputOutputItem]) –

  The input to the Responses API request

Returns:

• bool –

  True if the final message should be continued, False otherwise

Source code in vllm/entrypoints/openai/responses/utils.py

def should_continue_final_message(
    request_input: str | list[ResponseInputOutputItem],
) -> bool:
    """Determine if the last input message is a partial assistant message
    that should be continued rather than starting a new generation.

    This enables partial message completion similar to Anthropic's Messages API,
    where users can provide an incomplete assistant message and have the model
    continue from where it left off.

    A message is considered partial if:
    1. It's a ResponseOutputMessage or ResponseReasoningItem
    2. Its status is "in_progress" or "incomplete"

    Args:
        request_input: The input to the Responses API request

    Returns:
        True if the final message should be continued, False otherwise

    """
    if isinstance(request_input, str):
        # Simple string input is always a user message
        return False

    if not request_input:
        return False

    last_item = request_input[-1]

    # Check if the last item is a partial assistant message
    if isinstance(last_item, ResponseOutputMessage):
        return last_item.status in ("in_progress", "incomplete")

    # Check if the last item is a partial reasoning item
    if isinstance(last_item, ResponseReasoningItem):
        return last_item.status in ("in_progress", "incomplete")

    if isinstance(last_item, dict):
        # only support partial completion for messages for now
        if last_item.get("type", "message") not in ("message", "reasoning"):
            return False
        return last_item.get("status") in ("in_progress", "incomplete")

    return False