bito-code-review[bot] commented on code in PR #37200:
URL: https://github.com/apache/superset/pull/37200#discussion_r2697310645
##########
superset/mcp_service/middleware.py:
##########
@@ -755,3 +755,150 @@ def _filter_response(self, response: Any, object_type:
str, user: Any) -> Any:
"Unknown response type for field filtering: %s", type(response)
)
return response
+
+
+class ResponseSizeGuardMiddleware(Middleware):
+ """
+ Middleware that prevents oversized responses from overwhelming LLM clients.
+
+ When a tool response exceeds the configured token limit, this middleware
+ intercepts it and returns a helpful error message with suggestions for
+ reducing the response size.
+
+ This is critical for protecting LLM clients like Claude Desktop which can
+ crash or become unresponsive when receiving extremely large responses.
+
+ Configuration via MCP_RESPONSE_SIZE_CONFIG in superset_config.py:
+ - enabled: Toggle the guard on/off (default: True)
+ - token_limit: Maximum estimated tokens per response (default: 50,000)
+ - warn_threshold_pct: Log warnings above this % of limit (default: 80%)
+ - excluded_tools: Tools to skip checking
+ """
+
+ def __init__(
+ self,
+ token_limit: int = 25000,
+ warn_threshold_pct: int = 80,
+ excluded_tools: list[str] | None = None,
+ ) -> None:
+ self.token_limit = token_limit
+ self.warn_threshold_pct = warn_threshold_pct
+ self.warn_threshold = int(token_limit * warn_threshold_pct / 100)
+ self.excluded_tools = set(excluded_tools or [])
+
+ async def on_call_tool(
+ self,
+ context: MiddlewareContext,
+ call_next: Callable[[MiddlewareContext], Awaitable[Any]],
+ ) -> Any:
+ """Check response size after tool execution."""
+ tool_name = getattr(context.message, "name", "unknown")
+
+ # Skip excluded tools
+ if tool_name in self.excluded_tools:
+ return await call_next(context)
+
+ # Execute the tool
+ response = await call_next(context)
+
+ # Estimate response token count
+ from superset.mcp_service.utils.token_utils import (
+ estimate_response_tokens,
+ format_size_limit_error,
+ )
+
+ estimated_tokens = estimate_response_tokens(response)
+
+ # Log warning if approaching limit
+ if estimated_tokens > self.warn_threshold:
+ logger.warning(
+ "Response size warning for %s: ~%d tokens (%.0f%% of %d
limit)",
+ tool_name,
+ estimated_tokens,
+ (estimated_tokens / self.token_limit) * 100,
+ self.token_limit,
+ )
+
+ # Block if over limit
+ if estimated_tokens > self.token_limit:
+ # Extract params for smart suggestions
+ params = getattr(context.message, "params", {}) or {}
+
+ # Log the blocked response
+ logger.error(
+ "Response blocked for %s: ~%d tokens exceeds limit of %d",
+ tool_name,
+ estimated_tokens,
+ self.token_limit,
+ )
+
+ # Log to event logger for monitoring
+ try:
+ user_id = get_user_id()
+ event_logger.log(
+ user_id=user_id,
+ action="mcp_response_size_exceeded",
+ curated_payload={
+ "tool": tool_name,
+ "estimated_tokens": estimated_tokens,
+ "token_limit": self.token_limit,
+ "params": params,
+ },
+ )
+ except Exception as log_error:
+ logger.warning("Failed to log size exceeded event: %s",
log_error)
+
+ # Generate helpful error message with suggestions
+ error_message = format_size_limit_error(
+ tool_name=tool_name,
+ params=params,
+ estimated_tokens=estimated_tokens,
+ token_limit=self.token_limit,
+ response=response,
+ )
+
+ raise ToolError(error_message)
+
+ return response
+
+
+def create_response_size_guard_middleware() -> ResponseSizeGuardMiddleware |
None:
+ """
+ Factory function to create ResponseSizeGuardMiddleware from config.
+
+ Reads configuration from Flask app's MCP_RESPONSE_SIZE_CONFIG.
+ Returns None if the guard is disabled.
+
+ Returns:
+ ResponseSizeGuardMiddleware instance or None if disabled
+ """
+ try:
+ from superset.mcp_service.flask_singleton import get_flask_app
+ from superset.mcp_service.mcp_config import MCP_RESPONSE_SIZE_CONFIG
+
+ flask_app = get_flask_app()
+
+ # Get config from Flask app, falling back to defaults
+ config = flask_app.config.get(
+ "MCP_RESPONSE_SIZE_CONFIG", MCP_RESPONSE_SIZE_CONFIG
+ )
+
+ if not config.get("enabled", True):
+ logger.info("Response size guard is disabled")
+ return None
+
+ middleware = ResponseSizeGuardMiddleware(
+ token_limit=config.get("token_limit", 25000),
+ warn_threshold_pct=config.get("warn_threshold_pct", 80),
+ excluded_tools=config.get("excluded_tools"),
+ )
+
+ logger.info(
+ "Created ResponseSizeGuardMiddleware with token_limit=%d",
+ middleware.token_limit,
+ )
+ return middleware
+
+ except Exception as e:
+ logger.error("Failed to create ResponseSizeGuardMiddleware: %s", e)
+ return None
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Blind exception catch should be specific</b></div>
<div id="fix">
Replace the broad `Exception` catch on line 902 with specific exception
types to avoid masking unexpected errors. Consider catching `(ImportError,
AttributeError, KeyError)` based on the operations performed.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
````suggestion
except (ImportError, AttributeError, KeyError) as e:
logger.error("Failed to create ResponseSizeGuardMiddleware: %s", e)
return None
````
</div>
</details>
</div>
<small><i>Code Review Run #9f50b8</i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/mcp_service/middleware.py:
##########
@@ -755,3 +755,150 @@ def _filter_response(self, response: Any, object_type:
str, user: Any) -> Any:
"Unknown response type for field filtering: %s", type(response)
)
return response
+
+
+class ResponseSizeGuardMiddleware(Middleware):
+ """
+ Middleware that prevents oversized responses from overwhelming LLM clients.
+
+ When a tool response exceeds the configured token limit, this middleware
+ intercepts it and returns a helpful error message with suggestions for
+ reducing the response size.
+
+ This is critical for protecting LLM clients like Claude Desktop which can
+ crash or become unresponsive when receiving extremely large responses.
+
+ Configuration via MCP_RESPONSE_SIZE_CONFIG in superset_config.py:
+ - enabled: Toggle the guard on/off (default: True)
+ - token_limit: Maximum estimated tokens per response (default: 50,000)
+ - warn_threshold_pct: Log warnings above this % of limit (default: 80%)
+ - excluded_tools: Tools to skip checking
+ """
+
+ def __init__(
+ self,
+ token_limit: int = 25000,
+ warn_threshold_pct: int = 80,
+ excluded_tools: list[str] | None = None,
+ ) -> None:
+ self.token_limit = token_limit
+ self.warn_threshold_pct = warn_threshold_pct
+ self.warn_threshold = int(token_limit * warn_threshold_pct / 100)
+ self.excluded_tools = set(excluded_tools or [])
+
+ async def on_call_tool(
+ self,
+ context: MiddlewareContext,
+ call_next: Callable[[MiddlewareContext], Awaitable[Any]],
+ ) -> Any:
+ """Check response size after tool execution."""
+ tool_name = getattr(context.message, "name", "unknown")
+
+ # Skip excluded tools
+ if tool_name in self.excluded_tools:
+ return await call_next(context)
+
+ # Execute the tool
+ response = await call_next(context)
+
+ # Estimate response token count
+ from superset.mcp_service.utils.token_utils import (
+ estimate_response_tokens,
+ format_size_limit_error,
+ )
+
+ estimated_tokens = estimate_response_tokens(response)
+
+ # Log warning if approaching limit
+ if estimated_tokens > self.warn_threshold:
+ logger.warning(
+ "Response size warning for %s: ~%d tokens (%.0f%% of %d
limit)",
+ tool_name,
+ estimated_tokens,
+ (estimated_tokens / self.token_limit) * 100,
+ self.token_limit,
+ )
+
+ # Block if over limit
+ if estimated_tokens > self.token_limit:
+ # Extract params for smart suggestions
+ params = getattr(context.message, "params", {}) or {}
+
+ # Log the blocked response
+ logger.error(
+ "Response blocked for %s: ~%d tokens exceeds limit of %d",
+ tool_name,
+ estimated_tokens,
+ self.token_limit,
+ )
+
+ # Log to event logger for monitoring
+ try:
+ user_id = get_user_id()
+ event_logger.log(
+ user_id=user_id,
+ action="mcp_response_size_exceeded",
+ curated_payload={
+ "tool": tool_name,
+ "estimated_tokens": estimated_tokens,
+ "token_limit": self.token_limit,
+ "params": params,
+ },
+ )
+ except Exception as log_error:
+ logger.warning("Failed to log size exceeded event: %s",
log_error)
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Blind exception catch should be specific</b></div>
<div id="fix">
Replace the broad `Exception` catch on line 848 with specific exception
types (e.g., `(ValueError, TypeError, AttributeError)`) to avoid masking
unexpected errors.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
````suggestion
# Log to event logger for monitoring
try:
user_id = get_user_id()
event_logger.log(
user_id=user_id,
action="mcp_response_size_exceeded",
curated_payload={
"tool": tool_name,
"estimated_tokens": estimated_tokens,
"token_limit": self.token_limit,
"params": params,
},
)
except (ValueError, TypeError, AttributeError) as log_error:
logger.warning("Failed to log size exceeded event: %s",
log_error)
````
</div>
</details>
</div>
<small><i>Code Review Run #9f50b8</i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/mcp_service/utils/token_utils.py:
##########
@@ -0,0 +1,402 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Token counting and response size utilities for MCP service.
+
+This module provides utilities to estimate token counts and generate smart
+suggestions when responses exceed configured limits. This prevents large
+responses from overwhelming LLM clients like Claude Desktop.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List
+
+logger = logging.getLogger(__name__)
+
+# Approximate characters per token for estimation
+# Claude tokenizer averages ~4 chars per token for English text
+# JSON tends to be more verbose, so we use a slightly lower ratio
+CHARS_PER_TOKEN = 3.5
+
+
+def estimate_token_count(text: str | bytes) -> int:
+ """
+ Estimate the token count for a given text.
+
+ Uses a character-based heuristic since we don't have direct access to
+ the actual tokenizer. This is conservative to avoid underestimating.
+
+ Args:
+ text: The text to estimate tokens for (string or bytes)
+
+ Returns:
+ Estimated number of tokens
+ """
+ if isinstance(text, bytes):
+ text = text.decode("utf-8", errors="replace")
+
+ # Simple heuristic: ~3.5 characters per token for JSON/code
+ return int(len(text) / CHARS_PER_TOKEN)
+
+
+def estimate_response_tokens(response: Any) -> int:
+ """
+ Estimate token count for an MCP tool response.
+
+ Handles various response types including Pydantic models, dicts, and
strings.
+
+ Args:
+ response: The response object to estimate
+
+ Returns:
+ Estimated number of tokens
+ """
+ try:
+ from superset.utils import json
+
+ # Convert response to JSON string for accurate estimation
+ if hasattr(response, "model_dump"):
+ # Pydantic model
+ response_str = json.dumps(response.model_dump())
+ elif isinstance(response, (dict, list)):
+ response_str = json.dumps(response)
+ elif isinstance(response, (str, bytes)):
+ response_str = response if isinstance(response, str) else
response.decode()
+ else:
+ response_str = str(response)
+
+ return estimate_token_count(response_str)
+ except Exception as e:
+ logger.warning("Failed to estimate response tokens: %s", e)
+ # Return a high estimate to be safe
+ return 100000
+
+
+def get_response_size_bytes(response: Any) -> int:
+ """
+ Get the size of a response in bytes.
+
+ Args:
+ response: The response object
+
+ Returns:
+ Size in bytes
+ """
+ try:
+ from superset.utils import json
+
+ if hasattr(response, "model_dump"):
+ response_str = json.dumps(response.model_dump())
+ elif isinstance(response, (dict, list)):
+ response_str = json.dumps(response)
+ elif isinstance(response, bytes):
+ return len(response)
+ elif isinstance(response, str):
+ return len(response.encode("utf-8"))
+ else:
+ response_str = str(response)
+
+ return len(response_str.encode("utf-8"))
+ except Exception as e:
+ logger.warning("Failed to get response size: %s", e)
+ return 0
+
+
+def extract_query_params(params: Dict[str, Any] | None) -> Dict[str, Any]:
+ """
+ Extract relevant query parameters from tool params for suggestions.
+
+ Args:
+ params: The tool parameters dict
+
+ Returns:
+ Extracted parameters relevant for size reduction suggestions
+ """
+ if not params:
+ return {}
+
+ # Handle nested request object (common pattern in MCP tools)
+ if "request" in params and isinstance(params["request"], dict):
+ params = params["request"]
+
+ extracted = {}
+
+ # Pagination params
+ if "page_size" in params:
+ extracted["page_size"] = params["page_size"]
+ if "limit" in params:
+ extracted["limit"] = params["limit"]
+
+ # Column selection
+ if "select_columns" in params:
+ extracted["select_columns"] = params["select_columns"]
+ if "columns" in params:
+ extracted["columns"] = params["columns"]
+
+ # Filters
+ if "filters" in params:
+ extracted["filters"] = params["filters"]
+
+ # Search
+ if "search" in params:
+ extracted["search"] = params["search"]
+
+ return extracted
+
+
+def generate_size_reduction_suggestions(
+ tool_name: str,
+ params: Dict[str, Any] | None,
+ estimated_tokens: int,
+ token_limit: int,
+ response: Any = None,
+) -> List[str]:
+ """
+ Generate smart suggestions for reducing response size.
+
+ Analyzes the tool and parameters to provide actionable recommendations.
+
+ Args:
+ tool_name: Name of the MCP tool
+ params: The tool parameters
+ estimated_tokens: Estimated token count of the response
+ token_limit: Configured token limit
+ response: Optional response object for additional analysis
+
+ Returns:
+ List of suggestion strings
+ """
+ suggestions = []
+ query_params = extract_query_params(params)
+ reduction_needed = estimated_tokens - token_limit
+ reduction_pct = int((reduction_needed / estimated_tokens) * 100)
+
+ # Suggestion 1: Reduce page_size or limit
+ current_page_size = query_params.get("page_size") or
query_params.get("limit")
+ if current_page_size:
+ # Calculate suggested new limit based on reduction needed
+ suggested_limit = max(
+ 1, int(current_page_size * (token_limit / estimated_tokens))
+ )
+ suggestions.append(
+ f"Reduce page_size/limit from {current_page_size} to
{suggested_limit} "
+ f"(need ~{reduction_pct}% reduction)"
+ )
+ else:
+ # No limit specified - suggest adding one
+ if "list_" in tool_name or tool_name.startswith("get_chart_data"):
+ suggestions.append(
+ f"Add a 'limit' or 'page_size' parameter (suggested: 10-25
items) "
+ f"to reduce response size by ~{reduction_pct}%"
+ )
+
+ # Suggestion 2: Use select_columns to reduce fields
+ current_columns = query_params.get("select_columns") or
query_params.get("columns")
+ if current_columns and len(current_columns) > 5:
+ suggestions.append(
+ f"Reduce select_columns from {len(current_columns)} columns to
only "
+ f"essential fields (currently: {current_columns[:3]}...)"
+ )
+ elif not current_columns and "list_" in tool_name:
+ # Analyze response to suggest specific columns to exclude
+ large_fields = _identify_large_fields(response)
+ if large_fields:
+ suggestions.append(
+ f"Use 'select_columns' to exclude large fields:
{large_fields[:3]}. "
+ f"Only request the columns you need."
+ )
+ else:
+ suggestions.append(
+ "Use 'select_columns' to request only the specific columns you
need "
+ "instead of fetching all fields"
+ )
+
+ # Suggestion 3: Add filters to reduce result set
+ current_filters = query_params.get("filters")
+ if not current_filters and "list_" in tool_name:
+ suggestions.append(
+ "Add filters to narrow down results (e.g., filter by owner, "
+ "date range, or specific attributes)"
+ )
+
+ # Suggestion 4: Tool-specific suggestions
+ tool_suggestions = _get_tool_specific_suggestions(tool_name, query_params,
response)
+ suggestions.extend(tool_suggestions)
+
+ # Suggestion 5: Use search instead of listing all
+ if "list_" in tool_name and not query_params.get("search"):
+ suggestions.append(
+ "Use the 'search' parameter to find specific items instead of "
+ "listing all and filtering client-side"
+ )
+
+ return suggestions
+
+
+def _identify_large_fields(response: Any) -> List[str]:
+ """
+ Identify fields that contribute most to response size.
+
+ Args:
+ response: The response object to analyze
+
+ Returns:
+ List of field names that are particularly large
+ """
+ large_fields: List[str] = []
+
+ try:
+ from superset.utils import json
+
+ if hasattr(response, "model_dump"):
+ data = response.model_dump()
+ elif isinstance(response, dict):
+ data = response
+ else:
+ return large_fields
+
+ # Check for list responses (e.g., charts, dashboards)
+ items_key = None
+ for key in ["charts", "dashboards", "datasets", "data", "items",
"results"]:
+ if key in data and isinstance(data[key], list) and data[key]:
+ items_key = key
+ break
+
+ if items_key and data[items_key]:
+ first_item = data[items_key][0]
+ if isinstance(first_item, dict):
+ # Analyze field sizes in first item
+ field_sizes = {}
+ for field, value in first_item.items():
+ if value is not None:
+ field_sizes[field] = len(json.dumps(value))
+
+ # Sort by size and identify large fields (>500 chars)
+ sorted_fields = sorted(
+ field_sizes.items(), key=lambda x: x[1], reverse=True
+ )
+ large_fields = [
+ f
+ for f, size in sorted_fields
+ if size > 500 and f not in ("id", "uuid", "name",
"slice_name")
+ ]
+
+ except Exception as e:
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Blind exception catch should be specific</b></div>
<div id="fix">
Catch specific exception types instead of bare `Exception`. Multiple similar
issues exist (lines 86, 117, 301).
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
````suggestion
except (ValueError, TypeError, AttributeError) as e:
````
</div>
</details>
</div>
<small><i>Code Review Run #9f50b8</i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/mcp_service/utils/token_utils.py:
##########
@@ -0,0 +1,402 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Token counting and response size utilities for MCP service.
+
+This module provides utilities to estimate token counts and generate smart
+suggestions when responses exceed configured limits. This prevents large
+responses from overwhelming LLM clients like Claude Desktop.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List
+
+logger = logging.getLogger(__name__)
+
+# Approximate characters per token for estimation
+# Claude tokenizer averages ~4 chars per token for English text
+# JSON tends to be more verbose, so we use a slightly lower ratio
+CHARS_PER_TOKEN = 3.5
+
+
+def estimate_token_count(text: str | bytes) -> int:
+ """
+ Estimate the token count for a given text.
+
+ Uses a character-based heuristic since we don't have direct access to
+ the actual tokenizer. This is conservative to avoid underestimating.
+
+ Args:
+ text: The text to estimate tokens for (string or bytes)
+
+ Returns:
+ Estimated number of tokens
+ """
+ if isinstance(text, bytes):
+ text = text.decode("utf-8", errors="replace")
+
+ # Simple heuristic: ~3.5 characters per token for JSON/code
+ return int(len(text) / CHARS_PER_TOKEN)
+
+
+def estimate_response_tokens(response: Any) -> int:
+ """
+ Estimate token count for an MCP tool response.
+
+ Handles various response types including Pydantic models, dicts, and
strings.
+
+ Args:
+ response: The response object to estimate
+
+ Returns:
+ Estimated number of tokens
+ """
+ try:
+ from superset.utils import json
+
+ # Convert response to JSON string for accurate estimation
+ if hasattr(response, "model_dump"):
+ # Pydantic model
+ response_str = json.dumps(response.model_dump())
+ elif isinstance(response, (dict, list)):
+ response_str = json.dumps(response)
+ elif isinstance(response, (str, bytes)):
+ response_str = response if isinstance(response, str) else
response.decode()
+ else:
+ response_str = str(response)
+
+ return estimate_token_count(response_str)
+ except Exception as e:
+ logger.warning("Failed to estimate response tokens: %s", e)
+ # Return a high estimate to be safe
+ return 100000
+
+
+def get_response_size_bytes(response: Any) -> int:
+ """
+ Get the size of a response in bytes.
+
+ Args:
+ response: The response object
+
+ Returns:
+ Size in bytes
+ """
+ try:
+ from superset.utils import json
+
+ if hasattr(response, "model_dump"):
+ response_str = json.dumps(response.model_dump())
+ elif isinstance(response, (dict, list)):
+ response_str = json.dumps(response)
+ elif isinstance(response, bytes):
+ return len(response)
+ elif isinstance(response, str):
+ return len(response.encode("utf-8"))
+ else:
+ response_str = str(response)
+
+ return len(response_str.encode("utf-8"))
+ except Exception as e:
+ logger.warning("Failed to get response size: %s", e)
+ return 0
+
+
+def extract_query_params(params: Dict[str, Any] | None) -> Dict[str, Any]:
+ """
+ Extract relevant query parameters from tool params for suggestions.
+
+ Args:
+ params: The tool parameters dict
+
+ Returns:
+ Extracted parameters relevant for size reduction suggestions
+ """
+ if not params:
+ return {}
+
+ # Handle nested request object (common pattern in MCP tools)
+ if "request" in params and isinstance(params["request"], dict):
+ params = params["request"]
+
+ extracted = {}
+
+ # Pagination params
+ if "page_size" in params:
+ extracted["page_size"] = params["page_size"]
+ if "limit" in params:
+ extracted["limit"] = params["limit"]
+
+ # Column selection
+ if "select_columns" in params:
+ extracted["select_columns"] = params["select_columns"]
+ if "columns" in params:
+ extracted["columns"] = params["columns"]
+
+ # Filters
+ if "filters" in params:
+ extracted["filters"] = params["filters"]
+
+ # Search
+ if "search" in params:
+ extracted["search"] = params["search"]
+
+ return extracted
+
+
+def generate_size_reduction_suggestions(
+ tool_name: str,
+ params: Dict[str, Any] | None,
+ estimated_tokens: int,
+ token_limit: int,
+ response: Any = None,
+) -> List[str]:
+ """
+ Generate smart suggestions for reducing response size.
+
+ Analyzes the tool and parameters to provide actionable recommendations.
+
+ Args:
+ tool_name: Name of the MCP tool
+ params: The tool parameters
+ estimated_tokens: Estimated token count of the response
+ token_limit: Configured token limit
+ response: Optional response object for additional analysis
+
+ Returns:
+ List of suggestion strings
+ """
+ suggestions = []
+ query_params = extract_query_params(params)
+ reduction_needed = estimated_tokens - token_limit
+ reduction_pct = int((reduction_needed / estimated_tokens) * 100)
+
+ # Suggestion 1: Reduce page_size or limit
+ current_page_size = query_params.get("page_size") or
query_params.get("limit")
+ if current_page_size:
+ # Calculate suggested new limit based on reduction needed
+ suggested_limit = max(
+ 1, int(current_page_size * (token_limit / estimated_tokens))
+ )
+ suggestions.append(
+ f"Reduce page_size/limit from {current_page_size} to
{suggested_limit} "
+ f"(need ~{reduction_pct}% reduction)"
+ )
+ else:
+ # No limit specified - suggest adding one
+ if "list_" in tool_name or tool_name.startswith("get_chart_data"):
+ suggestions.append(
+ f"Add a 'limit' or 'page_size' parameter (suggested: 10-25
items) "
+ f"to reduce response size by ~{reduction_pct}%"
+ )
+
+ # Suggestion 2: Use select_columns to reduce fields
+ current_columns = query_params.get("select_columns") or
query_params.get("columns")
+ if current_columns and len(current_columns) > 5:
+ suggestions.append(
+ f"Reduce select_columns from {len(current_columns)} columns to
only "
+ f"essential fields (currently: {current_columns[:3]}...)"
+ )
+ elif not current_columns and "list_" in tool_name:
+ # Analyze response to suggest specific columns to exclude
+ large_fields = _identify_large_fields(response)
+ if large_fields:
+ suggestions.append(
+ f"Use 'select_columns' to exclude large fields:
{large_fields[:3]}. "
+ f"Only request the columns you need."
+ )
+ else:
+ suggestions.append(
+ "Use 'select_columns' to request only the specific columns you
need "
+ "instead of fetching all fields"
+ )
+
+ # Suggestion 3: Add filters to reduce result set
+ current_filters = query_params.get("filters")
+ if not current_filters and "list_" in tool_name:
+ suggestions.append(
+ "Add filters to narrow down results (e.g., filter by owner, "
+ "date range, or specific attributes)"
+ )
+
+ # Suggestion 4: Tool-specific suggestions
+ tool_suggestions = _get_tool_specific_suggestions(tool_name, query_params,
response)
+ suggestions.extend(tool_suggestions)
+
+ # Suggestion 5: Use search instead of listing all
+ if "list_" in tool_name and not query_params.get("search"):
+ suggestions.append(
+ "Use the 'search' parameter to find specific items instead of "
+ "listing all and filtering client-side"
+ )
+
+ return suggestions
+
+
+def _identify_large_fields(response: Any) -> List[str]:
+ """
+ Identify fields that contribute most to response size.
+
+ Args:
+ response: The response object to analyze
+
+ Returns:
+ List of field names that are particularly large
+ """
+ large_fields: List[str] = []
+
+ try:
+ from superset.utils import json
+
+ if hasattr(response, "model_dump"):
+ data = response.model_dump()
+ elif isinstance(response, dict):
+ data = response
+ else:
+ return large_fields
+
+ # Check for list responses (e.g., charts, dashboards)
+ items_key = None
+ for key in ["charts", "dashboards", "datasets", "data", "items",
"results"]:
+ if key in data and isinstance(data[key], list) and data[key]:
+ items_key = key
+ break
+
+ if items_key and data[items_key]:
+ first_item = data[items_key][0]
+ if isinstance(first_item, dict):
+ # Analyze field sizes in first item
+ field_sizes = {}
+ for field, value in first_item.items():
+ if value is not None:
+ field_sizes[field] = len(json.dumps(value))
+
+ # Sort by size and identify large fields (>500 chars)
+ sorted_fields = sorted(
+ field_sizes.items(), key=lambda x: x[1], reverse=True
+ )
+ large_fields = [
+ f
+ for f, size in sorted_fields
+ if size > 500 and f not in ("id", "uuid", "name",
"slice_name")
+ ]
+
+ except Exception as e:
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Blind exception catch should be specific</b></div>
<div id="fix">
Catch specific exception types instead of bare `Exception`.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
````suggestion
except (ValueError, TypeError, KeyError) as e:
````
</div>
</details>
</div>
<small><i>Code Review Run #9f50b8</i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
