"""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"<(?Ptool_call|function_call)>\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"<(?Ptool_call|function_call)>\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 _looks_like_tool_payload(payload: object) -> bool: """Return whether a mapping has the OpenAI/Qwen tool-call shape.""" if not isinstance(payload, Mapping): return False function = payload.get("function") if isinstance(function, Mapping): return isinstance(function.get("name"), str) and bool(function.get("name")) return isinstance(payload.get("name"), str) and bool(payload.get("name")) def _terminal_json_tool_payload(text: str) -> tuple[int, int, Mapping[str, Any]] | None: """Recover a complete bare JSON tool call at the end of model output. Small coder models sometimes obey the JSON schema but omit the surrounding ```` tags, occasionally after a short explanatory prefix. The normal extractor can parse a *pure* JSON response, but the generation stopping criterion previously failed to stop there, allowing the model to continue with prose and additional simulated calls. Scan JSON-object starts and accept only a terminal mapping with a tool-call shape. """ candidate_text = text.rstrip() decoder = json.JSONDecoder() for start, char in enumerate(candidate_text): if char != "{": continue try: payload, consumed = decoder.raw_decode(candidate_text[start:]) except json.JSONDecodeError: continue end = start + consumed if candidate_text[end:].strip(): continue if _looks_like_tool_payload(payload): return start, len(candidate_text), payload return None def has_complete_tool_call( text: str, allowed_names: set[str] | None = None, ) -> bool: """Return true once generation ended a valid supported tool-call form. When ``allowed_names`` is supplied, a syntactically complete hallucinated call to an unadvertised function must *not* stop generation. This matters for OpenClaude because the final parser rejects unadvertised names. """ if allowed_names is not None: calls, _ = extract_tool_calls(text, allowed_names) return bool(calls) fenced_json = FENCED_JSON_RE.fullmatch(text) 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 ( fenced_json and _looks_like_tool_payload(_mapping_literal(fenced_json.group("payload"))) ) or _terminal_json_tool_payload(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 Qwen2.5'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 recover_forced_tool_call(text: str, tool_name: str) -> dict[str, Any] | None: """Recover argument-only JSON when exactly one tool is mandated. Some OpenAI-compatible coder models occasionally emit only the function argument object when the caller has already forced a single tool. The normal parser correctly refuses to guess a function name from that object. In the *single forced-tool* case, however, the name is unambiguous and the structured OpenAI call can be reconstructed safely without executing or evaluating arbitrary text. """ raw = text.strip() fenced = FENCED_JSON_RE.fullmatch(raw) if fenced: raw = fenced.group("payload") payload = _mapping_literal(raw) if not isinstance(payload, Mapping): return None if _looks_like_tool_payload(payload): return None arguments: Mapping[str, Any] = payload nested_arguments = payload.get("arguments") if len(payload) == 1 and isinstance(nested_arguments, Mapping): arguments = nested_arguments return _openai_call(tool_name, arguments, {tool_name}) 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: terminal_json = _terminal_json_tool_payload(text) if terminal_json is not None: start, end, payload = terminal_json call = _payload_call(payload, allowed_names) if call: candidates.append((start, end, 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, ""