[gemini] Structured output (json_schema) cannot be combined with grounding/url_context — needs Interactions API migration for Gemini 3+

[hpiclaranet]

Description

When using gemini-3.5-flash (or any Gemini 3+ model) with both Structured Output (json_schema) and Google Search grounding (or url_context / code_execution) enabled simultaneously, the plugin raises:

PluginInvokeError: "Structured output (json_schema) cannot be used with: grounding, url_context"

This makes it impossible to get both grounded (web-searched) answers AND structured JSON output in a single LLM node — a common requirement for agentic workflows.

Root Cause

The plugin's _generate() calls the legacy generateContent endpoint (genai_client.models.generate_content()). A model-agnostic static validator _validate_feature_compatibility() hard-blocks json_schema whenever any other feature is active (Rule 1: "json_schema is exclusive").

This rule was correct for Gemini <=2.5 models, where the Google API genuinely rejects the combination:

400 INVALID_ARGUMENT: "controlled generation is not supported with google_search tool"

However, Gemini 3+ models support combining Structured Outputs with built-in tools natively — but only via the Interactions API (client.interactions.create()), which is GA since June 2026 and recommended by Google for all new development.

On the legacy generateContent endpoint, the combination silently de-grounds on GA Gemini 3 models (HTTP 200, but groundingMetadata is empty — model answers from parametric memory only, no error). See google-gemini/cookbook#1274 for a deterministic repro.

Evidence

1. Google docs (structured-output, updated 2026-07-07):

   "Gemini 3 lets you combine Structured Outputs with built-in tools, including Grounding with Google Search, URL Context, Code Execution, File Search, and Function Calling."

2. Migration guide (migrate-to-interactions): all structured-output + tools examples use client.interactions.create(), not models.generate_content().

3. google_search grounding silently disabled when structured/JSON output is requested (gemini-3.5-flash) google-gemini/cookbook#1274 (Jun 22, 2026): deterministic silent de-grounding on gemini-3.5-flash via generateContent — 0/5 grounded in JSON mode vs 5/5 in prose mode.

4. [Feature Request] Support Google Search Grounding with Structured Outputs - Gemini 3 vercel/ai#11599 (closed) + #11815 (closed): combination works on preview models (gemini-3-pro-preview, gemini-3-flash-preview) but not GA models via the legacy endpoint.

5. SDK: google-genai>=1.55.0 already supports the Interactions API (client.interactions.create()). The plugin's current google-genai dependency satisfies this.

Proposed Fix

Add a conditional Interactions API code path for Gemini 3+ models when json_schema + native tools are both active. Keep generateContent as the default for everything else (backward compatible).

1. Make _validate_feature_compatibility model-aware

@staticmethod
def _validate_feature_compatibility(
    model_parameters: Mapping[str, Any],
    tools: Optional[list[PromptMessageTool]] = None,
    model: Optional[str] = None,          # <-- NEW param
) -> dict[str, Any]:
    # ... existing code ...

    # Rule 1: json_schema is mutually exclusive with all other features
    # Exception: Gemini 3+ models support combining json_schema with
    # native tools via the Interactions API.
    if features["json_schema"] and len(enabled_features) > 1:
        _has_native_tools = any(
            adjusted_params.get(f)
            for f in ["grounding", "url_context", "code_execution"]
        )
        if model and model.startswith("gemini-3") and _has_native_tools and not tools:
            pass  # Gemini 3+ with native tools -> Interactions API path
        else:
            other_features = [f for f in enabled_features if f != "json_schema"]
            raise InvokeError(
                f"Structured output (json_schema) cannot be used with: {', '.join(other_features)}"
            )

2. Route to Interactions API in _generate()

# After _validate_feature_compatibility(model_parameters, tools, model=model):
_is_gemini3 = model.startswith("gemini-3")
_has_json_schema = bool(model_parameters.get("json_schema"))
_has_native_tools = any(
    model_parameters.get(f) for f in ["grounding", "url_context", "code_execution"]
)
_has_custom_tools = bool(tools)
if _is_gemini3 and _has_json_schema and _has_native_tools and not _has_custom_tools:
    return self._generate_via_interactions(
        model, credentials, prompt_messages, model_parameters, tools, stop, stream
    )

3. New _generate_via_interactions()

