Couchbase MCP Server

Instructions

Args:
    limit: Number of queries to return (default: 10)

Returns:
    List of queries with their average result size in bytes, KB, and MB

Implementation Reference

• src/tools/query.py:198-234 (handler)

  The core handler function that implements the tool logic by executing a SQL++ query on system:completed_requests to identify queries with the largest average response sizes, formatting sizes in bytes, KB, and MB, and handling empty results gracefully.

  def get_queries_with_largest_response_sizes(
      ctx: Context, limit: int = 10
  ) -> list[dict[str, Any]]:
      """Get queries with the largest response sizes from the system:completed_requests catalog.
  
      Args:
          limit: Number of queries to return (default: 10)
  
      Returns:
          List of queries with their average result size in bytes, KB, and MB
      """
      query = """
      SELECT statement,
          avgResultSize AS avgResultSizeBytes,
          (avgResultSize / 1000) AS avgResultSizeKB,
          (avgResultSize / 1000000) AS avgResultSizeMB,
          COUNT(1) AS queries
      FROM system:completed_requests
      WHERE UPPER(statement) NOT LIKE 'INFER %'
          AND UPPER(statement) NOT LIKE 'CREATE INDEX%'
          AND UPPER(statement) NOT LIKE 'CREATE PRIMARY INDEX%'
          AND UPPER(statement) NOT LIKE '% SYSTEM:%'
      GROUP BY statement
      LETTING avgResultSize = AVG(resultSize)
      ORDER BY avgResultSize DESC
      LIMIT $limit
      """
  
      return _run_query_tool_with_empty_message(
          ctx,
          query,
          limit=limit,
          empty_message=(
              "No completed queries were available to calculate response sizes."
          ),
      )

• src/mcp_server.py:175-177 (registration)

  The tool is registered to the FastMCP server instance by iterating over ALL_TOOLS (which includes this tool) and calling add_tool().

  # Register all tools
  for tool in ALL_TOOLS:
      mcp.add_tool(tool)

• src/tools/__init__.py:58-61 (registration)

  The tool handler is imported from query.py and explicitly included in the ALL_TOOLS list, which is used by mcp_server.py for registration.

  get_queries_not_using_covering_index,
  get_queries_using_primary_index,
  get_queries_with_large_result_count,
  get_queries_with_largest_response_sizes,

• src/tools/query.py:108-127 (helper)

  Supporting helper utility invoked by the handler to run the cluster query and provide a standardized empty results response.

  def _run_query_tool_with_empty_message(
      ctx: Context,
      query: str,
      *,
      limit: int,
      empty_message: str,
      extra_payload: dict[str, Any] | None = None,
      **query_kwargs: Any,
  ) -> list[dict[str, Any]]:
      """Execute a cluster query with a consistent empty-result response."""
      results = run_cluster_query(ctx, query, limit=limit, **query_kwargs)
  
      if results:
          return results
  
      payload: dict[str, Any] = {"message": empty_message, "results": []}
      if extra_payload:
          payload.update(extra_payload)
      return [payload]