"""Translate common Qwen/OpenClaude textual tool calls to OpenAI payloads."""
from __future__ import annotations
import ast
import html
import json
import re
import shlex
import uuid
from collections.abc import Mapping
from typing import Any
JSON_TOOL_CALL_RE = re.compile(
r"\s*(?P\{.*?\})\s*",
re.DOTALL | re.IGNORECASE,
)
XML_JSON_TOOL_CALL_RE = re.compile(
r"\s*(?P\{.*?\})\s*",
re.DOTALL | re.IGNORECASE,
)
STANDARD_XML_TOOL_CALL_RE = re.compile(
r"\s*(?P.*?)\s*",
re.DOTALL | re.IGNORECASE,
)
FUNCTION_NAME_RE = re.compile(
r"[A-Za-z_][\w.-]*)\s*>",
re.IGNORECASE,
)
PARAMETER_RE = re.compile(
r"[A-Za-z_][\w.-]*)\s*>"
r"(?P.*?)",
re.DOTALL | re.IGNORECASE,
)
DASHED_XML_TOOL_CALL_RE = re.compile(
r"\s*\s*(?P[A-Za-z_][\w.-]*)\s*\s*"
r"\s*(?P.*?)\s*\s*",
re.DOTALL | re.IGNORECASE,
)
NAMED_ARGUMENT_RE = re.compile(
r"[\"'])"
r"(?P[A-Za-z_][\w.-]*)(?P=quote)\s*>"
r"(?P.*?)",
re.DOTALL | re.IGNORECASE,
)
ELEMENT_ARGUMENT_RE = re.compile(
r"<(?P[A-Za-z_][\w.-]*)\s*>(?P.*?)",
re.DOTALL | re.IGNORECASE,
)
SELF_CLOSING_TOOL_RE = re.compile(
r"<(?P[A-Za-z_][\w.-]*)\b(?P[^<>]*?)/\s*>",
re.DOTALL,
)
ATTRIBUTE_RE = re.compile(
r'''(?P[A-Za-z_][\w.-]*)\s*=\s*(?:
"(?P(?:\\.|[^"\\])*)"
|'(?P(?:\\.|[^'\\])*)'
)''',
re.DOTALL | re.VERBOSE,
)
HTML_ENTITY_RE = re.compile(
r"&(?:#[0-9]+|#[xX][0-9A-Fa-f]+|[A-Za-z][A-Za-z0-9]+);"
)
ASSISTANT_CALLED_TOOL_RE = re.compile(
r"^\s*\[Assistant called tool (?P[A-Za-z_][\w.-]*) "
r"with arguments (?P\{.*\})\]\s*$",
re.DOTALL,
)
TEXTUAL_TOOL_CALL_RE = re.compile(
r"^[ \t]*(?P[A-Za-z_][\w.-]*)[ \t]+"
r"(?:with|using)[ \t]+(?P.+?)[ \t]*$",
re.MULTILINE | re.IGNORECASE,
)
FENCED_JSON_RE = re.compile(
r"^\s*```(?:json)?\s*(?P\{.*\})\s*```\s*$",
re.DOTALL | re.IGNORECASE,
)
TOOL_CALL_CLOSE_RE = re.compile(r"\s*$", re.IGNORECASE)
XML_JSON_TOOL_CALL_CLOSE_RE = re.compile(
r"\s*\{.*?\}\s*\s*$",
re.DOTALL | re.IGNORECASE,
)
SELF_CLOSING_TOOL_AT_END_RE = re.compile(
r"<(?:tool\b|[A-Z][A-Za-z0-9_.-]*)\b[^<>]*/\s*>\s*(?:```)?\s*$",
re.DOTALL,
)
KNOWN_TEXTUAL_TOOL_NAMES = frozenset(
{
"agent",
"askuserquestion",
"bash",
"edit",
"enterplanmode",
"glob",
"grep",
"lsp",
"notebookedit",
"read",
"skill",
"task",
"todowrite",
"webfetch",
"websearch",
"write",
}
)
def has_complete_tool_call(text: str) -> bool:
"""Return true once generation has ended a supported XML tool-call form."""
return bool(
TOOL_CALL_CLOSE_RE.search(text)
or XML_JSON_TOOL_CALL_CLOSE_RE.search(text)
or SELF_CLOSING_TOOL_AT_END_RE.search(text)
or any(
match.group("name").casefold() in KNOWN_TEXTUAL_TOOL_NAMES
for match in TEXTUAL_TOOL_CALL_RE.finditer(text)
)
)
def _canonical_name(name: Any, allowed_names: set[str] | None) -> str | None:
if not isinstance(name, str) or not name:
return None
if not allowed_names:
return name
by_casefold = {candidate.casefold(): candidate for candidate in allowed_names}
normalized = name.casefold()
canonical = by_casefold.get(normalized)
if canonical is not None:
return canonical
# OpenClaude exposes the legacy Agent executor as Task. Accept both names
# in textual generations while returning the advertised catalog name.
alias = {"agent": "task", "task": "agent"}.get(normalized)
return by_casefold.get(alias) if alias else None
def _coerce_value(value: str) -> Any:
value = _unescape_entities(value.strip())
try:
return json.loads(value)
except json.JSONDecodeError:
return value
def _decode_attribute(value: str) -> str:
try:
value = json.loads(f'"{value}"')
except json.JSONDecodeError:
pass
return _unescape_entities(value)
def _unescape_entities(value: str) -> str:
"""Decode explicit entities without treating a URL's bare ``&`` as HTML.
``html.unescape`` accepts legacy semicolon-less names such as ``¤``.
That turns a query key like ``¤t_weather`` into ``¤t_weather``.
XML entities are terminated with a semicolon, so only decode that form.
"""
return HTML_ENTITY_RE.sub(lambda match: html.unescape(match.group(0)), value)
def _attributes(raw: str) -> dict[str, str]:
values: dict[str, str] = {}
for match in ATTRIBUTE_RE.finditer(raw):
value = match.group("double")
if value is None:
value = match.group("single")
if value is not None:
values[match.group("key")] = _decode_attribute(value)
return values
def _arguments(value: Any) -> dict[str, Any] | None:
if isinstance(value, Mapping):
return dict(value)
if not isinstance(value, str):
return None
parsed = _mapping_literal(_unescape_entities(value))
return dict(parsed) if isinstance(parsed, Mapping) else None
def _mapping_literal(value: str) -> Mapping[str, Any] | None:
"""Parse JSON or a Python-style mapping without evaluating expressions."""
try:
parsed = json.loads(value)
except json.JSONDecodeError:
try:
parsed = ast.literal_eval(value)
except (SyntaxError, ValueError):
return None
return parsed if isinstance(parsed, Mapping) else None
def normalize_openai_tool_arguments(value: Any) -> dict[str, Any]:
"""Return the mapping required by Qwen3's chat-template ``items`` filter.
OpenAI serializes function arguments as a JSON string, while Qwen3's
official template iterates them as a mapping when replaying tool history.
Accept both representations so a completed tool call can be followed by a
tool result without raising a template ``TypeError``.
"""
parsed = _arguments(value)
return parsed if parsed is not None else {}
def _openai_call(
name: Any,
arguments: Any,
allowed_names: set[str] | None,
) -> dict[str, Any] | None:
canonical_name = _canonical_name(name, allowed_names)
if canonical_name is None:
return None
if isinstance(arguments, str):
parsed = _arguments(arguments)
arguments = parsed if parsed is not None else {}
if not isinstance(arguments, Mapping):
arguments = {}
return {
"id": f"call_{uuid.uuid4().hex[:24]}",
"type": "function",
"function": {
"name": canonical_name,
"arguments": json.dumps(
dict(arguments), ensure_ascii=False, separators=(",", ":")
),
},
}
def _payload_call(
payload: Any,
allowed_names: set[str] | None,
) -> dict[str, Any] | None:
if not isinstance(payload, Mapping):
return None
function = payload.get("function")
if isinstance(function, Mapping):
return _openai_call(
function.get("name"),
function.get("arguments", {}),
allowed_names,
)
return _openai_call(
payload.get("name"), payload.get("arguments", {}), allowed_names
)
def _xml_arguments(arguments: str) -> dict[str, Any]:
named = {
match.group("key"): _coerce_value(match.group("value"))
for match in NAMED_ARGUMENT_RE.finditer(arguments)
}
if named:
return named
return {
match.group("key"): _coerce_value(match.group("value"))
for match in ELEMENT_ARGUMENT_RE.finditer(arguments)
}
def _textual_arguments(value: str) -> dict[str, Any] | None:
"""Parse Qwen's compact ``tool with key=value`` representation."""
raw = value.strip().rstrip(";").strip()
mapping = _mapping_literal(raw)
if isinstance(mapping, Mapping):
return dict(mapping)
# Normalize optional whitespace around '=' before shlex handles quoted
# values containing spaces. No expressions are evaluated here.
raw = re.sub(
r"(?P[A-Za-z_][\w.-]*)\s*=\s*",
r"\g=",
raw,
)
try:
tokens = shlex.split(raw, posix=True)
except ValueError:
return None
arguments: dict[str, Any] = {}
for token in tokens:
if "=" not in token:
continue
key, item = token.split("=", 1)
if not re.fullmatch(r"[A-Za-z_][\w.-]*", key):
continue
arguments[key] = _coerce_value(item)
return arguments or None
def extract_tool_call(
text: str,
allowed_names: set[str] | None = None,
) -> tuple[dict[str, Any] | None, str]:
"""Extract the first supported tool call for backward compatibility."""
calls, visible = extract_tool_calls(text, allowed_names)
return (calls[0] if calls else None), visible
def extract_tool_calls(
text: str,
allowed_names: set[str] | None = None,
) -> tuple[list[dict[str, Any]], str]:
"""Extract all tool calls while accepting Qwen's common XML variations.
Matching calls deliberately clear visible content. Agent clients should
receive structured OpenAI calls rather than Markdown/XML renditions of the
same calls before they execute the tools.
"""
candidates: list[tuple[int, int, dict[str, Any]]] = []
for match in JSON_TOOL_CALL_RE.finditer(text):
call = _payload_call(
_mapping_literal(match.group("payload")),
allowed_names,
)
if call:
candidates.append((match.start(), match.end(), call))
for match in XML_JSON_TOOL_CALL_RE.finditer(text):
call = _payload_call(
_mapping_literal(match.group("payload")),
allowed_names,
)
if call:
candidates.append((match.start(), match.end(), call))
for match in STANDARD_XML_TOOL_CALL_RE.finditer(text):
body = match.group("body")
function = FUNCTION_NAME_RE.search(body)
if function:
call = _openai_call(
function.group("name"),
{
parameter.group("key"): _coerce_value(parameter.group("value"))
for parameter in PARAMETER_RE.finditer(body)
},
allowed_names,
)
if call:
candidates.append((match.start(), match.end(), call))
for match in DASHED_XML_TOOL_CALL_RE.finditer(text):
call = _openai_call(
match.group("name"),
_xml_arguments(match.group("arguments")),
allowed_names,
)
if call:
candidates.append((match.start(), match.end(), call))
for match in SELF_CLOSING_TOOL_RE.finditer(text):
tag_name = match.group("name")
attributes = _attributes(match.group("attributes"))
if tag_name.casefold() == "tool":
tool_name = attributes.pop("name", None)
arguments = _arguments(
attributes.pop("arguments", attributes.pop("args", ""))
)
if arguments is None:
arguments = attributes
else:
tool_name = tag_name
arguments = attributes
call = _openai_call(tool_name, arguments, allowed_names)
if call:
candidates.append((match.start(), match.end(), call))
for match in TEXTUAL_TOOL_CALL_RE.finditer(text):
arguments = _textual_arguments(match.group("arguments"))
if arguments is None:
continue
call = _openai_call(match.group("name"), arguments, allowed_names)
if call:
candidates.append((match.start(), match.end(), call))
assistant_called = ASSISTANT_CALLED_TOOL_RE.fullmatch(text)
if assistant_called:
call = _openai_call(
assistant_called.group("name"),
_arguments(assistant_called.group("arguments")),
allowed_names,
)
if call:
candidates.append((assistant_called.start(), assistant_called.end(), call))
if not candidates:
fenced_json = FENCED_JSON_RE.fullmatch(text)
raw_json = fenced_json.group("payload") if fenced_json else text.strip()
call = _payload_call(_mapping_literal(raw_json), allowed_names)
if call:
candidates.append((0, len(text), call))
if not candidates:
return [], text
# Different parsers can recognize the same outer wrapper. Keep one result
# per source span while preserving the order produced by the model.
unique: list[dict[str, Any]] = []
seen_spans: set[tuple[int, int]] = set()
for start, end, call in sorted(candidates, key=lambda item: (item[0], item[1])):
span = (start, end)
if span in seen_spans:
continue
seen_spans.add(span)
unique.append(call)
return unique, ""