def _generate_via_interactions(
    self, model, credentials, prompt_messages, model_parameters,
    tools=None, stop=None, stream=True,
) -> Union[LLMResult, Generator[LLMResultChunk]]:
    genai_client = genai.Client(
        api_key=credentials["google_api_key"],
        http_options=types.HttpOptions(
            base_url=credentials.get("google_base_url", None)
        ),
    )

    # Build contents via existing method (captures system_instruction as side-effect)
    config = types.GenerateContentConfig()
    contents = self._build_gemini_contents(
        prompt_messages=prompt_messages, genai_client=genai_client,
        config=config,
        file_server_url_prefix=credentials.get("file_url") or None,
        model_parameters=model_parameters,
    )

    # Extract system_instruction (str or Content -> str)
    system_instruction = None
    if config.system_instruction:
        if isinstance(config.system_instruction, str):
            system_instruction = config.system_instruction
        elif hasattr(config.system_instruction, "parts") and config.system_instruction.parts:
            system_instruction = "\n".join(
                p.text for p in config.system_instruction.parts
                if hasattr(p, "text") and p.text
            ) or None

    # Tools in Interactions API dict format
    interactions_tools = []
    if model_parameters.get("grounding"):
        interactions_tools.append({"type": "google_search"})
    if model_parameters.get("url_context"):
        interactions_tools.append({"type": "url_context"})
    if model_parameters.get("code_execution"):
        interactions_tools.append({"type": "code_execution"})

    # response_format: single dict, mime_type INSIDE (no separate response_mime_type)
    response_format = None
    if schema := model_parameters.get("json_schema"):
        if isinstance(schema, str):
            schema = json.loads(schema)
        response_format = {
            "type": "text",
            "mime_type": "application/json",
            "schema": schema,
        }

    # generation_config: Interactions API subset
    # NOTE: top_k, thinking_budget, include_thoughts, media_resolution NOT available
    gen_config = {}
    if model_parameters.get("temperature") is not None:
        gen_config["temperature"] = float(model_parameters["temperature"])
    if model_parameters.get("max_output_tokens") is not None:
        gen_config["max_output_tokens"] = int(model_parameters["max_output_tokens"])
    if model_parameters.get("top_p") is not None:
        gen_config["top_p"] = float(model_parameters["top_p"])
    if stop:
        gen_config["stop_sequences"] = stop
    if tl := model_parameters.get("thinking_level"):
        # Interactions API expects lowercase: "minimal"/"low"/"medium"/"high"
        gen_config["thinking_level"] = {
            "Minimal": "minimal", "Low": "low",
            "Medium": "medium", "High": "high",
        }.get(tl, "medium")

    # Convert types.Content -> Interactions API Step format
    # (Content objects have "parts" which the API rejects: "Unknown parameter 'parts'")
    if not contents and system_instruction:
        contents = [types.Content(
            role="user", parts=[types.Part.from_text(text=system_instruction)]
        )]
    interactions_input = []
    for content in contents:
        step_type = "model_output" if content.role == "model" else "user_input"
        step_content = []
        for part in content.parts:
            if hasattr(part, "text") and part.text:
                step_content.append({"type": "text", "text": part.text})
            elif hasattr(part, "inline_data") and part.inline_data:
                inline = part.inline_data
                mime = (inline.mime_type or "").lower()
                b64 = base64.b64encode(inline.data).decode() if inline.data else ""
                if mime.startswith("image/"):
                    step_content.append({"type": "image", "mime_type": mime, "data": b64})
                elif mime.startswith("audio/"):
                    step_content.append({"type": "audio", "mime_type": mime, "data": b64})
                elif mime.startswith("video/"):
                    step_content.append({"type": "video", "mime_type": mime, "data": b64})
                elif mime == "application/pdf":
                    step_content.append({"type": "document", "mime_type": mime, "data": b64})
        if step_content:
            interactions_input.append({"type": step_type, "content": step_content})

    kwargs = {"model": model, "input": interactions_input, "store": False}
    if system_instruction:
        kwargs["system_instruction"] = system_instruction
    if interactions_tools:
        kwargs["tools"] = interactions_tools
    if response_format:
        kwargs["response_format"] = response_format
    if gen_config:
        kwargs["generation_config"] = gen_config

    if stream:
        response = genai_client.interactions.create(stream=True, **kwargs)
        return self._handle_interactions_stream_response(
            model, credentials, response, prompt_messages, genai_client
        )
    interaction = genai_client.interactions.create(**kwargs)
    return self._handle_interactions_response(
        model, credentials, interaction, prompt_messages
    )

4. New _handle_interactions_response()

