Fix tool call constrained decoding and parsing for models with native formats (#21593)

python/sglang/srt/constrained/xgrammar_backend.py

@@ -23,6 +23,7 @@ from xgrammar import (
     CompiledGrammar,
     GrammarCompiler,
     GrammarMatcher,
+    StructuralTag,
     StructuralTagItem,
     TokenizerInfo,
     allocate_token_bitmask,
@@ -295,9 +296,11 @@ class XGrammarGrammarBackend(BaseGrammarBackend):
                     )
                     for structure in structural_tag["structures"]
                 ]
-                ctx = self.grammar_compiler.compile_structural_tag(
+                new_tag = StructuralTag.from_legacy_structural_tag(
                     tags, structural_tag["triggers"]
                 )
+                new_tag.format.at_least_one = structural_tag.get("at_least_one", False)
+                ctx = self.grammar_compiler.compile_structural_tag(new_tag)
             else:
                 format_dict = structural_tag.get("format")
                 if isinstance(format_dict, dict):

python/sglang/srt/entrypoints/openai/protocol.py

@@ -166,6 +166,7 @@ class LegacyStructuralTagResponseFormat(BaseModel):
     type: Literal["structural_tag"]
     structures: List[StructuresResponseFormat]
     triggers: List[str]
+    at_least_one: bool = False
 
 
 StructuralTagResponseFormat: TypeAlias = Union[

python/sglang/srt/entrypoints/openai/serving_chat.py

@@ -361,9 +361,11 @@ class OpenAIServingChat(OpenAIServingBase):
                     request.tool_choice,
                     parallel_tool_calls=request.parallel_tool_calls,
                 )
-            # Handle JSON schema constraint directly for required or named tool choice
+            # Fallback: use generic JSON schema for required/named tool choice
-            if request.tool_choice == "required" or isinstance(
+            # only when no parser-specific constraint was set
-                request.tool_choice, ToolChoice
+            if tool_call_constraint is None and (
+                request.tool_choice == "required"
+                or isinstance(request.tool_choice, ToolChoice)
             ):
                 json_schema = get_json_schema_constraint(
                     request.tools,
@@ -1136,22 +1138,56 @@ class OpenAIServingChat(OpenAIServingBase):
     ) -> ToolCallProcessingResult:
         """Process tool calls in the response"""
 
-        # Handle required or named tool choice
+        is_required = tool_choice == "required" or isinstance(tool_choice, ToolChoice)
-        if tool_choice == "required" or (
+
-            isinstance(tool_choice, ToolChoice) and tool_choice.type == "function"
+        # Try model-specific parser when output is in native format.
-        ):
+        # For required/named: only use parser when structural_tag was used
-            # Set finish reason to tool_calls since we're processing tool calls
+        # as constraint (mirrors the streaming path). For auto: always try.
+        if self.tool_call_parser:
+            parser = FunctionCallParser(tools, self.tool_call_parser)
+            should_try_parser = (
+                not is_required or parser.detector.supports_structural_tag()
+            )
+            if should_try_parser and parser.has_tool_call(text):
+                original_finish_type = finish_reason["type"]
+                if finish_reason["type"] == "stop":
+                    finish_reason["type"] = "tool_calls"
+                    finish_reason["matched"] = None
+                try:
+                    text, call_info_list = parser.parse_non_stream(text)
+                    tool_calls = []
+                    for call_info in call_info_list:
+                        tool_id = self._process_tool_call_id(
+                            call_info, history_tool_calls_cnt
+                        )
+                        tool_calls.append(
+                            ToolCall(
+                                id=tool_id,
+                                index=getattr(call_info, "tool_index", None),
+                                function=FunctionResponse(
+                                    name=call_info.name,
+                                    arguments=call_info.parameters,
+                                ),
+                            )
+                        )
+                    return ToolCallProcessingResult(tool_calls, text, finish_reason)
+                except Exception as e:
+                    logger.error(f"Tool call parsing error: {e}")
+                    finish_reason["type"] = original_finish_type
+                    return ToolCallProcessingResult(None, text, finish_reason)
+
+        # json_schema constraint → JSON array output for required/named
+        if is_required:
+            original_finish_type = finish_reason["type"]
             if finish_reason["type"] == "stop":
                 finish_reason["type"] = "tool_calls"
                 finish_reason["matched"] = None
             try:
-                # For required tool choice, we expect a JSON array of tool calls
                 tool_call_data = orjson.loads(text)
                 tool_calls = []
                 for i, tool in enumerate(tool_call_data):
-                    # Create a ToolCallItem from the JSON data
                     call_info = ToolCallItem(
-                        tool_index=i,  # Use the loop index as tool_index
+                        tool_index=i,
                         name=tool["name"],
                         parameters=json.dumps(tool["parameters"], ensure_ascii=False),
                     )
@@ -1171,36 +1207,9 @@ class OpenAIServingChat(OpenAIServingBase):
                         )
                     )
                 return ToolCallProcessingResult(tool_calls, "", finish_reason)
