fix: add plain JSON mode override for LLM requests (#60)

Co-authored-by: Vineeth Sai Narajala <vnarajal@cisco.com>

Author: barryqy

.env.example

@@ -10,6 +10,7 @@
 # SKILL_SCANNER_LLM_MODEL=azure/gpt-4.1
 # SKILL_SCANNER_LLM_BASE_URL=https://your-resource.openai.azure.com/
 # SKILL_SCANNER_LLM_API_VERSION=2025-01-01-preview
+# SKILL_SCANNER_LLM_FORCE_JSON_OBJECT=true
 
 # AWS Bedrock (bearer token)
 # SKILL_SCANNER_LLM_API_KEY=bedrock-api-key-...

docs/reference/configuration-reference.md

@@ -27,6 +27,7 @@ Primary settings for the LLM semantic analyzer.
 | `SKILL_SCANNER_LLM_MODEL` | Primary model identifier for semantic analysis. | `anthropic/claude-sonnet-4-20250514` |
 | `SKILL_SCANNER_LLM_BASE_URL` | Optional custom endpoint base URL for provider routing. | `https://api.openai.com/v1` |
 | `SKILL_SCANNER_LLM_API_VERSION` | Optional API version for providers that require one. | `2024-02-15-preview` |
+| `SKILL_SCANNER_LLM_FORCE_JSON_OBJECT` | Skip json_schema and start in plain JSON mode for incompatible proxies. | `true` |
 
 ## Meta Analyzer
 
@@ -117,6 +118,7 @@ Paths, allowlists, and other advanced settings.
 | `SKILL_SCANNER_LLM_API_KEY` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/config/config.py`, `skill_scanner/core/analyzer_factory.py`, `skill_scanner/core/analyzers/behavioral_analyzer.py`, `skill_scanner/core/analyzers/llm_analyzer.py`, `skill_scanner/core/analyzers/llm_provider_config.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |
 | `SKILL_SCANNER_LLM_API_VERSION` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/core/analyzer_factory.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |
 | `SKILL_SCANNER_LLM_BASE_URL` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/core/analyzer_factory.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |
+| `SKILL_SCANNER_LLM_FORCE_JSON_OBJECT` | `.env.example` |
 | `SKILL_SCANNER_LLM_MODEL` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/config/config.py`, `skill_scanner/core/analyzer_factory.py`, `skill_scanner/core/analyzers/behavioral_analyzer.py`, `skill_scanner/core/analyzers/llm_analyzer.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |
 | `SKILL_SCANNER_META_LLM_API_KEY` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |
 | `SKILL_SCANNER_META_LLM_API_VERSION` | `.env.example`, `skill_scanner/cli/cli.py`, `skill_scanner/core/analyzers/meta_analyzer.py` |

docs/user-guide/installation-and-configuration.md

@@ -54,6 +54,7 @@ You only need to set these if you're using the corresponding features. Click a s
 - `SKILL_SCANNER_LLM_MODEL`
 - `SKILL_SCANNER_LLM_BASE_URL`
 - `SKILL_SCANNER_LLM_API_VERSION`
+- `SKILL_SCANNER_LLM_FORCE_JSON_OBJECT` — start in plain JSON mode for proxies that reject `json_schema`
 
 </details>
 

scripts/generate_reference_docs.py

@@ -393,6 +393,7 @@ def _describe_env_var(var: str) -> str:
         "SKILL_SCANNER_LLM_MODEL": "Primary model identifier for semantic analysis.",
         "SKILL_SCANNER_LLM_BASE_URL": "Optional custom endpoint base URL for provider routing.",
         "SKILL_SCANNER_LLM_API_VERSION": "Optional API version for providers that require one.",
+        "SKILL_SCANNER_LLM_FORCE_JSON_OBJECT": "Skip json_schema and start in plain JSON mode for incompatible proxies.",
         "SKILL_SCANNER_META_LLM_API_KEY": "Meta-analyzer API key override.",
         "SKILL_SCANNER_META_LLM_MODEL": "Meta-analyzer model override.",
         "SKILL_SCANNER_META_LLM_BASE_URL": "Meta-analyzer base URL override.",
@@ -426,6 +427,7 @@ def _describe_env_var(var: str) -> str:
             "SKILL_SCANNER_LLM_MODEL",
             "SKILL_SCANNER_LLM_BASE_URL",
             "SKILL_SCANNER_LLM_API_VERSION",
+            "SKILL_SCANNER_LLM_FORCE_JSON_OBJECT",
         ],
     ),
     (
@@ -484,6 +486,7 @@ def _describe_env_var(var: str) -> str:
     "SKILL_SCANNER_LLM_MODEL": "anthropic/claude-sonnet-4-20250514",
     "SKILL_SCANNER_LLM_BASE_URL": "https://api.openai.com/v1",
     "SKILL_SCANNER_LLM_API_VERSION": "2024-02-15-preview",
+    "SKILL_SCANNER_LLM_FORCE_JSON_OBJECT": "true",
     "SKILL_SCANNER_META_LLM_API_KEY": "(falls back to LLM_API_KEY)",
     "SKILL_SCANNER_META_LLM_MODEL": "(falls back to LLM_MODEL)",
     "SKILL_SCANNER_META_LLM_BASE_URL": "(falls back to LLM_BASE_URL)",

skill_scanner/core/analyzers/llm_request_handler.py

@@ -25,6 +25,7 @@
 import asyncio
 import json
 import logging
+import os
 import warnings
 from pathlib import Path
 from typing import Any
@@ -95,6 +96,12 @@ def __init__(
 
         # Load JSON schema for structured outputs
         self.response_schema = self._load_response_schema()
+        self._use_plain_json_output = self._env_flag_enabled("SKILL_SCANNER_LLM_FORCE_JSON_OBJECT")
+
+    def _env_flag_enabled(self, env_name: str) -> bool:
+        """Treat common truthy env values as enabled."""
+        raw_value = os.getenv(env_name, "")
+        return raw_value.strip().lower() in {"1", "true", "yes", "on"}
 
     def _load_response_schema(self) -> dict[str, Any] | None:
         """Load JSON schema for structured outputs."""
@@ -160,6 +167,55 @@ def _sanitize_schema_for_google(self, schema: dict[str, Any]) -> dict[str, Any]:
 
         return sanitized
 
+    def _should_use_json_object(self) -> bool:
+        """Pick the safest response format for the current backend."""
+        if self._use_plain_json_output:
+            return True
+
+        model_lower = self.provider_config.model.lower()
+        unsupported_json_schema_providers = ["deepseek"]
+        return any(name in model_lower for name in unsupported_json_schema_providers)
+
+    def _build_response_format(self) -> dict[str, Any] | None:
+        """Build the response format for LiteLLM requests."""
+        if not self.response_schema:
+            return None
+
+        if self._should_use_json_object():
+            return {"type": "json_object"}
+
+        return {
+            "type": "json_schema",
+            "json_schema": {
+                "name": "security_analysis_response",
+                "schema": self.response_schema,
+                "strict": True,
+            },
+        }
+
+    def _should_fallback_to_json_object(self, error: Exception, response_format: dict[str, Any] | None) -> bool:
+        """Detect backends that reject structured output and need plain JSON mode."""
+        if not response_format or response_format.get("type") != "json_schema":
+            return False
+
+        error_msg = str(error).lower()
+        if "response_format.json_schema" in error_msg:
+            return True
+
+        if "json_schema" in error_msg and any(
+            phrase in error_msg
+            for phrase in [
+                "missing required parameter",
+                "unsupported",
+                "not supported",
+                "invalid",
+                "unknown parameter",
+            ]
+        ):
+            return True
+
+        return False
+
     async def make_request(self, messages: list[dict[str, str]], context: str = "") -> str:
         """
         Make LLM request with retry logic and exponential backoff.
@@ -206,30 +262,28 @@ async def _make_litellm_request(self, messages: list[dict[str, str]], context: s
                     **self.provider_config.get_request_params(),
                 }
 
-                # Add structured output support using LiteLLM's unified format
-                # According to LiteLLM docs: https://docs.litellm.ai/docs/completion/json_mode
-                # Format: response_format={ "type": "json_schema", "json_schema": { "name": "...", "schema": {...}, "strict": true } }
-                # Works for: OpenAI, Anthropic Claude, Gemini (via LiteLLM), Bedrock, Vertex AI, Groq, Ollama, Databricks
-                if self.response_schema:
-                    model_lower = self.provider_config.model.lower()
-                    unsupported_json_schema_providers = ["deepseek"]
-                    if any(p in model_lower for p in unsupported_json_schema_providers):
-                        request_params["response_format"] = {"type": "json_object"}
-                    else:
-                        request_params["response_format"] = {
-                            "type": "json_schema",
-                            "json_schema": {
-                                "name": "security_analysis_response",
-                                "schema": self.response_schema,
-                                "strict": True,
-                            },
-                        }
+                response_format = self._build_response_format()
+                if response_format:
+                    request_params["response_format"] = response_format
 
                 response = await acompletion(**request_params, drop_params=True)
                 content: str = response.choices[0].message.content or ""
                 return content
 
             except Exception as e:
+                response_format = request_params.get("response_format")
+                if self._should_fallback_to_json_object(e, response_format):
+                    logger.warning(
+                        "Structured output rejected for %s, retrying with plain JSON output",
+                        context,
+                    )
+                    self._use_plain_json_output = True
+                    retry_params = dict(request_params)
+                    retry_params["response_format"] = {"type": "json_object"}
+                    response = await acompletion(**retry_params, drop_params=True)
+                    content: str = response.choices[0].message.content or ""
+                    return content
+
                 last_exception = e
                 error_msg = str(e).lower()
 

tests/test_llm_request_handler.py

@@ -21,8 +21,9 @@
 """
 
 import json
+import os
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -172,3 +173,60 @@ async def test_acompletion_called_with_drop_params(self):
         call_kwargs = mock_acompletion.call_args
         kwargs = call_kwargs.kwargs if call_kwargs.kwargs else call_kwargs[1]
         assert kwargs.get("drop_params") is True, f"acompletion must be called with drop_params=True, got: {kwargs}"
+
+
+class TestLiteLLMRequestFallback:
+    """Tests for switching from json_schema to plain JSON output."""
+
+    @pytest.fixture
+    def litellm_handler(self) -> LLMRequestHandler:
+        provider_config = MagicMock()
+        provider_config.model = "gpt-4o"
+        provider_config.use_google_sdk = False
+        provider_config.get_request_params.return_value = {}
+        return LLMRequestHandler(provider_config=provider_config, max_retries=0)
+
+    @staticmethod
+    def _mock_litellm_response(content: str) -> MagicMock:
+        response = MagicMock()
+        response.choices = [MagicMock(message=MagicMock(content=content))]
+        return response
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_json_object_when_backend_rejects_schema(self, litellm_handler: LLMRequestHandler):
+        error = RuntimeError("Azure error: Missing required parameter: 'response_format.json_schema'.")
+        plain_json_response = TestLiteLLMRequestFallback._mock_litellm_response(
+            '{"overall_assessment":"unsafe","findings":[]}'
+        )
+
+        with patch(
+            "skill_scanner.core.analyzers.llm_request_handler.acompletion",
+            AsyncMock(side_effect=[error, plain_json_response]),
+        ) as mocked_acompletion:
+            result = await litellm_handler.make_request([{"role": "user", "content": "Scan this"}], context="demo")
+
+        assert result == '{"overall_assessment":"unsafe","findings":[]}'
+        assert mocked_acompletion.await_count == 2
+        assert mocked_acompletion.await_args_list[0].kwargs["response_format"]["type"] == "json_schema"
+        assert mocked_acompletion.await_args_list[1].kwargs["response_format"]["type"] == "json_object"
+        assert litellm_handler._use_plain_json_output is True
+
+    @pytest.mark.asyncio
+    async def test_force_json_object_env_skips_schema_attempt(self, litellm_handler: LLMRequestHandler):
+        plain_json_response = TestLiteLLMRequestFallback._mock_litellm_response(
+            '{"overall_assessment":"unsafe","findings":[]}'
+        )
+
+        with (
+            patch.dict(os.environ, {"SKILL_SCANNER_LLM_FORCE_JSON_OBJECT": "1"}, clear=False),
+            patch(
+                "skill_scanner.core.analyzers.llm_request_handler.acompletion",
+                AsyncMock(return_value=plain_json_response),
+            ) as mocked_acompletion,
+        ):
+            forced_handler = LLMRequestHandler(provider_config=litellm_handler.provider_config, max_retries=0)
+            result = await forced_handler.make_request([{"role": "user", "content": "Scan this"}], context="demo")
+
+        assert result == '{"overall_assessment":"unsafe","findings":[]}'
+        assert mocked_acompletion.await_count == 1
+        assert mocked_acompletion.await_args_list[0].kwargs["response_format"]["type"] == "json_object"