def _handle_interactions_response(
    self, model, credentials, interaction, prompt_messages,
) -> LLMResult:
    # Extract text (convenience property with manual fallback)
    text = ""
    if hasattr(interaction, "output_text") and interaction.output_text:
        text = interaction.output_text
    elif hasattr(interaction, "steps") and interaction.steps:
        for step in interaction.steps:
            if getattr(step, "type", None) == "model_output" and getattr(step, "content", None):
                for part in step.content:
                    if getattr(part, "type", None) == "text" and getattr(part, "text", None):
                        text += part.text

    assistant_prompt_message = AssistantPromptMessage(
        content=[TextPromptMessageContent(data=text)] if text else []
    )

    # Token usage (field names differ from generateContent)
    prompt_tokens, completion_tokens = 0, 0
    if usage := getattr(interaction, "usage", None):
        prompt_tokens = getattr(usage, "total_input_tokens", 0) or 0
        completion_tokens = (
            getattr(usage, "total_output_tokens", 0) or 0
        ) + (
            getattr(usage, "total_thought_tokens", 0) or 0
        )

    if prompt_tokens == 0 or completion_tokens == 0:
        prompt_tokens = self.get_num_tokens(model, credentials, prompt_messages)
        completion_tokens = self.get_num_tokens(model, credentials, [assistant_prompt_message])

    usage = self._calc_response_usage(
        model=model, credentials=dict(credentials),
        prompt_tokens=prompt_tokens, completion_tokens=completion_tokens,
    )
    return LLMResult(
        model=model, prompt_messages=prompt_messages,
        message=assistant_prompt_message, usage=usage,
    )

5. New _handle_interactions_stream_response()

def _handle_interactions_stream_response(
    self, model, credentials, response, prompt_messages, genai_client=None,
) -> Generator[LLMResultChunk]:
    # Keep reference to prevent GC closing the HTTP connection
    _client_ref = genai_client
    index = -1

    for event in response:
        event_type = getattr(event, "event_type", None) or getattr(event, "type", None)

    if event_type in ("step.delta", "content.delta"):
        delta = getattr(event, "delta", None)
        if delta:
            delta_type = getattr(delta, "type", None) or (
                delta.get("type") if isinstance(delta, dict) else None
            )
            if delta_type == "text":
                text_chunk = getattr(delta, "text", "") or (
                    delta.get("text", "") if isinstance(delta, dict) else ""
                )
                if text_chunk:
                    index += 1
                    yield LLMResultChunk(
                        model=model, prompt_messages=list(prompt_messages),
                        delta=LLMResultChunkDelta(
                            index=index,
                            message=AssistantPromptMessage(
                                content=[TextPromptMessageContent(data=text_chunk)]
                            ),
                        ),
                    )

    elif event_type in ("interaction.completed", "interaction.complete"):
        interaction_evt = getattr(event, "interaction", None) or event
        usage_obj = getattr(interaction_evt, "usage", None) if interaction_evt else None
        p_tok = getattr(usage_obj, "total_input_tokens", 0) or 0 if usage_obj else 0
        c_tok = (
            (getattr(usage_obj, "total_output_tokens", 0) or 0) +
            (getattr(usage_obj, "total_thought_tokens", 0) or 0)
        ) if usage_obj else 0
        usage = self._calc_response_usage(
            model=model, credentials=dict(credentials),
            prompt_tokens=p_tok, completion_tokens=c_tok,
        )
        yield LLMResultChunk(
            model=model, prompt_messages=list(prompt_messages),
            delta=LLMResultChunkDelta(
                index=index,
                message=AssistantPromptMessage(content=[]),
                finish_reason="stop", usage=usage,
            ),
        )

    elif event_type == "error":
        err = getattr(event, "error", None)
        msg = getattr(err, "message", None) or str(err) if err else "Unknown error"
        raise InvokeError(f"Interactions API error: {msg}")

Key gotchas discovered during development

1. input format: Must be Step dicts {"type": "user_input"/"model_output", "content": [...]}. Passing types.Content directly causes Unknown parameter 'parts' at 'input[0]'.
2. response_format: Must be a single dict, NOT a list. Passing both response_format (as list) + response_mime_type (as separate kwarg) causes responseFormat must be set when responseMimeType is set.
3. thinking_level: Must be lowercase "minimal"/"low"/"medium"/"high", NOT the uppercase enum values used by generateContent.
4. Streaming GC: Must keep _client_ref = genai_client in the stream handler to prevent garbage collection from closing the HTTP connection ([Errno 9] Bad file descriptor).

Dify version

1.14.2

Plugin version

gemini 0.9.1

Cloud or Self Hosted

Self Hosted (Docker)

Steps to reproduce

1. Create a Dify workflow with an LLM node
2. Select model gemini-3.5-flash
3. Enable Grounding (Google Search)
4. Enable Structured Output with any JSON schema
5. Run the workflow -> error: "Structured output (json_schema) cannot be used with: grounding"

✔️ Error log

PluginInvokeError: {"args":{"description":"[models] Error: Structured output (json_schema) cannot be used with: grounding, url_context"},"error_type":"InvokeError","message":"[models] Error: Structured output (json_schema) cannot be used with: grounding, url_context"}

[Harsh23Kashyap]

I'd like to work on this. I'll open a PR shortly.

[Harsh23Kashyap]

Opened PR #3429 for this.