-            except json.JSONDecodeError as e:
-                logger.error(f"Tool call parsing error: {e}")
-                return ToolCallProcessingResult(None, text, finish_reason)
-
-        # Use parser since output is not constrained by JSON schema
-        parser = FunctionCallParser(tools, self.tool_call_parser)
-        if parser.has_tool_call(text):
-            if finish_reason["type"] == "stop":
-                finish_reason["type"] = "tool_calls"
-                finish_reason["matched"] = None
-            try:
-                text, call_info_list = parser.parse_non_stream(text)
-                tool_calls = []
-                for call_info in call_info_list:
-                    tool_id = self._process_tool_call_id(
-                        call_info, history_tool_calls_cnt
-                    )
-                    tool_calls.append(
-                        ToolCall(
-                            id=tool_id,
-                            index=getattr(call_info, "tool_index", None),
-                            function=FunctionResponse(
-                                name=call_info.name, arguments=call_info.parameters
-                            ),
-                        )
-                    )
-                return ToolCallProcessingResult(tool_calls, text, finish_reason)
             except Exception as e:
                 logger.error(f"Tool call parsing error: {e}")
-                # Return error but don't fail the whole request
+                finish_reason["type"] = original_finish_type
                 return ToolCallProcessingResult(None, text, finish_reason)
 
         return ToolCallProcessingResult(None, text, finish_reason)
@@ -1341,11 +1350,26 @@ class OpenAIServingChat(OpenAIServingBase):
     ):
         """Process tool calls in streaming response"""
         if index not in parser_dict:
-            # Use JSON detector directly for required or named tool choice
+            is_required = request.tool_choice == "required" or isinstance(
-            if request.tool_choice == "required" or isinstance(
                 request.tool_choice, ToolChoice
-            ):
+            )
-                parser_dict[index] = JsonArrayParser()
+            # For required/named tool choice: use JsonArrayParser when the
+            # constrained output is plain JSON (detector doesn't support
+            # structural_tag or no parser configured). Use FunctionCallParser
+            # only when the detector supports structural_tag and will produce
+            # native format output.
+            if is_required:
+                use_native_parser = False
+                if self.tool_call_parser:
+                    probe = FunctionCallParser(
+                        tools=request.tools,
+                        tool_call_parser=self.tool_call_parser,
+                    )
+                    use_native_parser = probe.detector.supports_structural_tag()
+                if use_native_parser:
+                    parser_dict[index] = probe
+                else:
+                    parser_dict[index] = JsonArrayParser()
             else:
                 parser_dict[index] = FunctionCallParser(
                     tools=request.tools,

python/sglang/srt/function_call/base_format_detector.py

@@ -270,7 +270,7 @@ class BaseFormatDetector(ABC):
                 cur_arguments = current_tool_call.get("arguments")
                 res = StreamingParseResult()
 
-                if cur_arguments:
+                if cur_arguments is not None:
                     # Calculate how much of the arguments we've already streamed
                     sent = len(self.streamed_args_for_tool[self.current_tool_id])
                     cur_args_json = json.dumps(cur_arguments, ensure_ascii=False)

python/sglang/srt/function_call/deepseekv3_detector.py

@@ -203,7 +203,9 @@ class DeepSeekV3Detector(BaseFormatDetector):
 
     def structure_info(self) -> _GetInfoFunc:
         return lambda name: StructureInfo(
-            begin=">" + name + "\n```json\n",
+            begin="<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>function<｜tool▁sep｜>"
-            end="\n```<",
+            + name
-            trigger=">" + name + "\n```json\n",
+            + "\n```json\n",
+            end="\n```<｜tool▁call▁end｜><｜tool▁calls▁end｜>",
+            trigger="<｜tool▁calls▁begin｜>",
         )

python/sglang/srt/function_call/function_call_parser.py

@@ -3,6 +3,7 @@ from typing import Dict, List, Literal, Optional, Set, Tuple, Type, Union
 
 from sglang.srt.entrypoints.openai.protocol import (
     LegacyStructuralTagResponseFormat,
+    StructuralTagResponseFormat,
     StructuresResponseFormat,
     Tool,
     ToolCallConstraint,
@@ -32,7 +33,10 @@ from sglang.srt.function_call.qwen3_coder_detector import Qwen3CoderDetector
 from sglang.srt.function_call.qwen25_detector import Qwen25Detector
 from sglang.srt.function_call.step3_detector import Step3Detector
 from sglang.srt.function_call.trinity_detector import TrinityDetector
-from sglang.srt.function_call.utils import get_json_schema_constraint
+from sglang.srt.function_call.utils import (
+    _get_tool_schema_defs,
+    get_json_schema_constraint,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -146,12 +150,24 @@ class FunctionCallParser:
 
         return final_normal_text, final_calls
 
-    def get_structure_tag(self) -> LegacyStructuralTagResponseFormat:
+    def get_structure_tag(
+        self, at_least_one: bool = False
+    ) -> StructuralTagResponseFormat:
         """
         Generate a structural tag response format for all available tools.
 
         This creates the necessary structural tags that guide the model's output format.
+
+        Args:
+            at_least_one: If True, the grammar forces at least one tool call
+                (no free text allowed). Used for required/named tool_choice.
+
+        Raises:
+            ValueError: If tools have conflicting $defs schemas.
         """
+        # Validate $defs consistency before building structural tags
+        _get_tool_schema_defs(self.tools)
+
         tool_structures: List[StructuresResponseFormat] = list()
         tool_trigger_set: Set[str] = set()
 
@@ -183,6 +199,7 @@ class FunctionCallParser:
             type="structural_tag",
             structures=tool_structures,
             triggers=list(tool_trigger_set),
+            at_least_one=at_least_one,
         )
 
     def get_structure_constraint(
@@ -203,16 +220,23 @@ class FunctionCallParser:
         """
         # NOTE: structural_tag only supports JSON-compatible content between the begin and end.
         # It cannot parse or validate function call Pythonic or XML-ish syntax.
-        if (
+        if self.detector.supports_structural_tag():
-            self.detector.supports_structural_tag()
+            # For "required"/named: always use structural_tag to preserve the
-            and tool_choice == "auto"
+            # model's native tool call format. Schema is only included when
-            and (
+            # strict=True, per OpenAI protocol semantics.
-                any(tool.function.strict for tool in self.tools)
+            # For "auto": only constrain when strict is enabled.
-                or self.tool_strict_level >= ToolStrictLevel.FUNCTION
+            is_required = tool_choice == "required" or isinstance(
+                tool_choice, ToolChoice
             )
-        ):
+            if is_required or (
-            tag = self.get_structure_tag()
+                tool_choice == "auto"
-            return ("structural_tag", tag)
+                and (
+                    any(tool.function.strict for tool in self.tools)
+                    or self.tool_strict_level >= ToolStrictLevel.FUNCTION
+                )
+            ):
+                tag = self.get_structure_tag(at_least_one=is_required)
+                return ("structural_tag", tag)
         elif tool_choice == "required" or isinstance(tool_choice, ToolChoice):
             json_schema = get_json_schema_constraint(
                 self.tools, tool_choice, parallel_tool_calls=parallel_tool_calls

test/registered/openai_server/basic/test_serving_chat.py

@@ -822,5 +822,77 @@ class ServingChatTestCase(unittest.TestCase):
         self.assertIn("must be an integer", context.exception.detail)
 
 
+class TestProcessToolCallsWithRequiredToolChoice(unittest.TestCase):
+    """Test _process_tool_calls with tool_choice='required' uses model-specific parser."""
+
+    def setUp(self):
+        tm = _MockTokenizerManager()
+        tm.server_args.tool_call_parser = "kimi_k2"
+        self.chat = OpenAIServingChat(tm, _MockTemplateManager())
+
+    def test_required_with_parser_uses_function_call_parser(self):
+        """tool_choice='required' should use FunctionCallParser when tool_call_parser is set."""
+        with patch(
+            "sglang.srt.entrypoints.openai.serving_chat.FunctionCallParser"
+        ) as ParserMock:
+            call_info = Mock()
+            call_info.name = "get_weather"
+            call_info.parameters = '{"location":"Tokyo"}'
+            call_info.tool_index = 0
+
+            parser_instance = ParserMock.return_value
+            parser_instance.has_tool_call.return_value = True
+            parser_instance.parse_non_stream.return_value = ("", [call_info])
+
+            finish_reason = {"type": "stop", "matched": None}
+            tools = [{"type": "function", "function": {"name": "get_weather"}}]
+
+            tool_calls, text, fr = self.chat._process_tool_calls(
+                text="<|tool_calls_section_begin|>...<|tool_calls_section_end|>",
+                tools=tools,
+                finish_reason=finish_reason,
+                tool_choice="required",
+            )
+
+            self.assertIsNotNone(tool_calls)
+            self.assertEqual(len(tool_calls), 1)
+            self.assertEqual(tool_calls[0].function.name, "get_weather")
+            self.assertEqual(fr["type"], "tool_calls")
+
+    def test_required_without_parser_falls_back_to_json(self):
+        """tool_choice='required' without parser should parse as JSON array."""
+        self.chat.tool_call_parser = None
+
+        finish_reason = {"type": "stop", "matched": None}
+        tools = [{"type": "function", "function": {"name": "get_weather"}}]
+
+        tool_calls, text, fr = self.chat._process_tool_calls(
+            text='[{"name":"get_weather","parameters":{"location":"Tokyo"}}]',
+            tools=tools,
+            finish_reason=finish_reason,
+            tool_choice="required",
+        )
+
+        self.assertIsNotNone(tool_calls)
+        self.assertEqual(len(tool_calls), 1)
+        self.assertEqual(tool_calls[0].function.name, "get_weather")
+
+    def test_required_without_parser_invalid_json_returns_none(self):
+        """tool_choice='required' without parser and invalid JSON returns tool_calls=None."""
+        self.chat.tool_call_parser = None
+
+        finish_reason = {"type": "stop", "matched": None}
+        tools = [{"type": "function", "function": {"name": "get_weather"}}]
+
+        tool_calls, text, fr = self.chat._process_tool_calls(
+            text="<|tool_calls_section_begin|>not json",
+            tools=tools,
+            finish_reason=finish_reason,
+            tool_choice="required",
+        )
+
+        self.assertIsNone(tool_calls)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

test/registered/openai_server/function_call/test_tool_choice.py

@@ -348,8 +348,12 @@ class TestToolChoiceLlama32(CustomTestCase):
         self.assertEqual(found_name, "get_weather")
 
     def test_required_streaming_arguments_chunks_json(self):
-        """In streaming required mode, complete tool call arguments should be valid JSON when all chunks are combined"""
+        """In streaming required mode, complete tool call arguments should be valid JSON when all chunks are combined.
+        Uses strict=True so the grammar enforces the parameter schema."""
         tools = self.get_test_tools()
+        # Add strict=True so arguments are schema-constrained
+        for tool in tools:
+            tool["function"]["strict"] = True
         messages = self.get_test_messages()
 
         response = self.client.chat.completions.create(
@@ -406,13 +410,15 @@ class TestToolChoiceLlama32(CustomTestCase):
                 )
 
     def test_complex_parameters_required_non_streaming(self):
-        """Validate complex nested parameter schemas in non-streaming required mode"""
+        """Validate complex nested parameter schemas in non-streaming required mode.
+        Uses strict=True so the grammar enforces the parameter schema."""
         complex_tools = [
             {
                 "type": "function",
                 "function": {
                     "name": "analyze_data",
                     "description": "Analyze complex data structures",
+                    "strict": True,
                     "parameters": {
                         "type": "object",
                         "properties": {

test/registered/unit/function_call/test_function_call_parser.py

@@ -3925,6 +3925,119 @@ function call<|role_sep|>
         self.assertEqual(params["city"], "Rome")
 
 
+class TestGetStructureConstraint(unittest.TestCase):
+    """Tests for FunctionCallParser.get_structure_constraint() logic.
+
+    Verifies that detectors supporting structural_tag use it for required/named
+    tool_choice, and that the generic json_schema fallback is used otherwise.
+    """
+
+    def _make_tools(self, strict=False):
+        return [
+            Tool(
+                type="function",
+                function=Function(
+                    name="get_weather",
+                    description="Get weather",
+                    parameters={
+                        "type": "object",
+                        "properties": {"city": {"type": "string"}},
+                        "required": ["city"],
+                    },
+                    strict=strict,
+                ),
+            ),
+        ]
+
+    def _make_parser(self, parser_name, strict=False):
+        from sglang.srt.function_call.function_call_parser import FunctionCallParser
+
+        return FunctionCallParser(self._make_tools(strict=strict), parser_name)
+
+    # --- structural_tag detectors (kimi_k2, deepseekv3, qwen25, etc.) ---
+
+    def test_kimi_required_strict_returns_structural_tag(self):
+        parser = self._make_parser("kimi_k2", strict=True)
+        result = parser.get_structure_constraint("required")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+        self.assertTrue(result[1].at_least_one)
+
+    def test_kimi_required_no_strict_returns_structural_tag(self):
+        """required should use structural_tag even without strict, to preserve native format."""
+        parser = self._make_parser("kimi_k2", strict=False)
+        result = parser.get_structure_constraint("required")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+        self.assertTrue(result[1].at_least_one)
+
+    def test_kimi_auto_strict_returns_structural_tag(self):
+        parser = self._make_parser("kimi_k2", strict=True)
+        result = parser.get_structure_constraint("auto")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+        self.assertFalse(result[1].at_least_one)
+
+    def test_kimi_auto_no_strict_returns_none(self):
+        """auto without strict should not constrain."""
+        parser = self._make_parser("kimi_k2", strict=False)
+        result = parser.get_structure_constraint("auto")
+        self.assertIsNone(result)
+
+    def test_kimi_named_tool_choice_returns_structural_tag(self):
+        from sglang.srt.entrypoints.openai.protocol import (
+            ToolChoice,
+            ToolChoiceFuncName,
+        )
+
+        parser = self._make_parser("kimi_k2", strict=False)
+        tool_choice = ToolChoice(function=ToolChoiceFuncName(name="get_weather"))
+        result = parser.get_structure_constraint(tool_choice)
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+
+    def test_deepseekv3_required_no_strict_returns_structural_tag(self):
+        parser = self._make_parser("deepseekv3", strict=False)
+        result = parser.get_structure_constraint("required")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+
+    def test_qwen25_required_no_strict_returns_structural_tag(self):
+        parser = self._make_parser("qwen25", strict=False)
+        result = parser.get_structure_constraint("required")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], "structural_tag")
+
+    # --- structural_tag content verification ---
+
+    def test_kimi_structural_tag_has_kimi_tokens(self):
+        """Verify structural_tag contains kimi-specific special tokens."""
+        parser = self._make_parser("kimi_k2", strict=True)
+        result = parser.get_structure_constraint("required")
+        tag = result[1]
+        structures = tag.structures
+        self.assertTrue(len(structures) > 0)
+        self.assertIn("<|tool_calls_section_begin|>", structures[0].begin)
+        self.assertIn("<|tool_call_end|>", structures[0].end)
+
+    def test_kimi_required_no_strict_uses_empty_schema(self):
+        """Without strict, structural_tag should use empty schema per OpenAI
+        protocol: strict=False means no parameter schema enforcement."""
+        parser = self._make_parser("kimi_k2", strict=False)
+        result = parser.get_structure_constraint("required")
+        tag = result[1]
+        self.assertEqual(tag.structures[0].schema_, {})
+
+    def test_kimi_required_strict_uses_tool_schema(self):
+        """With strict, structural_tag should include the tool's parameter schema."""
+        parser = self._make_parser("kimi_k2", strict=True)
+        result = parser.get_structure_constraint("required")
+        tag = result[1]
+        schema = tag.structures[0].schema_
+        self.assertIn("properties", schema)
+        self.assertIn("city", schema["properties"])
+
+
 class TestQwen25Detector(unittest.TestCase):
     """Test Qwen25Detector streaming and non-streaming multi-tool-call parsing."""